解析rocketmq业务消息——“事务消息”

关键词：

引言：在分布式系统调用场景中存在这样一个通用问题，即在执行一个核心业务逻辑的同时，还需要调用多个下游做业务处理，而且要求多个下游业务和当前核心业务必须同时成功或者同时失败，进而避免部分成功和失败的不一致情况出现。简单来说，消息队列中的“事务”，主要解决的是消息生产者和消费者的数据一致性问题。本篇文章通过拆解 RocketMQ 事务消息的使用场景、基本原理、实现细节和实战使用，帮助大家更好的理解和使用 RocketMQ 的事务消息。

点击链接，查看视频讲解：https://yqh.aliyun.com/live/detail/29199

场景：为什么需要事务消息

以电商交易场景为例，用户支付订单这一核心操作的同时会涉及到下游物流发货、积分变更、购物车状态清空等多个子系统的变更。当前业务的处理分支包括：

• 主分支订单系统状态更新：由未支付变更为支付成功；
• 物流系统状态新增：新增待发货物流记录，创建订单物流记录；
• 积分系统状态变更：变更用户积分，更新用户积分表；
• 购物车系统状态变更：清空购物车，更新用户购物车记录。

分布式系统调用的特点是：一个核心业务逻辑的执行，同时需要调用多个下游业务进行处理。因此，如何保证核心业务和多个下游业务的执行结果完全一致，是分布式事务需要解决的主要问题。

传统 XA 事务方案：性能不足

为了保证上述四个分支的执行结果一致性，典型方案是基于XA协议的分布式事务系统来实现。将四个调用分支封装成包含四个独立事务分支的大事务，基于XA分布式事务的方案可以满足业务处理结果的正确性，但最大的缺点是多分支环境下资源锁定范围大，并发度低，随着下游分支的增加，系统性能会越来越差。

基于普通消息方案：一致性保障困难

将上述基于 XA 事务的方案进行简化，将订单系统变更作为本地事务，剩下的系统变更作为普通消息的下游来执行，事务分支简化成普通消息+订单表事务，充分利用消息异步化的能力缩短链路，提高并发度。

该方案中消息下游分支和订单系统变更的主分支很容易出现不一致的现象，例如：

• 消息发送成功，订单没有执行成功，需要回滚整个事务；
• 订单执行成功，消息没有发送成功，需要额外补偿才能发现不一致；
• 消息发送超时未知，此时无法判断需要回滚订单还是提交订单变更。

基于RocketMQ分布式事务消息：支持最终一致性

上述普通消息方案中，普通消息和订单事务无法保证一致的本质原因是普通消息无法像单机数据库事务一样，具备提交、回滚和统一协调的能力。

而基于消息队列 RocketMQ 版实现的分布式事务消息功能，在普通消息基础上，支持二阶段的提交能力。将二阶段提交和本地事务绑定，实现全局提交结果的一致性。

消息队列 RocketMQ 版事务消息的方案，具备高性能、可扩展、业务开发简单的优势。

基本原理

概念介绍

• 事务消息：RocketMQ 提供类似 XA 或 Open XA 的分布式事务功能，通过 RocketMQ 事务消息能达到分布式事务的最终一致；
• 半事务消息：暂不能投递的消息，生产者已经成功地将消息发送到了 RocketMQ 服务端，但是 RocketMQ 服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息；
• 消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，RocketMQ 服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或是 Rollback），该询问过程即消息回查。

事务消息生命周期

• 初始化：半事务消息被生产者构建并完成初始化，待发送到服务端的状态；
• 事务待提交：半事务消息被发送到服务端，和普通消息不同，并不会直接被服务端持久化，而是会被单独存储到事务存储系统中，等待第二阶段本地事务返回执行结果后再提交。此时消息对下游消费者不可见；
• 消息回滚：第二阶段如果事务执行结果明确为回滚，服务端会将半事务消息回滚，该事务消息流程终止；
• 提交待消费：第二阶段如果事务执行结果明确为提交，服务端会将半事务消息重新存储到普通存储系统中，此时消息对下游消费者可见，等待被消费者获取并消费；
• 消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，RocketMQ 会对消息进行重试处理。具体信息，请参见消息重试；
• 消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）；RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。
• 消息删除：当消息存储时长到期或存储空间不足时，RocketMQ 会按照滚动机制清理最早保存的消息数据，将消息从物理文件中删除。

