正文

构建基于rocketmq的分布式事务服务(代码片段)

author  author  2023-01-03  738

关键词：

说在前面

Apache RocketMQ-4.3.0正式Release了事务消息的特性，顺着最近的这个热点。第一篇文章，就来聊一下在软件工程学上的长久的难题——分布式事务（Distributed Transaction）。

这个技术也在各个诸如阿里，腾讯等大厂的内部，被广泛地实现，利用及优化。但是由于理论上就有难点，所以分布式事务就隐晦得成了大厂对于小厂的技术壁垒。相信来看这篇文章的同学，一定都听过很多关于分布式事务的术语，比较二阶段提交，TCC，最终一致性等，所以这里也不多普及概念。

基于RocketMQ的分布式事务

我们直接上正题，利用RocketMQ设计自己的分布式事务组件。

举个虚拟场景引出问题

用户从农行转账100元去招行?，农行的系统和招行的系统分别部署在自己的机房，系统之间通过消息进行通信，防止过度耦合。

整个模型可以不恰当得描述为：农行扣了100元后，发送“已经扣款”的消息给招行，招行收到消息，知道农行扣款成功了，然后在招行账户上加100元。

问题是，农行这边，方案1. 先扣100元再发消息，方案2. 先发消息再扣100元

整理下整个事务不一致的场景：

方案1，

农行扣100后成功，但是消息发送失败，招行没有加100

方案2，

消息发送成功，但是农行扣100元失败，招行收到消息加了100

各位同学应该已经发现问题所在了，扣款和发送消息这两个事情，没有办法通过调换顺序实现「同时成功」，或者「同时失败」。如果前者成功，后者失败，就会造成不一致。

RocketMQ，以下简称RMQ，为了实现事务消息引入了一种新的消息类型：TransactionMsg

一个完整的事务消息分成两个部分：

HalfMsg(Prepare)?+?Commit/RollbackMsg

Producer发送了HalfMsg后，由于HalfMsg不是一个完整的事务消息，Consumer无法立刻就消费到该消息，Producer可以对HalfMsg进行Commit或者Rollback来终结事务（EndTransacaction）。只有当Commit了HalfMsg后，Consumer才能消费到这条消息。RMQ会定期去向Producer询问，是否可以Commit或者Rollback那些由于错误没有被终结的HalfMsg来结束它们的生命周期，以达成事务最终的一致。

依然是刚刚的转账场景，我们用RMQ事务消息来优化下流程：

农行向RMQ同步发送HalfMsg，消息中携带农行即将要扣100元的信息
农行HalfMsg成功发送后，执行数据库本地事务，在自己的系统中扣100元
农行查看本地事务执行情况
本地事务返回成功，农行向RMQ提交（Commit）HalfMsg
招行系统订阅了RMQ，顺利收到农行已经扣款100元的信息
招行系统执行本地事务，在招行的系统中加100元

技术分享图片

图1：RMQ事务消息原理

同样得，我们逐个来分析下这个流程是不是会出现不一致：

农行发送HalfMsg是同步发送（Sync），如果HalfMsg发送不成功，压根就不会执行本地事务
发送HalfMsg成功，但是农行扣款**本地事务失败，也没事，如果本地事务没有成功，立刻就发送Rollback去回滚HalfMsg**。就当之前啥事都没有发生过
农行本地事务成功了，但是Commit却失败了，但是由于HalfMsg已经在RMQ中，RMQ就能通过定时程序让农行重新检测本地事务是否成功，重新Commit。Rollback失败了也是同理
招行消费了消息后，加钱本地事务失败了，但是招行收到的消息持久化在MQ，甚至可以持久化在招行数据库，可以进行事务重试

刚刚讨论的案例是非常理想化的，整个分布式事务中，只涉及到了金额的变化，但是，真正的线上系统，作为消息发送方的本地事务可能就非常复杂，可能涉及到了几十张不同的表，那RMQ用定时器来Check HalfMsg，难道去查下涉及该事务的每一张表的数据是否提交成功？显然这种方案非常业务侵入非常大，并且很难组件化。所以需要在本地事务中设计一张Transaction表，将业务表和Transaction绑定在同一个本地事务中，如果农行的扣款本地事务成功时，Transaction中应当已经记录该TransactionId的状态为「已完成」。当最后需要检查时，只需要检查对应的TransactionId的状态是否是「已完成」就好，而不用关心具体的业务数据。

再谈一个小细节，

细心的同学可能发现，刚刚No.3的讨论其实是有点不严谨的，RMQ在调用Commit或者Rollback时，用的是Oneway的方式，熟悉RMQ源码的话，知道这种网络调用是只单向发送Request，不会去获取Response。消息发送性能上是有非常大的提升的，但是如果真的发送失败，Producer是不会知晓的，最后只能通过定时检查HalfMsg才能终结事务。

public void endTransactionOneway(
        final String addr,
        final EndTransactionRequestHeader requestHeader,
        final String remark,
        final long timeoutMillis
    ) throws RemotingException, MQBrokerException, InterruptedException 
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, requestHeader);

        request.setRemark(remark);
        // 使用Oneway发送end transaction类型的
        this.remotingClient.invokeOneway(addr, request, timeoutMillis);

脱离RocketMQ的分布式事务

不是所有的MQ都能支持事务消息，如何使用一般的MQ来搭建分布式事务组件，甚至抽象成一个事务SOA服务？

