RocketMQ事务消息原理

可视化演示

展示事务消息半消息发送、本地事务执行、提交/回滚三阶段流程，回查补偿机制及与2PC的性能对比。

事务消息概述

在物流公司的故事里，如果你是调度员，最怕遇到什么？最怕的是“货发了，但订单没记录”或者“订单记了，货却没发”。在分布式系统中，这就是典型的“分布式事务问题”——两个系统各自有各自的数据库，没办法用一个本地事务把两边包起来。RocketMQ的事务消息，就是为了解决这个问题而设计的。

所谓“最终一致性”，简单说就是不要求任何一个瞬间两边数据都完全一样，但经过一段时间后，两边的数据最终会达成一致。这就像物流公司不会要求仓库盘货和客户确认订单在同一台电脑上同时完成，但只要流程走完，仓库的入库记录和客户的订单记录一定是匹配的。RocketMQ通过“半消息”加“确认机制”的方式来实现这种最终一致性。

RocketMQ事务消息的核心思想，其实借鉴了分布式事务中的“两阶段提交”协议。第一阶段：先发送一个“半消息”给Broker暂存，此时消息对消费者不可见；第二阶段：执行本地事务，根据结果决定是Commit（提交）还是Rollback（回滚）。这种设计让消息的发送和业务逻辑的执行之间有了一个“缓冲区”，不再要求两者必须同时成功。

发送流程

现在我们来看看，当李雷下单寄贵重货物时，这套机制具体是怎么运转的。作为调度员，你第一步要做的是：先给中转站发一个“预通知”——“有一批货要经过这里，先帮我占个位置！”但注意，此时中转站虽然收到了货，却不会把它装车发给客户，只是放在一个特殊的暂存区里。这个“预通知”对应的就是RocketMQ中的半消息。

为什么需要这个“暂存”的步骤呢？因为在分布式系统中，网络是可能出问题的。如果直接发消息给消费者，而本地事务还没执行，那么消费者可能会基于一个“尚未确认”的数据去做事，这就会产生数据不一致。所以Broker收到半消息后，会把它标记为“待确认”状态，对消费者不可见——就像中转站的暂存区货架，只有内部人员能看到，不会进入正常的快递分拣流程。

生产者发送半消息之后，Broker会返回一个发送结果。这个结果标志着第一阶段完成。这时候，你的物流系统就像一个在红绿灯前刹住的车——物流信息已经送出去了，但还没走远，随时可以掉头。如果你的仓库系统中发现李雷的订单信息有误，你可以让这条路直接作废；如果一切正常，你就可以踩下油门继续走。

这套流程就像物流调度中的“预占仓位”：先确保中转站有位置放货，但货物不会提前运送。只要中转站确认收货，这批货就“跑不掉”了——即使生产者宕机，半消息还留在Broker上，等待后续处理。

本地事务

半消息发送成功后，你作为调度员就要开始做正事了：处理李雷的订单。你需要在自己的订单系统里执行一系列操作——记录订单信息、更新客户账户、通知仓库备货。这一系列操作在你的系统里是一个本地事务，要么全部成功，要么全部失败。

为什么本地事务是整个流程的核心？因为它是判断“到底要不要真正发消息”的裁判。如果本地事务执行成功，说明李雷的订单一切正常，这时候就需要告诉Broker：“没问题，把暂存的半消息转正吧！”——这就是Commit操作。如果本地事务执行失败，比如发现李雷的余额不足，就需要告诉Broker：“不好意思，把暂存的半消息退回去吧！”——这就是Rollback操作。

说到Commit和Rollback，我们可以把它们类比为一个“决赛评审团”的投票过程。半消息暂时存放在Broker，相当于候选人在后台准备；本地事务就是评审团的内部讨论；最终投票结果可以是通过（Commit）或者不通过（Rollback），而Broker则根据投票结果来决定是否放行消息。这样的设计，确保了消息是否被投递，完全由业务逻辑的执行结果决定，两者之间没有任何时间差。

本地事务在这个机制中扮演“守护者”角色，确保消息发送不会绕过业务逻辑的判断。这就像快递员不会在仓库还没盘完货时就装车出发——本地事务成功，消息才被放行；本地事务失败，消息直接作废，从源头避免了数据不一致的可能性。

提交回查

现在来到了整个事务消息机制中最重要的环节：提交与回查。在理想情况下，生产者发送完半消息，执行完本地事务，会立即把Commit或Rollback指令告诉Broker，流程结束。但现实往往不理想——比如，生产者可能刚执行完本地事务，还没来得及发送Commit指令，就突然宕机或者网络断开了。这时候，Broker上的半消息就像被遗忘在中转站的包裹，既没有被转正，也没有被退回，处于一种“悬而未决”的状态。

