3. RocketMQ事务消息：从理论到实战

RocketMQ的事务消息，说白了就是解决「本地数据库操作」和「发MQ消息」这两件事的原子性问题。我刚开始接触时也觉得挺绕的，但搞明白Half Message机制后，其实思路很清晰。

3.1 为什么需要事务消息？

先看一个经典场景：用户下单，扣库存，发消息通知物流系统。

伪代码大概是这样的：

// 伪代码：下单 + 发消息
beginTransaction();
    insertOrder(order);
    sendMessage("物流系统", orderId);
commitTransaction();

问题来了——如果消息发送成功，但数据库事务回滚了，物流系统就收到了一个假订单。反过来，数据库提交了，消息发送失败，物流系统永远不知道有新订单。

这就是分布式事务的典型困境。RocketMQ的事务消息，就是专门为这类场景设计的。

3.2 Half Message机制

Half Message，翻译过来就是「半消息」。为什么叫半消息？因为它对消费者来说是不可见的。

流程是这样的：

1. 生产者发送Half Message到Broker——此时消息状态是「预备态」，消费者无法消费。
2. Broker存储Half Message成功，返回确认给生产者。
3. 生产者执行本地事务（比如插入订单数据）。
4. 生产者根据本地事务结果，向Broker发送Commit或Rollback指令。
5. 如果收到Commit，Broker将Half Message转为正式消息，消费者可见。
6. 如果收到Rollback，Broker删除Half Message，相当于没发过。

嗯，这里要注意：如果第4步生产者挂了，或者网络超时了，Broker会怎么办？

这就是事务回查的用武之地。

3.3 本地事务执行与回查

我参与过一个电商项目，订单服务每天处理几十万笔交易。刚开始我们用的是普通消息加定时任务补偿，后来发现补偿逻辑越来越复杂，干脆切到了RocketMQ事务消息。

事务回查，说白了就是Broker发现某个Half Message长时间没有收到Commit/Rollback，主动问生产者：「嘿，你那个本地事务到底成没成？」

生产者需要实现一个回调接口：

// 事务监听器接口
public interface TransactionListener {
    // 执行本地事务
    LocalTransactionState executeLocalTransaction(Message msg, Object arg);
    
    // 回查本地事务状态
    LocalTransactionState checkLocalTransaction(MessageExt msg);
}

回查返回的状态有三种：

3.4 整体流程可视化

下面这张图把整个事务消息的流转画清楚了：

3.5 代码实战：事务消息生产者

直接上代码，这是我项目中实际用过的模板：

// 创建事务生产者
TransactionMQProducer producer = new TransactionMQProducer("order_group");
producer.setNamesrvAddr("localhost:9876");

// 设置事务监听器
producer.setTransactionListener(new TransactionListener() {
    
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 执行本地事务——插入订单
            Order order = (Order) arg;
            orderService.insert(order);
            
            // 返回成功
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            log.error("本地事务执行失败", e);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查：根据消息中的订单ID查询数据库
        String orderId = msg.getKeys();
        Order order = orderService.findById(orderId);
        
        if (order != null) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        // 注意：这里不要轻易返回UNKNOW，避免无限回查
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
});

producer.start();

// 发送事务消息
Message msg = new Message("order_topic", "物流系统", orderId.getBytes());
TransactionSendResult result = producer.sendMessageInTransaction(msg, order);

3.6 事务消息的局限性

说实话，RocketMQ事务消息也不是万能的。我总结了几点要注意的：

• 不支持延时消息：事务消息和延时消息不能同时使用。
• 回查次数有限：默认最多回查15次，超过后消息会被丢弃。
• 性能开销：相比普通消息，事务消息多了一次Half Message的存储和回查开销。
• 不适合高并发场景：如果每秒几万笔事务，Broker的回查压力会很大。

你想想看，如果业务场景能接受最终一致性，事务消息是个好选择。但如果要求强一致性，还是得考虑Seata这类分布式事务框架。

3.7 避坑总结