其实仔细分析下RMQ的事务消息，我们可以把它拆解成两个部分：

事务管理器?+?消息

所谓的事务管理器，就是对于事务的预备（Prepare）、提交（Commit）和回滚（Rollback）的管理，另外还包含预备事务的定时检查器。

消息，指的就是一般的同步消息，发送后能明确得到发送结果，用于事务系统与业务系统解耦。几乎所有的分布式MQ都是支持这种消息的。

我们来设计下自己的DistributedTransaction SOA，以下简称DT-SOA

技术分享图片

图2：分布式事务服务化

流程还是没有变，但分布式事务不再强依赖RMQ，而是用一般的MQ代替：

系统A发送事务，首先调用DT-SOA的Prepare方法准备开启事务，由于是同步调用，获取SendResult，如果发送成功，拿到全局分布式事务的ID——TID
系统A用获取到的TID执行本地事务，本地事务中包含Transaction状态表，成功后将TID对应的状态置为“已完成”
系统A调用DT-SOA提交事务，DT-SOA用MQ发送同步消息给系统B
系统B监听对应Topic，接收到消息后，执行对应的本地事务

说在后面

更多精彩的文章，请关注我的微信公众号: 艾瑞克的技术江湖
技术分享图片

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

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

rocketmq的分布式事务机制（事务消息）(代码片段)

...tMQ的事务消息可以用于实现基于可靠消息的最终一致性的分布式事务。文章目录1事务消息简要流程2一阶段半消息不可见的设计3二阶段Commit和Rollback操作4Op消息的设计5Commit消息变得可见6消息回查7最终一致性分布式事务常用于保... 查看详情

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

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

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

...库同时提交或回滚，这种夸多个数据库的事务操作叫分布式事务。分布式事物的解决方案有很多，如：2PC,TCC,最终一致性查看详情

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

...概流程说明白，后续再具体细节进行开篇说明。主题引出分布式事务相关内容。RocketMQ事务消息。RocketMQ事务消息如何使用。RocketMQ事务消息是怎么实现的。为什么需要事务消息会查查看详情

rocketmq事务消息机制(代码片段)

RocketMQ提供了事务消息，通过事务消息就能达到分布式事务的最终一致，从而实现了可靠消息服务。一、事务消息的实现步骤事务消息发送步骤：1.发送方将半事务消息发送至RocketMQ服务端。2.RocketMQ服务端将消息持久... 查看详情

rocketmq解决分布式事务(代码片段)

1.原理图： 2.设计实现思路：1.生产者（发送方）投递事务消息到Broker中，设置该消息为半消息不可以被消费；2.开始执行我们的本地事务，将本地事务执行的结果（回滚或者提交）发送给Broker；3.Broker获取回滚或者... 查看详情

11springboot整合rocketmq实现事务消息(代码片段)

...f0c;在4.x版本之后开源，可以利用事务消息轻松地实现分布式事务。RocketMQ在其消息定义的基础上，对事务消息扩展了两个相关的概念：Half(Prepare)Message——半消息(预处理消息)半消息是一种特殊的消息类型，该状态... 查看详情

rocketmq单机环境搭建(代码片段)

...，带来一篇搭建RocketMQ单机环境的文章，为后面的分布式事务专栏做准备。RocketMQ是阿里巴巴开源的一款高性能分布式消息中间件，有关RocketMQ的详细讲解，后面会单独开设一个RocketMQ专栏。这里，先简单介绍一... 查看详情

相较于rocketmq的事务消息，本地消息表才是真正的王者(代码片段)

1.概览在分布式系统中，系统间的通信除了大家所熟知的RPC外，基于MQ的异步通信也越来越流行，已经成为基础设施的重要组成部分。而MQ的引入对系统间的数据一致性提出了新的挑战，逐渐成为系统稳定性的一大... 查看详情

搞懂分布式技术19：使用rocketmq事务消息解决分布式事务

搞懂分布式技术19：使用RocketMQ事务消息解决分布式事务初步认识RocketMQ的核心模块rocketmq模块rocketmq-broker：接受生产者发来的消息并存储（通过调用rocketmq-store），消费者从这里取得消息。rocketmq-client：提供发送、接受消息的客... 查看详情

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

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

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

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

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

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

rocketmq使用(代码片段)

...试)2,有序消息.(秒杀,等需要有序的消费场景)3,事务消息.(分布式事务场景,需要mq回滚)rocketmq 消费类型常用:1,消费组(同一个组下面只消费一次,用于分布式集群场景)2,订阅模式消费.[consumer.setMessageModel(MessageModel.BROADCASTING);]... 查看详情

基于消息队列rocketmq的大型分布式应用上云实践(代码片段)

...a; ApacheRocketMQ作为阿里巴巴开源的支撑万亿级数据洪峰的分布式消息中间件，在众多行业广泛应用。在选型过程中，开发者一定会关注开源版与商业版的业务价值对比。那么，今天就围绕着商业版本的消息队列RocketMQ... 查看详情

rocketmq源码分析之rocketmq事务消息实现原理中篇----事务消息状态回查(代码片段)

上节已经梳理了RocketMQ发送事务消息的流程（基于二阶段提交），本节将继续深入学习事务状态消息回查，我们知道，第一次提交到消息服务器时消息的主题被替换为RMQ_SYS_TRANS_HALF_TOPIC，本地事务执行完后如果返回本地事务状态... 查看详情