事务消息基本流程

事务消息交互流程如下图所示：

1. 生产者将消息发送至 RocketMQ 服务端；
2. RocketMQ 服务端将消息持久化成功之后，向生产者返回 Ack 确认消息已经发送成功，此时消息被标记为“暂不能投递”，这种状态下的消息即为半事务消息；
3. 生产者开始执行本地事务逻辑；
4. 生产者根据本地事务执行结果向服务端提交二次确认结果（Commit 或是 Rollback），服务端收到确认结果后处理逻辑如下：

• 二次确认结果为 Commit：服务端将半事务消息标记为可投递，并投递给消费者；
• 二次确认结果为 Rollback：服务端将回滚事务，不会将半事务消息投递给消费者。

• 在断网或者是生产者应用重启的特殊情况下，若服务端未收到发送者提交的二次确认结果，或服务端收到的二次确认结果为Unknown未知状态，经过固定时间后，服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查；
• 生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果；
• 生产者根据检查到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤 4 对半事务消息进行处理。

实现细节：RocketMQ 事务消息如何实现

根据发送事务消息的基本流程的需要，实现分为三个主要流程：接收处理 Half 消息、Commit 或 Rollback 命令处理、事务消息 check。

处理 Half 消息

发送方第一阶段发送 Half 消息到 Broker 后，Broker 处理 Half 消息。Broker 流程参考下图：

具体流程是首先把消息转换 Topic 为 RMQ_SYS_TRANS_HALF_TOPIC，其余消息内容不变，写入 Half 队列。具体实现参考 SendMessageProcessor 的逻辑处理。

Commit 或 Rollback 命令处理

发送方完成本地事务后，继续发送 Commit 或 Rollback 到 Broker。由于当前事务已经完结，Broker 需要删除原有的 Half 消息，由于 RocketMQ 的 appendOnly 特性，Broker通过 OP 消息实现标记删除。Broker 流程参考下图：

• Commit。Broker 写入 OP 消息，OP 消息的 body 指定 Commit 消息的 queueOffset，标记之前 Half 消息已被删除；同时，Broker 读取原 Half 消息，把 Topic 还原，重新写入 CommitLog，消费者则可以拉取消费；
• Rollback。Broker 同样写入 OP 消息，流程和 Commit 一样。但后续不会读取和还原 Half 消息。这样消费者就不会消费到该消息。

具体实现在 EndTransactionProcessor 中。

事务消息 check

如果发送端事务时间执行过程，发送 UNKNOWN 命令，或者 Broker/发送端重启发布等原因，流程 2 的标记删除的 OP 消息可能会缺失，因此增加了事务消息 check 流程，该流程是在异步线程定期执行（transactionCheckInterval 默认 30s 间隔），针对这些缺失 OP 消息的 Half 消息进行 check 状态。具体参考下图：

事务消息 check 流程扫描当前的 OP 消息队列，读取已经被标记删除的 Half 消息的 queueOffset。如果发现某个 Half 消息没有 OP 消息对应标记，并且已经超时（transactionTimeOut 默认 6 秒），则读取该 Half 消息重新写入 half 队列，并且发送 check 命令到原发送方检查事务状态；如果没有超时，则会等待后读取 OP 消息队列，获取新的 OP 消息。

另外，为了避免发送方的异常导致长期无法确定事务状态，如果某个 Half 消息的 bornTime 超过最大保留时间（transactionCheckMaxTimeInMs 默认 12 小时），则会自动跳过此消息，不再 check。

