RocketMQ是一款开源的分布式消息中间件，支持高可靠、高吞吐量的消息传递。
事务消息是RocketMQ的一个重要特性，它能够确保在分布式环境下的消息可靠性和一致性。
在本文中，我将详细解释RocketMQ事务消息的工作原理和使用方法@mikechen
RocketMQ事务消息定义
事务消息是指在消息发送方发送消息时，会将消息发送到消息中间件，然后消息中间件将会将消息的确认消息回调给消息发送方。
发送方根据接收到的确认消息来决定是否提交事务或者回滚事务。事务消息通常在分布式事务场景下使用，确保在多个参与方之间的操作具有一致性。
RocketMQ事务消息原理
下面是RocketMQ事务消息的基本工作流程：
- 应用程序发送事务消息到RocketMQ。
- RocketMQ接收到事务消息后，会返回一个预提交（half commit）的状态给应用程序。
- 应用程序执行本地事务。
- 本地事务执行成功后，应用程序向RocketMQ发送确认消息，确认消息将提交到消息中间件。
- RocketMQ收到确认消息后，将该消息标记为已提交。
- 如果本地事务执行失败或超时，应用程序向RocketMQ发送回滚消息。
- RocketMQ收到回滚消息后，将消息标记为已回滚。
通过这个工作流程，RocketMQ可以保证事务消息的最终一致性。
事务消息的使用方法
使用RocketMQ的事务消息，需要按照以下步骤进行设置：
1.创建事务监听器（TransactionListener）
应用程序需要实现一个事务监听器，该监听器负责执行本地事务、消息的确认和回滚操作。
监听器包含三个方法：
- executeLocalTransaction：执行本地事务。在该方法中，应用程序执行实际的业务操作，可以是数据库的写入或其他操作。
- checkLocalTransaction：检查本地事务状态。该方法用于检查本地事务的状态，返回COMMIT、ROLLBACK或UNKNOWN。
- onMessage：处理回查逻辑。当消息中间件发起事务状态的回查时，会调用该方法。
如下所示：
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.common.message.Message;
public class MyTransactionListener implements TransactionListener {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 执行本地事务，例如数据库操作
// 返回COMMIT_MESSAGE、ROLLBACK_MESSAGE或UNKNOW
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 检查本地事务状态，根据消息的状态返回COMMIT_MESSAGE、ROLLBACK_MESSAGE或UNKNOW
}
@Override
public void onMessage(MessageExt msg) {
// 处理事务消息的回查逻辑
}
}
2.初始化事务生产者（TransactionMQProducer）
创建一个事务生产者实例，并设置事务监听器，同时可以设置其他参数，如NameServer地址、消息发送超时时间等。
如下所示：
import org.apache.rocketmq.client.producer.TransactionMQProducer;
public class TransactionProducer {
public static void main(String[] args) throws Exception {
// 创建事务监听器
TransactionListener transactionListener = new MyTransactionListener();
// 创建事务生产者实例
TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");
producer.setTransactionListener(transactionListener);
// 设置NameServer地址
producer.setNamesrvAddr("localhost:9876");
// 启动事务生产者
producer.start();
// 发送事务消息
Message msg = new Message("topic", "tag", "transaction message".getBytes());
producer.sendMessageInTransaction(msg, null);
// 关闭事务生产者
producer.shutdown();
}
}
3.发送事务消息
使用事务生产者发送事务消息，在发送消息时，需要指定目标主题（Topic）、消息标签（Tag）和消息内容。
如下所示：
Message msg = new Message("transaction_topic", "transaction_tag", "Hello, RocketMQ!".getBytes());
producer.sendMessageInTransaction(msg, null);
4.处理事务状态回查
在应用程序启动时，需要启动事务消息监听器，以便接收消息中间件的回调。
当RocketMQ无法确定事务的最终状态时，会触发事务状态回查。在事务监听器的checkLocalTransaction方法中，您可以根据消息的状态进行判断，并返回相应的LocalTransactionState状态。
// 检查本地事务状态
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
String transactionId = msg.getTransactionId();
Integer status = localTrans.get(transactionId);
if (status != null) {
switch (status) {
case 1:
return LocalTransactionState.COMMIT_MESSAGE;
case 2:
return LocalTransactionState.ROLLBACK_MESSAGE;
default:
return LocalTransactionState.UNKNOW;
}
}
return LocalTransactionState.UNKNOW;
}
通过以上步骤，您可以使用RocketMQ的事务消息来保证分布式环境下的消息可靠性和一致性。
5.运行启动
public static void main(String[] args) throws MQClientException, InterruptedException {
// 创建事务监听器
TransactionListener transactionListener = new TransactionListenerImpl();
// 创建事务生产者实例
TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");
producer.setNamesrvAddr("localhost:9876");
producer.setTransactionListener(transactionListener);
// 启动事务生产者
producer.start();
// 发送事务消息
try {
Message msg = new Message("transaction_topic", "transaction_tag", "Transaction Message Example".getBytes(RemotingHelper.DEFAULT_CHARSET));
TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);
System.out.println("发送事务消息，事务状态：" sendResult.getLocalTransactionState());
} catch (MQClientException | UnsupportedEncodingException | RemotingException e) {
e.printStackTrace();
}
// 关闭事务生产者
// producer.shutdown();
}
以上就是RocketMQ事务消息详解，更多RocketMQ内容请查看：RocketMQ最全详解(万字图文教程)
mikechen
mikechen睿哥，10年+大厂架构经验，资深技术专家，就职于阿里巴巴、淘宝、百度等一线互联网大厂。
关注「mikechen」公众号，获知最新一线技术干货！
