RocketMQ 事务消息详解

1. 事务消息简介

RocketMQ的事务消息，是指发送消息事件和其他事件需要同时成功或同时失败。比如银行转账， A银行的某账户要转一万元到B银行的某账户。A银行发送“B银行账户增加一万元”这个消息，要和“从A银 行账户扣除一万元”这个操作同时成功或者同时失败。
RocketMQ采用两阶段提交的方式实现事务消息，TransactionMQProducer处理上面情况的流程 是，先发一个“准备从B银行账户增加一万元”的消息，发送成功后做从A银行账户扣除一万元的操作，根 据操作结果是否成功，确定之前的“准备从B银行账户增加一万元”的消息是做commit还是rollback。
具体流程如下:

1. 发送方向RocketMQ发送“待确认”消息。
2. RocketMQ将收到的“待确认”消息持久化成功后，向发送方回复消息已经发送成功，此时第一阶段消息发送完成。
3. 发送方开始执行本地事件逻辑。
4. 发送方根据本地事件执行结果向RocketMQ发送二次确认(Commit或是Rollback)消息， RocketMQ收到Commit状态则将第一阶段消息标记为可投递，订阅方将能够收到该消息;收到 Rollback状态则删除第一阶段的消息，订阅方接收不到该消息。
5. 如果出现异常情况，步骤4)提交的二次确认最终未到达RocketMQ，服务器在经过固定时间段 后将对“待确认”消息发起回查请求。
6. 发送方收到消息回查请求后(如果发送一阶段消息的Producer不能工作，回查请求将被发送到 和Producer在同一个Group里的其他Producer)，通过检查对应消息的本地事件执行结果返回Commit 或Roolback状态。
7. RocketMQ收到回查请求后，按照步骤4)的逻辑处理。

上面的逻辑似乎很好地实现了事务消息功能，它也是RocketMQ之前的版本实现事务消息的逻辑。 但是因为RocketMQ依赖将数据顺序写到磁盘这个特征来提高性能，步骤4)却需要更改第一阶段消息 的状态，这样会造成磁盘Catch的脏页过多，降低系统的性能。所以RocketMQ在4.x的版本中将这部分 功能去除。系统中的一些上层Class都还在，用户可以根据实际需求实现自己的事务功能。
客户端有三个类来支持用户实现事务消息，
第一个类是LocalTransaction-Executer，用来实例化步 骤3)的逻辑，根据情况返回LocalTransactionState.ROLLBACK_MESSAGE或者 LocalTransactionState.COMMIT_MESSAGE状态。
第二个类是TransactionMQProducer，它的用法和 DefaultMQProducer类似，要通过它启动一个Producer并发消息，但是比DefaultMQProducer多设置 本地事务处理函数和回查状态函数。
第三个类是TransactionCheckListener，实现步骤5)中MQ服务器 的回查请求，返回LocalTransactionState.ROLLBACK_MESSAGE或者
或者LocalTransactionState.COMMIT_MESSAGE

2. 事务消息流程概要

上图说明了事务消息的大致方案，其中分为两个流程:正常事务消息的发送及提交、事务消息的补偿流程。

1.事务消息发送及提交:

1. 发送消息(half消息)。
2. 服务端响应消息写入结果。
3. 根据发送结果执行本地事务(如果写入失败，此时half消息对业务不可见，本地逻辑不执行)。
4. 根据本地事务状态执行Commit或者Rollback(Commit操作生成消息索引，消息对消费者可 见)

2.补偿流程:

1. 对没有Commit/Rollback的事务消息(pending状态的消息)，从服务端发起一次“回查”
2. Producer收到回查消息，检查回查消息对应的本地事务的状态
3. 根据本地事务状态，重新Commit或者Rollback 其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

3. 事务消息设计

1.事务消息在一阶段对用户不可见

在RocketMQ事务消息的主要流程中，一阶段的消息如何对用户不可见。其中，事务消息相对普通 消息最大的特点就是一阶段发送的消息对用户是不可见的。那么，如何做到写入消息但是对用户不可见 呢?RocketMQ事务消息的做法是:如果消息是half消息，将备份原消息的主题与消息消费队列，然后 改变主题为RMQ_SYS_TRANS_HALF_TOPIC。由于消费组未订阅该主题，故消费端无法消费half类型的 消息。然后二阶段会显示执行提交或者回滚half消息(逻辑删除)。当然，为了防止二阶段操作失败， RocketMQ会开启一个定时任务，从Topic为RMQ_SYS_TRANS_HALF_TOPIC中拉取消息进行消费，根 据生产者组获取一个服务提供者发送回查事务状态请求，根据事务状态来决定是提交或回滚消息。

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

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

2.Commit和Rollback操作以及Op消息的引入