这就是回查机制出场的时刻。Broker会定时检查那些长时间没有得到确认的半消息，发现超过一定时间还没收到Commit/Rollback指令，就会主动询问生产者：“喂，那批货到底要不要发？你的本地事务到底执行成功了没有？”——这个主动询问的过程，就是回查（CheckBack）。

生产者收到回查请求后，需要检查自己本地事务的执行结果，然后再次告诉Broker应该Commit还是Rollback。为了应对生产者可能已经宕机的情况，RocketMQ要求生产者在启动时就注册一个“回查监听器”（TransactionListener），这个监听器里包含了查询本地事务状态的逻辑。这样即使生产者重启了，也能通过查询数据库或日志来恢复对事务状态的判断。

回查机制体现了分布式系统的核心哲学：不要假设一切都会顺利运行，要为异常情况设计自动恢复的路径。就像快递公司不会因为送件员失联就让包裹永远丢在中转站，而是会派其他快递员去联系寄件人确认如何处理——回查确保了每条半消息都能得到最终的处理，不会出现“消息黑洞”。

回查补偿

如果生产者在回查后仍然无法给出明确答复怎么办？比如，生产者收到回查请求，但它的本地事务还在执行中，或者它自己也查询不到本地事务的结果（比如数据库也出了问题）。这时候，RocketMQ会根据生产者的返回值来决定下一步动作。

RocketMQ的回查监听器有三种返回值：COMMIT_MESSAGE（提交）、ROLLBACK_MESSAGE（回滚）、UNKNOWN（未知）。如果返回UNKNOWN，说明生产者自己也不确定本地事务的状态，Broker不会放弃，而是会再次进行回查。这个重试会持续多次，直到生产者给出明确的答复，或者超过最大回查次数后，消息被标记为处理失败，等待人工介入。

这种“不断追问”的策略，可以类比为财务对账流程。月初你给客户发了账单（半消息），月底还没收到付款（未确认），于是你开始催款（回查）。如果客户说“我再查查”（UNKNOWN），你下周继续催。经过几轮催款后，客户要么确认付款（Commit），要么说账单有误要作废（Rollback）。这个过程中，关键是“持续追踪、绝不放弃”，确保每一笔悬而未决的账务最终都有明确的归属。

回查补偿机制给整个系统加了“保险丝”——既能保证绝大多数消息在几次回查内就能明确去向，又防止了消息一直占着资源不放。就像物流公司对长期无人认领的包裹会最终转入失物招领处，而不是永远占着中转站的货架。

最终一致性

经过前面的机制，我们终于可以来回答最初的问题：RocketMQ事务消息如何保证最终一致性？其实思路非常简单：通过半消息的暂存、本地事务的执行、以及回查的兜底，让消息的可见性与业务逻辑的执行结果之间建立了强绑定关系。

我们来看一个完整的成功案例。李雷下单（发送半消息）→ Broker暂存 → 你的系统执行本地事务（扣库存、生成订单）→ 本地事务成功 → 告诉Broker提交 → Broker将半消息转为可见消息 → 消费者（比如李雷的App）收到消息推送。整个过程，消息的可见性始终由业务逻辑的执行结果决定，不存在“业务成功了但消息丢了”或者“业务失败了但消息发出去了”的可能性。

最终一致性还有一个重要的补充机制——幂等性设计。在回查过程中，生产者可能多次收到同一个消息的回查请求；消费者也可能因为消费重试而重复收到同一条消息。因此，RocketMQ的生产者和消费者都需要具备幂等处理能力。就像物流公司给每个包裹一个唯一的运单号，不管系统重复处理了多少次，只要认准运单号，就不会重复发货——RocketMQ生成的唯一消息ID就扮演了这个角色。

最终一致性是整个事务消息机制的“灵魂”。它告诉我们一个深刻的道理：在分布式系统中，追求“强一致”往往代价极高甚至不可能，但追求“最终一致”却是现实可行的。事务消息提供的方案——先暂存、后确认、异常兜底——正是这种哲学在消息中间件领域的完美落地。

AI 应用场景

Python 演示

