谈谈 RocketMQ 的事务消息？

结论：事务消息解决的是"本地数据库事务"和"发送消息"两个操作的原子性。做法是先把消息以"半消息（Half Message）"写入 Broker 但对消费者不可见，本地事务执行成功后提交（Commit）让消息可见，失败则回滚（Rollback）；如果 Broker 迟迟收不到二次确认，就反向回查生产者的本地事务状态。它保证的是"本地事务与消息发送同生共死"，不保证消费端只消费一次。

一、执行时序

Producer                     Broker                      Consumer
   │  1. 发送半消息              │
   ├──────────────────────────>│ 写入 RMQ_SYS_TRANS_HALF_TOPIC
   │  2. 半消息写入成功           │ (对 Consumer 不可见)
   │<──────────────────────────┤
   │  3. 执行本地事务(DB)         │
   │     成功 → Commit           │
   ├──────────────────────────>│ 4. 消息转正写入真实 Topic ──> 可见
   │     失败 → Rollback         │
   ├──────────────────────────>│ 4'. 丢弃半消息
   │                            │
   │  5. 若没收到 Commit/Rollback，定时回查(默认 60s 一次，最多 15 次)
   │<──────────────────────────┤
   │  6. 返回本地事务状态(COMMIT/ROLLBACK/UNKNOW)
   ├──────────────────────────>│

关键点：Commit 之后消息才对消费者可见，所以"消息先到、事务后提交"的脏读问题被消除了；回查机制保证生产者宕机后不会留下永远悬空的半消息。

二、三种事务状态

回查次数用尽仍是 UNKNOW，Broker 会丢弃该半消息并打印日志，这时必须靠告警人工介入，否则就是"扣了钱没发货"。

三、代码实现

// 1. 事务监听器：执行本地事务 + 回查
public class OrderTransactionListener implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            orderService.createOrder((OrderEvent) arg);   // 本地事务
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (BizException e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.UNKNOW;          // 交给回查
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String orderId = msg.getKeys();
        Order o = orderMapper.selectByOrderId(orderId);   // 查本地状态
        if (o == null) return LocalTransactionState.ROLLBACK_MESSAGE;
        return o.isPaid() ? LocalTransactionState.COMMIT_MESSAGE
                          : LocalTransactionState.UNKNOW;
    }
}

// 2. 生产者装配
TransactionMQProducer producer = new TransactionMQProducer("order_tx_producer");
producer.setNamesrvAddr("10.0.0.11:9876");
producer.setTransactionListener(new OrderTransactionListener());
producer.setExecutorService(new ThreadPoolExecutor(4, 8, 60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(2000)));
producer.start();

Message msg = new Message("ORDER_TOPIC", "CREATED", orderId, body);
producer.sendMessageInTransaction(msg, orderEvent);

# 3. Broker 侧回查参数
transactionCheckMax=15
transactionCheckInterval=60000
transactionTimeOut=6000

四、回查怎么查才可靠

回查不能依赖内存状态，必须查持久化数据。常见两种设计：

推荐第二种：事务日志表用 orderId 做唯一键，executeLocalTransaction 与本地业务在同一个数据库事务里写这条日志，回查只看日志状态，判断逻辑就不依赖业务表结构。

五、常见追问

• 事务消息能保证消费端 exactly-once 吗？ 不能。它只保证"本地事务成功则消息一定可投递"，消费端仍然可能重复消费，必须靠唯一键去重。
• 为什么不能先发普通消息再执行本地事务？ 消息可能先被消费，此时本地事务还没提交，消费者查不到数据；也不满足"事务失败但消息已发出"的原子性。
• 半消息存在哪里，会不会被消费？ 存在 RMQ_SYS_TRANS_HALF_TOPIC 这个内部 Topic，消费者订阅不到；Commit 时通过写入一条 op 消息把原消息恢复到真实 Topic。
• 回查一直 UNKNOW 会怎样？ 超过 transactionCheckMax 次后被丢弃，消息永久不可见，属于数据不一致，必须监控半消息堆积并告警。
• 和 Kafka 的事务有什么区别？ Kafka 事务保证的是"一批消息 + 位点提交"在流内的原子性，不能和外部数据库事务绑定；RocketMQ 的事务消息是面向"本地库 + 消息"的双写场景。

小结

• 事务消息 = 半消息 + 本地事务 + Commit/Rollback + 定时回查。
• 它解决生产端"发消息与本地事务"的原子性，不解决消费端重复与下游一致性。
• 回查必须查持久化状态，事务日志表是最稳的实现方式。
• 回查次数耗尽意味着数据不一致，必须配监控而不能只靠日志。