RocketMQ源码：事务消息原理

电玩女神 2024-03-22 11:49 164阅读 0赞

RocketMQ在**4.3.0**版中开始支持**分布式事务消息**，RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示。

![format_png][]

## 1.RocketMQ事务消息流程概要 ##

上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。  
**1.事务消息发送及提交**：  
(1) 发送消息（这个消息暂时称为：`half消息`(半事务消息))  
(2) 服务端响应消息写入结果。  
(3) 根据发送结果执行本地事务（如果写入失败，此时`half消息`对业务不可见，本地逻辑不执行）  
(4) 根据本地事务状态执行`Commit`或者`Rollback`（Commit操作生成消息索引，消息对消费者可见）

**2.补偿流程**：  
(1) 对没有`Commit/Rollback`的事务消息（pending状态的消息），从服务端发起一次“回查”  
(2) Producer收到回查消息，检查回查消息对应的本地事务的状态  
(3) 根据本地事务状态，重新`Commit`或者`Rollback`

> 其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

## 2. RocketMQ事务消息设计 ##

在分析源码之前，我们先补充一些概念

### 2.1 分布式事务 ###

对于分布式事务，通俗地说就是，一次操作由若干分支操作组成，这些分支操作分属不同应用，分布在不同服务器上。分布式事务需要保证这些分支操作要么全部成功，要么全部失败。分布式事务与普通事务一样，就是为了保证操作结果的一致性。

> 与之对应的就是本地事务：本地事务更多的是通过关系型数据库来控制事务，这是利用数据库本身的事务特性来实现的，因此叫数据库事务。数据库事务具有ACID特性，在实现时会将一次事务的所有操作全部纳入到一个不可分割的执行单元，该执行单元的所有操作要么都成功，要么都失败，只要其中任一操作执行失败，都将导致整个事务的回滚。

### 2.2 事务消息 ###

RocketMQ提供了类似`X/Open XA`的分布式事务功能，通过事务消息能达到分布式事务的最终一致。XA是一种分布式事务解决方案，一种分布式事务处理模式。

### 2.3 半事务消息 ###

暂不能消费的消息，Producer已经成功地将消息发送到了Broker，但是Broker未收到最终确认指令，此时该消息被标记成“`暂不能消费`”状态，即不能被消费者看到。处于该种状态下的消息即**半事务消息**。

### 2.4 消息回查 ###

**消息回查**，即重新查询本地事务的执行状态（也就是上图中的步骤6）。假如张三给李四转账，张三首先发送一条需要转账的消息到MQ, 发送成功后，张三开始从建设银行卡扣款，可能由于网络的问题，张三在扣款的时候发生了故障，出现了扣款未知(`UNKNOW`, 就不是成功也不是失败)的状态，紧接着张三将这条`UNKNOW`状态的消息发给MQ, MQ 接收到是`UNKNOW`状态，则需要发起回查，给张三个机会，看看网络是不是好了，好了就扣款，如果是余额不足，那就结束转账。在RocketMQ中，事务消息有三个状态：

public enum LocalTransactionState {
        //TODO:本地事务执行成功，给broker发送一个commit的标识
        COMMIT_MESSAGE,
        
        ROLLBACK_MESSAGE,
        
        //TODO: 这个状态将会引起回查
        UNKNOW,
    }
    复制代码
    复制代码

其中 `LocalTransactionState.UNKNOW` 状态，将会引起回查。

### 2.5 XA协议 ###

`XA（Unix Transaction）`是一种分布式事务解决方案，一种分布式事务处理模式，是基于XA协议的。`XA`协议由Tuxedo（Transaction for Unix has been Extended for Distributed Operation，分布式操作扩展之后的Unix事务系统）首先提出的，并交给X/Open组织，作为资源管理器与事务管理器的接口标准.

XA模式中有三个重要组件：**TC**、**TM**、**RM**

*  **TC** :`Transaction Coordinator`，事务协调者。维护全局和分支事务的状态，驱动全局事务提交或回滚。

> RocketMQ中Broker充当着TC。

*  **TM** : `Transaction Manager`，事务管理器。定义全局事务的范围：开始全局事务、提交或回滚全局事务。它实际是全局事务的发起者。

> RocketMQ中事务消息的Producer充当着TM。

*  **RM** : `Resource Manager`，资源管理器。管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

> RocketMQ中事务消息的Producer及Broker均是RM

XA模式是一个典型的2PC，其执行原理如下:

1.  TM向TC发起指令，开启一个全局事务。
2.  根据业务要求，各个RM会逐个向TC注册分支事务，然后TC会逐个向RM发出预执行指令。
3.  各个RM在接收到指令后会在进行本地事务预执行。
4.  RM将预执行结果上报给TC。当然，这个结果可能是成功，也可能是失败。
5.  TC在接收到各个RM的Report后会将汇总结果上报给TM，根据汇总结果TM会向TC发出确认指令。TC在接收到指令后再次向RM发送确认指令。

> 1.  若所有结果都是成功响应，则向TC发送Global Commit指令。
> 2.  只要有结果是失败响应，则向TC发送Global Rollback指令。

1.  TC在接收到指令后再次向RM发送确认指令。

了解这些基本概念后，我们在回过头来看RocketMQ的事务消息的设计

