【RocketMQ 生产者和消费者】- 事务消息的使用

本文章基于 RocketMQ 4.9.3

1. 前言

• 【RocketMQ】- 源码系列目录
• 【RocketMQ 生产者消费者】- 同步、异步、单向发送消费消息
• 【RocketMQ 生产者和消费者】- 消费者启动源码
• 【RocketMQ 生产者和消费者】- 消费者重平衡（1）
• 【RocketMQ 生产者和消费者】- 消费者重平衡（2）- 分配策略
• 【RocketMQ 生产者和消费者】- 消费者重平衡（3）- 消费者 ID 对负载均衡的影响
• 【RocketMQ 生产者和消费者】- 消费者的订阅关系一致性
• 【RocketMQ 生产者和消费者】- 消费者发起消息拉取请求 PullMessageService
• 【RocketMQ 生产者和消费者】- broker 是如何处理消费者消息拉取的 Netty 请求的
• 【RocketMQ 生产者和消费者】- broker 处理消息拉取请求
• 【RocketMQ 生产者和消费者】- 消费者处理消息拉取结果
• 【RocketMQ 生产者和消费者】- 消费者处理消息拉取结果
• 【RocketMQ 生产者和消费者】- ConsumeMessageConcurrentlyService 并发消费消息
• 【RocketMQ 生产者和消费者】- ConsumeMessageOrderlyService 顺序消费消息
• 【RocketMQ 生产者和消费者】- sendMessageBack 发送重试消息
• 【RocketMQ 生产者和消费者】- 延时消息的使用
• 【RocketMQ 生产者和消费者】- 延时消息原理解析-ScheduleMessageService

上两篇文章我们分析了延时消息的基本使用和原理，这篇文章就来看下事务消息的基本使用。

2. 事务消息

首先介绍下什么是事务消息，也可以看官网的介绍：事务消息发送。

事务消息可以用在消息队列中确保消息传递和业务操作原子性、一致性，尤其适用于分布式系统环境，特别是消息上下游服务相互依赖的场景。比如下面的两个场景：

• 订单支付：比如上游支付订单之后扣减本地库存，更新订单状态，接着通知下游服务比如物流系统做后续的处理，这种情况下可以先发送一条事务消息，然后执行本地事务逻辑（扣减库存，更新订单状态），做完之后根据本地事务消息的结果决定这条消息能不能被下游的服务看到，如果本地事务执行失败，那么下游也不需要继续处理。
• 金融系统中的转账业务：转账同时扣减转出账户余额和增加转入账户余额。用事务消息在发送转账成功消息之前执行本地账户余额更新事务，确保消息发送与账户余额变更操作的一致性。

上面说的本地事务可以是操作数据库、操作缓存等等，反正就是一个方法，根据业务来处理，下面就是大致的流程，这个图也是直接用官方的图了。

上面涉及到几个名词下面解释下：

• 半事务消息：半事务消息是指暂不能投递的消息，当生产者发送半事务消息到 Broker后，Broker 收到消息会将其先存储起来，但不会马上将该消息投递给消费者，此时消费者对这条消息不可见。只有生产者对该消息进行二次确认（即提交或回滚操作）后，Broker 才会根据确认结果进行相应处理。
• 本地事务：本地事务就是本地的逻辑，当生产者执行完本地事务之后，根据执行结果返回对应的状态码，broker 会根据对应结果来处理这条半事务消息。本地事务有三种状态：
  • UNKNOW表示中间状态，代表需要通过事务回查来确定最终的执行结果；
  • COMMIT_MESSAGE表示本地事务执行成功，此消息可以被消费者消费；
  • ROLLBACK_MESSAGE表示事务回滚，这条消息会被删掉，不会被消费者消费。
• 事务回查：如果由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，Broker 端会通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或是 Rollback），如果业务阻塞也可以通过这种方式给一个兜底的结果，在 broker.conf 可以配置transactionCheckInterval回查的时间间隔，根据自己实际业务决定。

