RocketMQ 4.3 版本之后提供了对分布式事务消息的支持，它采用了一种类似于两阶段提交（2PC）的机制，但又有所不同，可以实现最终一致性的分布式事务。RocketMQ 的事务消息主要用于解决生产者发送消息和本地事务的原子性问题。
应用场景：
典型的场景是电商中的下单流程：
- 用户下单，订单服务需要创建订单。
- 同时，需要发送一个消息到库存服务，通知其扣减库存。
如果订单创建成功，但消息发送失败，或者消息发送成功，但订单创建失败，都会导致数据不一致。RocketMQ 的事务消息可以解决这个问题。
RocketMQ 事务消息原理：
-
发送半消息（Half Message）：
- 生产者首先发送一个半消息（Half Message）到 RocketMQ Broker。
- 半消息对消费者不可见，它会被存储在一个特殊的内部 topic（
RMQ_SYS_TRANS_HALF_TOPIC）中。 - Broker 会响应生产者，确认半消息已接收。
-
执行本地事务：
- 生产者在收到 Broker 对半消息的确认后，执行本地事务（例如，创建订单）。
- 本地事务的执行结果可能是提交、回滚或未知。
-
提交/回滚半消息：
- 根据本地事务的执行结果，生产者向 Broker 发送提交（Commit）或回滚（Rollback）请求。
- 提交： Broker 将半消息标记为可投递，并将其从
RMQ_SYS_TRANS_HALF_TOPIC转移到真正的目标 topic。消费者现在可以看到并消费这条消息。 - 回滚： Broker 删除半消息。
- 未知： 生产者不发送任何请求，Broker 会定期回查生产者（见下文）。
- 提交： Broker 将半消息标记为可投递，并将其从
- 根据本地事务的执行结果，生产者向 Broker 发送提交（Commit）或回滚（Rollback）请求。
-
事务回查 (Transaction Check)：
- 如果 Broker 长时间没有收到生产者对半消息的提交或回滚请求（可能是生产者宕机、网络问题等原因），Broker 会主动向生产者发送事务回查请求。
- 生产者收到回查请求后，需要检查本地事务的状态（通常通过查询数据库），并再次向 Broker 发送提交或回滚请求。
- RocketMQ默认回查15次, 可以通过
transactionCheckMax参数配置.
-
消费者消费消息：
- 消费者从目标 topic 消费消息，执行相应的业务逻辑（例如，扣减库存）。
- 消费者的消费逻辑需要保证幂等性，因为在某些情况下（例如，消费者处理消息时宕机），消息可能会被重复消费。
关键概念：
- 半消息 (Half Message): 一种特殊的消息，对消费者不可见，用于在生产者和 Broker 之间进行事务协调。
- 事务回查 (Transaction Check): Broker 主动向生产者查询半消息状态的机制，用于处理生产者宕机或网络问题导致的消息状态不一致。
- 事务消息生产者 (TransactionMQProducer): RocketMQ 提供的特殊生产者，用于发送事务消息。需要配置
TransactionListener。 TransactionListener: 生产者端的监听器接口, 包含两个方法：executeLocalTransaction: 执行本地事务，并返回本地事务的状态（LocalTransactionState.COMMIT_MESSAGE、LocalTransactionState.ROLLBACK_MESSAGE或LocalTransactionState.UNKNOW）。checkLocalTransaction: 用于事务回查，检查本地事务的状态，并返回本地事务的状态。
代码示例 (Java):
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;import java.util.concurrent.*;public class TransactionProducer {public static void main(String[] args) throws Exception {// 1. 创建事务消息生产者TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");producer.setNamesrvAddr("localhost:9876"); // 设置 NameServer 地址// 2. 设置事务监听器ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {@Overridepublic Thread newThread(Runnable r) {Thread thread = new Thread(r);thread.setName("client-transaction-msg-check-thread");return thread;}});producer.setExecutorService(executorService);producer.setTransactionListener(new TransactionListener() {// 执行本地事务@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {System.out.println("Executing local transaction...");try {// 执行本地事务 (例如，创建订单)// ...boolean success = true; // 假设本地事务执行成功if (success) {System.out.println("Local transaction succeeded. Commit message.");return LocalTransactionState.COMMIT_MESSAGE; // 提交消息} else {System.out.println("Local transaction failed. Rollback message.");return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息}} catch (Exception e) {System.out.println("Local transaction exception. Unknown state.");return LocalTransactionState.UNKNOW; // 未知状态 (Broker 会回查)}}// 事务回查@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {System.out.println("Checking local transaction state...");// 检查本地事务的状态 (例如，查询订单数据库)// ...boolean exists = true; // 假设订单已创建if (exists) {System.out.println("Local transaction exists. Commit message.");return LocalTransactionState.COMMIT_MESSAGE;} else {System.out.println("Local transaction does not exist. Rollback message.");return LocalTransactionState.ROLLBACK_MESSAGE;}}});// 3. 启动生产者producer.start();// 4. 发送事务消息String[] tags = new String[]{"TagA", "TagB", "TagC"};for (int i = 0; i < 3; i++) {Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));// 发送事务消息SendResult sendResult = producer.sendMessageInTransaction(msg, null);System.out.printf("%s%n", sendResult);Thread.sleep(10);}// 5. 关闭生产者 (实际应用中，生产者通常不需要立即关闭)// producer.shutdown();}
}
总结：
RocketMQ 的事务消息通过半消息、本地事务执行、提交/回滚/回查机制，以及消费者端的幂等性保证，实现了最终一致性的分布式事务。它适用于需要保证消息发送和本地事务原子性的场景，例如电商中的下单、支付等。 相比于传统的 2PC，RocketMQ 的事务消息性能更高，对业务代码的侵入性更小。
需要注意的点：
- RocketMQ 的事务消息只保证最终一致性，不保证强一致性（即不保证隔离性）。
- 消费者的消费逻辑需要保证幂等性。
- 事务回查可能会导致消息重复发送，因此消费者的业务逻辑需要能够处理重复消息。
- 事务消息不支持延时消息和批量消息.
- 为了保证消息不丢失, Half 消息队列和 OP 消息队列也需要配置成主从同步复制模式.