在完成一阶段写入一条对用户不可见的消息后，二阶段如果是Commit操作，则需要让消息对用户 可见;如果是Rollback则需要撤销一阶段的消息。先说Rollback的情况。对于Rollback，本身一阶段的 消息对用户是不可见的，其实不需要真正撤销消息(实际上RocketMQ也无法去真正的删除一条消息， 因为是顺序写文件的)。但是区别于这条消息没有确定状态(Pending状态，事务悬而未决)，需要一 个操作来标识这条消息的最终状态。RocketMQ事务消息方案中引入了Op消息的概念，用Op消息标识 事务消息已经确定的状态(Commit或者Rollback)。如果一条事务消息没有对应的Op消息，说明这个 事务的状态还无法确定(可能是二阶段失败了)。引入Op消息后，事务消息无论是Commit或者 Rollback都会记录一个Op操作。Commit相对于Rollback只是在写入Op消息前创建Half消息的索引。

3.Op消息的存储和对应关系

RocketMQ将Op消息写入到全局一个特定的Topic中通过源码中的方法— TransactionalMessageUtil.buildOpTopic();这个Topic是一个内部的Topic(像Half消息的Topic一 样)，不会被用户消费。Op消息的内容为对应的Half消息的存储的Offset，这样通过Op消息能索引到 Half消息进行后续的回查操作。

4.Half消息的索引构建

在执行二阶段Commit操作时，需要构建出Half消息的索引。一阶段的Half消息由于是写到一个特 殊的Topic，所以二阶段构建索引时需要读取出Half消息，并将Topic和Queue替换成真正的目标的 Topic和Queue，之后通过一次普通消息的写入操作来生成一条对用户可见的消息。所以RocketMQ事 务消息二阶段其实是利用了一阶段存储的消息的内容，在二阶段时恢复出一条完整的普通消息，然后走 一遍消息写入流程。

5.如何处理二阶段失败的消息?

如果在RocketMQ事务消息的二阶段过程中失败了，例如在做Commit操作时，出现网络问题导致 Commit失败，那么需要通过一定的策略使这条消息最终被Commit。RocketMQ采用了一种补偿机制， 称为“回查”。Broker端对未确定状态的消息发起回查，将消息发送到对应的Producer端(同一个Group 的Producer)，由Producer根据消息来检查本地事务的状态，进而执行Commit或者Rollback。 Broker端通过对比Half消息和Op消息进行事务消息的回查并且推进CheckPoint(记录那些事务消息的 状态是确定的)。
值得注意的是，rocketmq并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还 是无法得知事务状态，rocketmq默认回滚该消息。

事务消息共有三种状态，提交状态、回滚状态、中间状态：

• TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
• TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
• TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。

4. 创建事务性生产者

使用 TransactionMQProducer类创建生产者，并指定唯一的 ProducerGroup，就可以设置自定义线程池来处理这些检查请求。执行本地事务后、需要根据执行结果对消息队列进行回复。回传的事务状态参考上面。

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class TransactionProducer {
   public static void main(String[] args) throws MQClientException, InterruptedException {
       TransactionListener transactionListener = new TransactionListenerImpl();
       TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
       ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
           @Override
           public Thread newThread(Runnable r) {
               Thread thread = new Thread(r);
               thread.setName("client-transaction-msg-check-thread");
               return thread;
           }
       });
       producer.setExecutorService(executorService);
       producer.setTransactionListener(transactionListener);
       producer.start();
       String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
       for (int i = 0; i < 10; i++) {
           try {
               Message msg =
                   new Message("TopicTest1234", 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);
           } catch (MQClientException | UnsupportedEncodingException e) {
               e.printStackTrace();
           }
       }
       for (int i = 0; i < 100000; i++) {
           Thread.sleep(1000);
       }
       producer.shutdown();
   }
}

5. 实现事务的监听接口

当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。checkLocalTransaction 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。

public class TransactionListenerImpl implements TransactionListener {
  private AtomicInteger transactionIndex = new AtomicInteger(0);
  private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
  @Override
  public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
      int value = transactionIndex.getAndIncrement();
      int status = value % 3;
      localTrans.put(msg.getTransactionId(), status);
      return LocalTransactionState.UNKNOW;
  }
  @Override
  public LocalTransactionState checkLocalTransaction(MessageExt msg) {
      Integer status = localTrans.get(msg.getTransactionId());
      if (null != status) {
          switch (status) {
              case 0:
                  return LocalTransactionState.UNKNOW;
              case 1:
                  return LocalTransactionState.COMMIT_MESSAGE;
              case 2:
                  return LocalTransactionState.ROLLBACK_MESSAGE;
          }
      }
      return LocalTransactionState.COMMIT_MESSAGE;
  }
}

6. 事务消息使用上的限制

1. 事务消息不支持延时消息和批量消息。
2. 为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionalMessageCheckListener 类来修改这个行为。
3. 事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionTimeout 参数。
4. 事务性消息可能不止一次被检查或消费。
5. 提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
6. 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。