下面把整个流程串起来：

1. 生产者将半事务消息发送至 broker。
2. broker 将这条消息持久化，这条消息属于半事务消息，还不可被消费者消费。
3. 生产者通过executeLocalTransaction开始执行本地事务逻辑。
4. 本地事务执行完成后，生产者向 broker 提交二次确认结果（Commit 或是 Rollback），服务端走不同的处理逻辑：
   • 二次确认结果为COMMIT_MESSAGE：服务端将半事务消息标记为可投递，并投递给消费者。
   • 二次确认结果为ROLLBACK_MESSAGE：服务端将回滚事务，不会将半事务消息投递给消费者。
   • 二次确认结果为UNKNOW：服务端将回滚事务，不会将半事务消息投递给消费者。
5. 在断网或者是生产者应用重启的特殊情况下，broker 迟迟收不到事务的二次提交结果，又或者如果事务提交结果给的是UNKNOW，那么服务端会对生产者发起事务回查。
6. 为了避免单个消息被检查太多次而导致半队列消息累积，默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的transactionCheckMax参数来修改检查次数。如果回查次数到达上限了，那么 broker 就丢掉这条消息，然后在默认情况下同时打印错误日志。不过我们可以通过重写AbstractTransactionalMessageCheckListener、 类来修改回查上限之后要做什么。
7. 生产者收到消息回查后，通过checkLocalTransaction检查对应消息的本地事务执行的最终结果。
8. 根据检查出来的结果进行二次提交，这时候又回到了 4 的逻辑。

3. 示例

下面看下事务消息的使用，首先在 broker 配置下消息回查的时间。

transactionCheckInterval=20000

然后设置下生产者，生产者一共发送 10 条消息，下标从 0 开始计算，然后每一条消息发送的 tag 不一样。