1.  **事务消息在一阶段对用户(消费者)不可见**  
    在RocketMQ事务消息的主要流程中，一阶段的消息如何**对用户不可见**。其中，事务消息相对普通消息最大的特点就是一阶段发送的消息对用户是不可见的。那么，如何做到写入消息但是对用户不可见呢？RocketMQ事务消息的做法是：如果消息是half消息，将备份原消息的主题与消息消费队列，然后改变主题为**RMQ\_SYS\_TRANS\_HALF\_TOPIC**。由于消费组未订阅该主题，故消费端无法消费half类型的消息，然后RocketMQ会开启一个定时任务，从Topic为RMQ\_SYS\_TRANS\_HALF\_TOPIC中拉取消息进行消费，根据生产者组获取一个服务提供者发送回查事务状态请求，根据事务状态来决定是提交或回滚消息。

在RocketMQ中，消息在服务端的存储结构如下，每条消息都会有对应的索引信息，Consumer通过ConsumeQueue这个二级索引来读取消息实体内容，其流程如下：

![format_png 1][]

RocketMQ的具体实现策略是：写入的如果事务消息，对消息的Topic和Queue等属性进行替换，同时将原来的Topic和Queue信息存储到消息的属性中，正因为消息主题被替换，故消息并不会转发到该原主题的消息消费队列，消费者无法感知消息的存在，不会消费。其实改变消息主题是RocketMQ的常用“**套路**”，回想一下延时消息的实现机制。

1.  **Commit和Rollback操作以及Op消息的引入**  
    在完成一阶段写入一条对用户不可见的消息后，二阶段如果是Commit操作，则需要让消息对用户可见；如果是Rollback则需要撤销一阶段的消息。先说Rollback的情况。对于Rollback，本身一阶段的消息对用户是不可见的，其实不需要真正撤销消息（实际上RocketMQ也无法去真正的删除一条消息，因为是顺序写文件的）。但是区别于这条消息没有确定状态（Pending状态，事务悬而未决），需要一个操作来标识这条消息的最终状态。RocketMQ事务消息方案中引入了Op消息的概念，用Op消息标识事务消息已经确定的状态（Commit或者Rollback）。如果一条事务消息没有对应的Op消息，说明这个事务的状态还无法确定（可能是二阶段失败了）。引入Op消息后，事务消息无论是Commit或者Rollback都会记录一个Op操作。Commit相对于Rollback只是在写入Op消息前创建Half消息的索引。
2.  **Op消息的存储和对应关系**  
    RocketMQ将Op消息写入到全局一个特定的Topic中通过源码中的方法—TransactionalMessageUtil.buildOpTopic()；这个Topic是一个内部的Topic（像Half消息的Topic一样），不会被用户消费。Op消息的内容为对应的Half消息的存储的Offset，这样通过Op消息能索引到Half消息进行后续的回查操作.

![format_png 2][]

1.  **Half消息的索引构建**  
    在执行二阶段Commit操作时，需要构建出Half消息的索引。一阶段的Half消息由于是写到一个特殊的Topic，所以二阶段构建索引时需要读取出Half消息，并将Topic和Queue替换成真正的目标的Topic和Queue，之后通过一次普通消息的写入操作来生成一条对用户可见的消息。所以RocketMQ事务消息二阶段其实是利用了一阶段存储的消息的内容，在二阶段时恢复出一条完整的普通消息，然后走一遍消息写入流程。
2.  **如何处理二阶段失败的消息？**  
    如果在RocketMQ事务消息的二阶段过程中失败了，例如在做Commit操作时，出现网络问题导致Commit失败，那么需要通过一定的策略使这条消息最终被Commit。RocketMQ采用了一种补偿机制，称为“**回查**”。Broker端对未确定状态的消息发起回查，将消息发送到对应的Producer端（同一个Group的Producer），由Producer根据消息来检查本地事务的状态，进而执行Commit或者Rollback。Broker端通过对比Half消息和Op消息进行事务消息的回查并且推进CheckPoint（记录那些事务消息的状态是确定的）。

> 值得注意的是，rocketmq并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，rocketmq默认回滚该消息。

## 3. 从源码看事务消息 ##

### 3.1 准备测试用例 ###

public class TransactionProducer {
    
        public static void main(String[] args) throws MQClientException, InterruptedException {
    
            //TODO:定义事务消息的发送者，它继承了DefaultMQProducer（就是发送普通消息的类）
            TransactionMQProducer producer = new TransactionMQProducer(MQConstant.DEFAULT_PRODUCER_GROUP_NAME);
            //TODO:指定一个线程池，如果不指定MQ本身也会为我们创建一个，一般都会主动创建。消息回查时将用到
            ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000), new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setName("client-transaction-msg-check-thread");
                    return thread;
                }
            });
    
            producer.setNamesrvAddr(MQConstant.NAME_SERVER_ADDR);
            producer.setExecutorService(executorService);
            //TODO: 指定本地事务监听器
            TransactionListener transactionListener = new MyLocalTransactionCheckListener();
            producer.setTransactionListener(transactionListener);
    
            //TODO:producer启动
            producer.start();
    
            String[] tags = new String[]{"tagA", "tagB", "tagC"};
    
            for (int i = 0; i < 3; i++) {
                try {
                    Message msg =
                            new Message(MQConstant.TX_TOPIC, tags[i], "KEY" + i,
                                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    
                    //第二个参数用于指定在执行本地事务时要使用的业务参数
                    SendResult sendResult = producer.sendMessageInTransaction(

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/94b0bb5d765843e38cc313f62612f4c6.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/36b76a736a284ffca6b02b249e2a435d.png
[format_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/87e013d61f1544178ae630cae5a312db.png