具体实现参考：

TransactionalMessageServiceImpl#check 方法。

实战：使用事务消息

了解了 RocketMQ 事务消息的原理后，我们看下如何使用事务。首先，我们需要创建一个 “事务消息” 类型的 Topic，可以使用控制台或者 CLi 命令创建。

事务消息相比普通消息发送时需要修改以下几点：

• 发送事务消息前，需要开启事务并关联本地的事务执行。
• 为保证事务一致性，在构建生产者时，必须设置事务检查器和预绑定事务消息发送的主题列表，客户端内置的事务检查器会对绑定的事务主题做异常状态恢复。

当事务消息 commit 之后，这条消息其实就是一条投递到用户 Topic 的普通消息而已。所以对于消费者来说，和普通消息的消费没有区别。

注意：

1. 避免大量未决事务导致超时：在事务提交阶段异常的情况下发起事务回查，保证事务一致性；但生产者应该尽量避免本地事务返回未知结果；大量的事务检查会导致系统性能受损，容易导致事务处理延迟；
2. 事务消息的 Group ID 不能与其他类型消息的 Group ID 共用:与其他类型的消息不同，事务消息有回查机制，回查时服务端会根据 Group ID 去查询生产者客户端；
3. 事务超时机制：半事务消息被生产者发送服务端后，如果在指定时间内服务端无法确认提交或者回滚状态，则消息默认会被回滚。

原文链接

本文为阿里云原创内容，未经允许不得转载。

消息中间件rocketmq源码解析：事务消息

650)this.width=650;"src="http://www.yunai.me/images/common/wechat_mp.jpeg"style="border:2pxsolidrgb(238,238,238);margin-top:0px;"/>关注微信公众号：【芋艿的后端小屋】有福利：RocketMQ/MyCAT/Sharding-JDBC 所有源码分析文章列表Rock 查看详情

rocketmq源码分析4-事务消息实现原理

为什么消息要具备事务能力参见还是比较清晰的。简单的说就是在你业务逻辑过程中，需要发送一条消息给订阅消息的人，但是期望是此逻辑过程完全成功完成之后才能使订阅者收到消息。业务逻辑过程假设是这样的：逻辑部分... 查看详情

rocketmq事务消息原理(代码片段)

一、RocketMQ事务消息原理：        RocketMQ在4.3版本之后实现了完整的事务消息，基于MQ的分布式事务方案，本质上是对本地消息表的一个封装，整体流程与本地消息表一致，唯一不同的就是将本地消息表存... 查看详情

rocketmq事务消息原理及使用方法解析(代码片段)

...#x1f34a;Java学习：Java从入门到精通总结🍊深入浅出RocketMQ设计思想：深入浅出RocketMQ设计思想🍊绝对不一样的职场干货：大厂最佳实践经验指南📆最近更新：2023年3月24日🍊个人简介：通信工程... 查看详情

一文详解，rocketmq事务消息

在RocketMQ中有一个非常有用的功能，就是事务消息功能，事务消息机制，可以让我们确保发送的消息一定能写进MQ里，绝不会丢失掉。MQ事务消息机制还是挺有用的，在业内还是比较常见的，所以今天我们就... 查看详情

rocketmq事务消息实战(代码片段)

RocketMQ事务消息阅读目录指引：RocketMQ源码分析之从官方示例窥探RocketMQ事务消息实现基本思想RocketMQ源码分析之RocketMQ事务消息实现原理上篇RocketMQ源码分析之RocketMQ事务消息实现原理中篇----事务消息状态回查RocketMQ源码分析... 查看详情

30rocketmq事务消息的代码实现细节(代码片段)

...。1.发送half事务消息出去packagecom.mqTrsMessage;importorg.apache.rocketmq.client.producer.DefaultMQProducer;importorg.apache.rocketmq.clien 查看详情

七.rocketmq极简入门-rocketmq事务消息(代码片段)