publicclassTransactionProducer{publicstaticvoidmain(String[]args)throwsMQClientException,InterruptedException{TransactionListenertransactionListener=newTransactionListenerImpl();// 事务消息生产者TransactionMQProducerproducer=newTransactionMQProducer("transactionMQProducer");// 执行本地回查的线程池ExecutorServiceexecutorService=newThreadPoolExecutor(2,5,100,TimeUnit.SECONDS,newArrayBlockingQueue<Runnable>(2000),newThreadFactory(){@OverridepublicThreadnewThread(Runnabler){Threadthread=newThread(r);thread.setName("client-transaction-msg-check-thread");returnthread;}});producer.setNamesrvAddr("localhost:9876");producer.setExecutorService(executorService);// 设置事务监听器producer.setTransactionListener(transactionListener);// 设置本地事务回查时间间隔producer.start();String[]tags=newString[]{"TagA","TagB","TagC","TagD","TagE"};for(inti=0;i<10;i++){try{// 消息发送Messagemsg=newMessage("TopicTest1234",tags[i%tags.length],"KEY"+i,("Hello RocketMQ "+i).getBytes(RemotingHelper.DEFAULT_CHARSET));SendResultsendResult=producer.sendMessageInTransaction(msg,null);System.out.printf(now()+" %s%n",sendResult);Thread.sleep(10);}catch(MQClientException|UnsupportedEncodingExceptione){e.printStackTrace();}}for(inti=0;i<100000;i++){Thread.sleep(1000);}producer.shutdown();}}

接下来初始化事务监听器TransactionListenerImpl，实现 TransactionListener 接口，在里面定义executeLocalTransaction本地事务执行的逻辑以及checkLocalTransaction本地事务回查的逻辑。

publicclassTransactionListenerImplimplementsTransactionListener{privateAtomicIntegertransactionIndex=newAtomicInteger(0);privateConcurrentHashMap<String,Integer>localTrans=newConcurrentHashMap<>();/** * 执行本地事务 * @param msg Half(prepare) message * @param arg Custom business parameter * @return */@OverridepublicLocalTransactionStateexecuteLocalTransaction(Messagemsg,Objectarg){// 执行本地事务, 根据不同下标在下面的 checkLocalTransaction 方法走不同的逻辑intvalue=transactionIndex.getAndIncrement();intstatus=value%3;localTrans.put(msg.getTransactionId(),status);// 全部返回 UNKNOW, 意味者要通过事务回查 checkLocalTransaction 方法去检查本地事务的提交结果returnLocalTransactionState.UNKNOW;}/** * 本地事务回查 * @param msg Check message * @return */@OverridepublicLocalTransactionStatecheckLocalTransaction(MessageExtmsg){System.out.println(now()+" checkLocalTransaction 执行本地事务回查, 当前消息事务 ID: "+msg.getTransactionId());Integerstatus=localTrans.get(msg.getTransactionId());if(null!=status){switch(status){case0:returnLocalTransactionState.UNKNOW;case1:returnLocalTransactionState.COMMIT_MESSAGE;case2:returnLocalTransactionState.ROLLBACK_MESSAGE;default:returnLocalTransactionState.COMMIT_MESSAGE;}}returnLocalTransactionState.COMMIT_MESSAGE;}}

executeLocalTransaction执行本地事务的时候，会先通过 transactionIndex + 1 算出这条消息是哪个下标，然后在事务回查checkLocalTransaction的时候将所有消息分为三类来处理。

• 0：LocalTransactionState.UNKNOW
• 1：LocalTransactionState.COMMIT_MESSAGE
• 2：LocalTransactionState.ROLLBACK_MESSAGE

大致流程就是执行本地事务的时候LocalTransactionState.UNKNOW，broker 会对每一条消息都进行回查，然后在回查里面会根据不同的下标来返回不同的结果，结合 executeLocalTransaction 方法，可以大概看出消息0、3、6、9会不断回查，1、4、7回查之后发现是COMMIT_MESSAGE，消息提交，消费者可以消费到这几条消息，2、5、8回查之后是ROLLBACK_MESSAGE，这几条消息会被丢掉，后续回查也不会对这几条消息发起回查了。

然后是消费者，消费者比较简单，就是订阅TopicTest1234下面的所有 tags 进行消费。

publicclassTransactionConsumer{publicstaticvoidmain(String[]args)throwsInterruptedException,MQClientException{DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("testGroupConsumer");consumer.setNamesrvAddr("localhost:9876");consumer.setMessageModel(MessageModel.CLUSTERING);consumer.subscribe("TopicTest1234","*");consumer.registerMessageListener(newMessageListenerConcurrently(){@OverridepublicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeConcurrentlyContextcontext){for(inti=0;i<msgs.size();i++){MessageExtmsg=msgs.get(i);System.out.printf("%s, %s Receive New Messages: %s %n",now(),Thread.currentThread().getName(),newString(msg.getBody(),StandardCharsets.UTF_8));context.setAckIndex(i);}returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.printf("Consumer Started.%n");}publicstaticStringnow(){returnnewSimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(newDate());}}

输出结果如下，首先是消费者，可以看到确实只消费了 1、4、7 三条消息。

然后是生产者，生产者发送消息之后由于本地事务执行都是返回的 `LocalTransactionState.UNKNOW``，所以经过一定时间后会消息回查，这里为什么不是 20s，后面源码会分析。

经过事务回查之后，发现2、5、8返回了ROLLBACK_MESSAGE，于是后续不会再回查这几条消息，而0、3、6、9由于还是返回UNKNOW，于是就一直不断回查，但是回查次数有上限，所以也不会一直回查，不过时间太长了，就不贴出所有输出。

4. 使用限制

这里的限制也是代码给的文档里面标注出来的，下面贴出来：

• 事务消息不支持延时消息和批量消息。
• 提交给用户的目标主题消息可能会失败，它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
• 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ 服务器能通过它们的生产者 ID 查询到消费者。

5. 小结

好了，到这里事务消息的使用和大致流程已经介绍完，下一篇文章就要进入源码分析环节。

如有错误，欢迎指出！！！