import timeimport threading  class TransactionMessageBroker:    """模拟RocketMQ Broker的事务消息处理"""    def __init__(self):        self.pending_messages = {}  # msg_id -> message        self.visible_messages = {}  # msg_id -> message        self.check_count = {}  # msg_id -> 当前回查次数        self.max_check = 3        self.check_interval = 1     def send_half_message(self, msg_id, data):        """存储半消息，消费者不可见"""        self.pending_messages[msg_id] = {"data": data, "check_times": 0}        print(f"[Broker] 半消息已暂存 msg_id={msg_id}, data={data}")        return "SEND_OK"     def commit(self, msg_id):        """提交：半消息转为可见消息"""        if msg_id in self.pending_messages:            msg = self.pending_messages.pop(msg_id)            self.visible_messages[msg_id] = msg            print(f"[Broker] 消息已提交，消费者可见 msg_id={msg_id}")            return "COMMIT_OK"        return "MSG_NOT_FOUND"     def rollback(self, msg_id):        """回滚：删除半消息"""        if msg_id in self.pending_messages:            self.pending_messages.pop(msg_id)            print(f"[Broker] 消息已回滚删除 msg_id={msg_id}")            return "ROLLBACK_OK"        return "MSG_NOT_FOUND"     def check_timeout_messages(self, callback):        """扫描超时未确认的半消息，执行回查"""        for msg_id in list(self.pending_messages.keys()):            self.pending_messages[msg_id]["check_times"] += 1            check_times = self.pending_messages[msg_id]["check_times"]            if check_times > self.max_check:                print(f"[Broker] 超过最大回查次数，转人工处理 msg_id={msg_id}")                self.pending_messages.pop(msg_id)                continue            result = callback(msg_id)            print(f"[Broker] 回查结果 msg_id={msg_id}, result={result}")            if result == "COMMIT":                self.commit(msg_id)            elif result == "ROLLBACK":                self.rollback(msg_id)            else:                print(f"[Broker] 回查未知，准备下次重试 msg_id={msg_id}")  class TransactionProducer:    """模拟生产者：发送半消息、执行本地事务、回查监听"""    def __init__(self, broker):        self.broker = broker        self.local_tx_result = "COMMIT"  # 本地事务执行结果     def send_transaction_message(self, msg_id, data):        """完整的事务消息发送流程"""        if self.broker.send_half_message(msg_id, data) != "SEND_OK":            print("[Producer] 半消息发送失败，终止流程")            return        # 执行本地事务        success = self.execute_local_transaction(data)        if success:            self.broker.commit(msg_id)        else:            self.broker.rollback(msg_id)     def execute_local_transaction(self, data):        """模拟本地事务（如写数据库）"""        print(f"[Producer] 执行本地事务: 处理 data={data}")        return self.local_tx_result == "COMMIT"     def check_local_transaction(self, msg_id):        """回查监听：返回本地事务的实际状态"""        print(f"[Producer] 收到回查 msg_id={msg_id}，查询本地事务状态")        return self.local_tx_result  if __name__ == "__main__":    broker = TransactionMessageBroker()    producer = TransactionProducer(broker)     print("=== 场景1: 正常提交流程 ===")    producer.send_transaction_message("M001", "订单-1001")     print("\n=== 场景2: 本地事务失败回滚 ===")    producer.local_tx_result = "ROLLBACK"    producer.send_transaction_message("M002", "订单-1002")     print("\n=== 场景3: 模拟网络异常后回查 ===")    producer.local_tx_result = "COMMIT"    broker.send_half_message("M003", "订单-1003")    # 模拟发送Commit前断网，Broker触发回查恢复    broker.check_timeout_messages(producer.check_local_transaction)     print("\n=== 最终Broker中的消息 ===")    print(f"可见消息: {list(broker.visible_messages.keys())}")    print(f"待确认消息: {list(broker.pending_messages.keys())}")

1. self.pending_messages[msg_id] = {"data": data, "check_times": 0}半消息存储在Broker的待确认集合中，消费者无法看到
2. def send_half_message(self, msg_id, data)实现半消息的发送和暂存逻辑，返回SEND_OK表示成功
3. def commit(self, msg_id)将半消息从pending集合转移到visible集合，消息对消费者可见
4. def rollback(self, msg_id)直接从pending集合中删除半消息，相当于撤销预占
5. self.pending_messages[msg_id]["check_times"] += 1记录每次回查次数，防止无限重试
6. if check_times > self.max_check超过最大回查次数后，转人工处理，避免死循环
7. def execute_local_transaction(self, data)模拟生产者执行本地业务事务的场景
8. def check_local_transaction(self, msg_id)回查监听器，Broker通过它询问本地事务的真实状态
9. broker.check_timeout_messages(producer.check_local_transaction)手动触发回查流程，模拟异常恢复

常见误区

面试题