概述如果业务只涉及到一个数据库的写操作，我们只需要保证这一个事物的提交和回滚，这种事务管理叫传统事物或本地事务，如果业务涉及到多个数据库(多个服务)的写操作，我们需要保证多个数据库同时提交... 查看详情

rabbitmq，rocketmq，kafka事务性，消息丢失和重复发送处理策略

参考技术A我们的服务器从单机发展到拥有多台机器的分布式系统，各个系统之前需要借助于网络进行通信，原有单机中相对可靠的方法调用以及进程间通信方式已经没有办法使用，同时网络环境也是不稳定的，造成了我们多个... 查看详情

rocketmq事务消息入门介绍(代码片段)

说明周五的时候发了篇：Rocketmq4.3支持事务啦！！！，趁着周末的时候把相关内容看了下，下面的主要内容就是关于RocketMQ事务相关内容介绍了。说明：今天这篇仅仅是入门介绍，并没有涉及到很多细节，先把大概流程说明白，... 查看详情

rocketmq事务消息篇之事务消息源码分析(代码片段)

前言RocketMQ事务消息篇（一）之事务消息的介绍RocketMQ事务消息篇（二）之事务消息的使用本文继前两篇对事务消息源码进行分析。事务消息处理基本流程在介绍事务消息的时候，画了一个简单的流程图说明事... 查看详情

rocketmq（09）——发送事务消息(代码片段)

发送事务消息RocketMQ支持发送事务消息，它的事务消息是基于二阶段提交机制实现的。当发送的消息是事务消息时，只有对应的消息被提交了才能被消费者进行消费。发送事务消息时生产者需要使用TransactionMQProducer，它还需要指... 查看详情

28rocketmq：事务消息机制的底层实现原理

1.half消息在commit前对消费者不可见的原因回顾：当订单系统去发送一个half消息给MQ的时候。对于这个half消息，红包系统在这时是看不到它的，也无法去消费这条消息并进行处理。ConsumeQueue的可见原理场景解析：当... 查看详情

rocketmq事务消息篇之事务消息的使用(代码片段)

前言在RocketMQ事务消息篇（一）之事务消息的介绍里对RocketMQ的事务消息作了相关说明，本文提供一些基本的开发示例。java示例依赖<dependencies><dependency><groupId>org.apache.rocketmq</groupId><artifactId>ro 查看详情

rocketmq和kafka到底选哪个

参考技术A1、适用场景kafka适合日志处理rocketmq适合业务处理结论：两者没有区别，根据具体业务定夺2、性能kafka单机写入TPS号称在百万条/秒rocketmq大约在10万条/秒结论：追求性能方面，kafka单机性能更高3、可靠性kafka使用异步刷... 查看详情

分布式事务解决方案|seata|本地消息表|事务消息|最大努力通知|消息丢失重复消费堆积有序

1.解决方案：1.1.Seata:1.2.本地消息表方案:1.3.RocketMQ事务消息方案:1.4.最大努力通知方案:1.5.小结:2.RocketMQ相关问题：2.1.RocketMq消息丢失怎么办:2.2.RocketMq消息重复消费怎么办:2.3.RocketMq消息堆积怎么办：2.4.RocketMq保证消息... 查看详情

rocketmq实现事务消息方案(代码片段)

RocketMQ是一个来自阿里巴巴的分布式消息中间件，于2012年开源，并在2017年正式成为Apache顶级项目。据了解，包括阿里云上的消息产品以及收购的子公司在内，阿里集团的消息产品全线都运行在RocketMQ之上，并且最近几年的双十一... 查看详情

rocketmq（09）——发送事务消息(代码片段)

发送事务消息RocketMQ支持发送事务消息，它的事务消息是基于二阶段提交机制实现的。当发送的消息是事务消息时，只有对应的消息被提交了才能被消费者进行消费。发送事务消息时生产者需要使用TransactionMQProducer，... 查看详情