【RocketMQ】第二篇-RocketMQ解决分布式事务

深碍√TFBOYSˉ_ 2023-10-09 14:59 5阅读 0赞

#### RocketMQ ####

*  一、RocketMQ的顺序消息
 *  二、RocketMQ的单向发送（直接发送）
 *  三、RocketMQ的事务消息
 *   *  1.事务消息发送及提交：
     *  2.补偿流程：
 *  四、分布式事务的解决方案
 *   *  1、X/Open DTP分布式事务处理参考模型
     *  2、分布式事务的实现主要有以下 5 种方案：
     *   *  1.XA方案
         *  2.TCC 补偿方案
         *  3.本地消息异步确保
         *  4.可靠消息最终一致性方案（国内互联网）
         *  5.最大努力通知方案（也是保证最终一致性）
         *  分布式事务总结
 *  四、RocketMQ解决分布式事务
 *  1、RocketMQ的事务消息：
 *   *  2、举例：减库存、下订单业务场景
     *   *  1、常规操作
         *  2、RocketMQ把消息分为两个阶段：\*\*预备阶段和确认阶段；\*\*
         *  3、\*\*回到减库存下订单案例中：\*\*
         *  4、\*\*异常情况分析：\*\*
         *  5、回查判断业务是否成功的设计
         *  6、消费端的设计

## 一、RocketMQ的顺序消息 ##

顺序消息是指可以按照消息的发送顺序来消费消息 (FIFO)；  
RocketMQ可以严格的保证消息有序，可以分为分区有序 和 全局有序；  
默认情况下消息发送会采取Round Robin轮询方式把消息发送到不同的`queue(分区队列)`；而消费消息的时候从多个queue上拉取消息，这种情况下发送和消费不能保证顺序；`但是如果控制发送的消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了消息的顺序`；  
`另外当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的；`

比如有一个秒杀订单的顺序消费：下单成功发送消息、支付成功发送消息；  
要保证这两个消息是顺序地消费，先接收下单消息，再接收支付消息，则可以将订单号相同的消息发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列，同一个队列的消息就是有顺序的；

## 二、RocketMQ的单向发送（直接发送） ##

这种方式主要用在不特别关心发送结果的场景，例如日志发送；

void sendOneway(...)

此方法返回前不会等待broker的确认，很显然，它具有最大的吞吐量，但有消息丢失的可能性；  
通常消息的发送是这样一个过程：  
.客户端发送请求到服务器  
.服务器处理请求  
.服务器向客户端返回应答  
所以，一次消息发送的耗时时间是上述三个步骤的总和，而某些场景要求耗时非常短，但是对可靠性要求并不高，例如日志收集类应用，此类应用可以采用oneway形式调用，oneway形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个操作系统系统调用的开销，即将数据写入客户端的socket缓冲区，此过程耗时通常在微秒级；

## 三、RocketMQ的事务消息 ##

Apache RocketMQ在4.3.0版中已经支持分布式事务消息，这里RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示：  
![在这里插入图片描述][f275431e2c264a7b9c6ee4a767ae0797.png]  
上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程；

### 1.事务消息发送及提交： ###

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

### 2.补偿流程： ###

(1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”；  
(2) Producer收到回查消息，检查回查消息对应的本地事务的状态；  
(3) 根据本地事务状态，重新Commit或者Rollback；  
其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况；  
事务消息一共有三种状态：提交状态、回滚状态、中间状态；  
TransactionStatus.CommitTransaction: 提交事务，代表消费者可以消费此消息；  
TransactionStatus.RollbackTransaction: 回滚事务，代表消息将被删除，不能被消费；  
TransactionStatus.Unknown: 中间状态，代表需要检查消息队列来确定状态；

## 四、分布式事务的解决方案 ##

事务是数据库的概念，数据库事务（ACID）；  
分布式事务的产生，是由于数据库的拆分和分布式架构(微服务)带来的，在常规情况下，我们在一个进程中操作一个数据库，这属于本地事务，如果在一个进程中操作多个数据库，或者在多个进程中操作一个或多个数据库，就产生了分布式事务；  
（1）数据库分库分表就产生了分布式事务；  
![在这里插入图片描述][1450529ede86464ba8e599bfc906ed09.png]

（2）项目拆分服务化也产生了分布式事务；  
![在这里插入图片描述][6aa810208beb442c99dc0f3d25233801.png]

### 1、X/Open DTP分布式事务处理参考模型 ###

全称X/Open Distributed Transaction Processing Reference Model，即分布式事务处理参考模型；  
X/Open是一个组织机构，它定义出的一套分布式事务处理的标准和规范，具体实现由不同厂商来实现；  
Java EE 遵循了X/open DTP规范，设计并实现了Java里面的分布式事务编程接口规范JTA；  
XA是X/Open DTP定义的中间件与数据库之间的接口规范，主流的数据库厂商都实现了XA接口函数规范；

### 2、分布式事务的实现主要有以下 5 种方案： ###

（1）XA 方案 （一般数据库实现了）  
（2）TCC 方案  
（3）本地消息异步确认 （不是标准也不是规范，是实践中总结出来的）  
（4）可靠消息最终一致性 （不是标准也不是规范，是实践中总结出来的）  
（5）最大努力通知 （不是标准也不是规范，是实践中总结出来的）

#### 1.XA方案 ####

XA方案即两阶段提交（Two Phase Commit），有一个事务管理器（协调者），负责协调多个数据库（资源管理器）的事务，事务管理器先询问各个数据库是否已经准备好？如果每个数据库都已就绪，那么就正式提交事务，在各个数据库上执行操作，如果任何其中一个数据库未就绪，那么就回滚事务；  
这种分布式事务方案，比较适合单体应用中跨多个数据库的分布式事务，依赖于数据库层面来实现分布式事务，效率较低，不适合高并发的场景；

在分布式微服务中，一个大的系统被拆分为几十个甚至几百个服务，标准规范的做法是每个服务只操作自己对应的一个数据库，如果要操作别的服务所对应的数据库，不允许直接连接而是通过调用别人的服务接口来实现，不能交叉去访问别人的数据库，避免一些混乱的问题，比如数据被别人修改出错，或者数据库被别的服务压垮等情况；  
XA的实现方案：  
（1）在Java中具体实现可以采用 Spring + JTA 解决；  
（2）Atomikos https://www.atomikos.com  
（3）Bitronix（btm） https://github.com/bitronix  
（4）JOTM http://jotm.objectweb.org 比较老了，可以不考虑了

#### 2.TCC 补偿方案 ####

TCC 的全称是：Try、Confirm、Cancel，这个也就是我们经常听到的三阶段提交；  
\*\*Try 阶段：\*\*尝试执行，对各个服务的资源做检测以及对资源进行锁定或者预留；  
\*\*Confirm 阶段：\*\*确认执行真正执行业务，在各个服务中执行实际的操作；  
\*\*Cancel 阶段：\*\*如果任何一个服务的业务方法执行异常，那么就进行补偿，把已经执行成功的业务操作进行回滚；  
该方案一般用于与支付交易相关的场景，用TCC严格保证分布式事务要么全部成功，要么全部自动回滚，严格保证资金交易的正确性，其他场景下一般很少采用该方案，因为该方案的事务回滚是依赖于我们自己编写的业务代码来回滚和补偿，对业务代码的侵入性比较强，增加了代码的复杂度，另外最好是各个业务执行的时间都比较短；  
（一般与钱打交道的，可以采用该方案，像支付宝内部使用了类似的方案）  
开源的tcc实现框架  
tcc-transaction [https://github.com/changmingxie/tcc-transaction][https_github.com_changmingxie_tcc-transaction]  
Bytetcc [https://github.com/liuyangming/ByteTCC （https://www.bytesoft.org）][https_github.com_liuyangming_ByteTCC _https_www.bytesoft.org]

#### 3.本地消息异步确保 ####

本地消息异步确保是 ebay 提出来的一套分布式事务解决方案；  
具体实现方案是：

*  A系统在自己本地一个事务里操作时，先插入一条数据到本地消息表，然后A系统将这个消息发送到消息中间件MQ中；
 *  B系统接收到消息之后，在一个事务里，往自己本地消息表里插入一条数据，同时执行其他的业务操作，如果这个消息已经被处理过了，那么此时这个事务会回滚，这样保证不会重复处理消息；
 *  B系统执行成功之后，就会更新自己本地消息表的状态以及 A 系统消息表的状态（通过zookeeper或者调用a系统的接口来更新A系统的消息表状态）；
 *  如果 B 系统处理失败了，那么就不会更新消息表状态，那么此时 A 系统起一个后台线程定时扫描自己的本地消息表，如果有未处理的消息，会再次发送消息到 MQ 中，让 B 再次处理；

这个方案保证了**最终一致性**，哪怕 B 事务失败了，但是 A 会不断重发消息，直到 B 那边成功为止；

这个方案的不足就在于严重依赖于数据库的消息表来管理事务，数据库的性能是大的瓶颈，所以在高并发场景下，该方案不被采用，在国内我们对该方案进行了升级，就是下面的可靠消息最终一致性方案；  
![在这里插入图片描述][d36e5fc07e0e4c6394bec2f3105a67bb.png]

#### 4.可靠消息最终一致性方案（国内互联网） ####

该方案是本地消息异步确保方案的升级和优化，该方案中去掉了本地消息表；  
![在这里插入图片描述][7390197efcb14476ab7a5ddf15375b1e.png]  
该方案是ebay改进版，可以适用高并发场景；（用得比较多）

#### 5.最大努力通知方案（也是保证最终一致性） ####

该方案的具体实现是：  
系统 A 本地事务执行完之后，发送个消息到 MQ；  
同时有一个专门消费 MQ 最大努力通知服务，该服务会消费 MQ 消息然后写入数据库中，或者是放入缓存中，然后调用系统 B 的接口；  
要是系统 B 执行成功就执行完毕，如果系统 B 执行失败，那么最大努力通知服务将定时尝试重新调用系统 B，反复多次，尝试多次后如果依然失败则放弃；  
（这时候 可以监控、业务后台处理，人工干预等）  
![在这里插入图片描述][cf72add9805b44858d359a4d9703c5a5.png]  
比如在支付业务，通知支付结果通常采用该方案；  
另外可以根据不同的业务定制不同的通知规则，比如通知支付结果等相对严谨的业务，可以将通知频率设置高一些，通知时间长一些，比如隔 5 分钟通知一次，持续时间 1 小时；  
如果不重要的业务，比如通知用户增加积分，则可以将通知频率设置低一些，时间短一些，比如 10 分钟通知一次，持续 30 分钟。

#### 分布式事务总结 ####

互联网的分布式事务解决方案  
目前互联网领域里的几种流行的分布式解决方案，都没有像之前所说的XA事务一样形成X/OpenDTP那样的工业规范，而是仅仅在具体的行业里获得较多的认可；  
1、业务接口整合，避免分布式事务；  
这个就是把一个业务流程中需要在一个事务里执行的多个相关业务接口包装整合到一个事务中，比如我们可以将A/B/C整合为一个服务D来实现单一事务业务服务；  
2、ebay 模式或升级版；  
3、最大努力通知方案；  
4、开源中间件方案；

阿里 Seata（曾用名Fescar）是阿里的开源分布式事务框架；  
Seata大家可以参考我的另外一篇博客 [【分布式事务】分布式事务Seata][Seata]

另外几款开源分布式事务项目：  
tcc-transaction https://github.com/changmingxie/tcc-transaction  
Hmily https://github.com/yu199195/hmily  
ByteTCC https://github.com/liuyangming/ByteTCC  
myth https://github.com/yu199195/myth  
EasyTransaction https://github.com/QNJR-GROUP/EasyTransaction  
tx-lcn https://github.com/codingapi/tx-lcn/  
TCC强一致性：**刚性事务**  
最终一致性：**柔性事务**

## 四、RocketMQ解决分布式事务 ##

分布式事务是一个复杂的技术问题，没有一统天下的通用解决方案，如果我们的系统不追求强一致性，那么最常用的分布式事务解决方案就是最终一致性方案，RocketMQ提供的事务消息特性可以实现消息最终一致性的分布式事务解决方案；

## 1、RocketMQ的事务消息： ##

![在这里插入图片描述][08610a566ada46cbafc52dfda120f958.png]

（1）Half Message，半消息  
暂时不能被Consumer消费的消息，Producer已经把消息发送到Broker端，但是此消息的状态被标记为不能投递，处于这种状态下的消息称为半消息，事实上，该状态下的消息会被放在一个叫做 RMQ\_SYS\_TRANS\_HALF\_TOPIC的主题下；  
当 Producer端对它二次确认后，也就是Commit之后， Consumer端才可以消费到，那么如果是Rollback，该消息则会被删除，永远不会被消费到；  
（2）事务状态回查  
可能因为网络原因、应用问题等导致Producer端一直没有对这个半消息进行确认，那么此时Broker服务器会定时扫描这些半消息，主动回调Producer端查询该消息的状态；

**总之RocketMQ事务消息的实现原理就是基于两阶段提交和事务状态回查，来决定消息最终是提交还是回滚；**

### 2、举例：减库存、下订单业务场景 ###

![在这里插入图片描述][a0f38d7be7ee4e53950c3667a1689190.png]

#### 1、常规操作 ####

**场景一：先减库存后发消息**  
先减库存再发消息，万一发送消息失败了，那用户就没法下单；

**场景二：先发消息，后减库存**  
消息发送成功，但用户减库存失败，下单业务订阅到了消息，导致用户下了单；  
总之都会出现两边数据不一致的问题，怎么解决？

#### 2、RocketMQ把消息分为两个阶段：**预备阶段和确认阶段；** ####

*  **Prepared阶段（预备阶段）**  
    该阶段主要发一个消息到rocketmq，但该消息只储存在commitlog中，但consumeQueue中不可见，也就是消费端（订阅端）无法看到此消息；
 *  **commit/rollback阶段（确认阶段）**  
    该阶段主要是把prepared消息保存到consumeQueue中，即让消费端可以看到此消息，也就是可以消费此消息；

#### 3、**回到减库存下订单案例中：** ####

1、在减库存之前，先发送预备消息；  
2、发送预备消息成功后，执行本地减库存事务；  
3、减库存成功后，再发送确认消息；  
4、消费端（下订单业务）可以看到确认消息，消费此消息，进行下订单；

上面的确认消息可以是commit消息，此时可以被订阅者消费；也可以是Rollback消息，即执行本地减库存事务失败后，提交rollback消息，即删除预备消息，订阅者无法消费；

#### 4、**异常情况分析：** ####

\*\*异常1：\*\*如果发送预备消息失败，下面的流程都不会执行，这是正常的不会有问题；  
\*\*异常2：\*\*如果发送预备消息成功，但执行本地事务失败，这也没有问题，因为此时预备消息不会被消费端订阅到，消费端不会执行业务；  
**异常3**：如果发送预备消息成功，执行本地事务成功，但发送确认消息失败，这就有问题了，因为用户减库存成功了，但下订单业务没有订阅到确认消息，无法下订单，导致两边数据不一致；  
对于异常3，发送预备消息成功，本地扣款事务成功，但发送确认消息失败；RocketMQ的解决方案是进行回查本地事务状态，在回查后发现业务已经减库存成功，那么就补发“发送commit确认消息”，从而下订单业务就可以订阅此消息；

#### 5、回查判断业务是否成功的设计 ####

设计一张`TransactionLog`表，将业务表的处理和`TransactionLog`表的处理放在同一个本地事务中，如果本地减库存事务成功，`TransactionLog`表中应该也成功记录了TransactionId,当RocketMQ回查时，只需检查对应的`TransactionId`是否存在即可；

#### 6、消费端的设计 ####

消费端（下订单业务）需要考虑幂等性设计，消费消息失败要重试几次；  
幂等性设计一般可以采用数据库唯一约束或者分布式锁解决，  
消费重试RocketMQ本身就支持，默认重试16次，可以修改

[f275431e2c264a7b9c6ee4a767ae0797.png]: https://img-blog.csdnimg.cn/f275431e2c264a7b9c6ee4a767ae0797.png
[1450529ede86464ba8e599bfc906ed09.png]: https://img-blog.csdnimg.cn/1450529ede86464ba8e599bfc906ed09.png
[6aa810208beb442c99dc0f3d25233801.png]: https://img-blog.csdnimg.cn/6aa810208beb442c99dc0f3d25233801.png
[https_github.com_changmingxie_tcc-transaction]: https://github.com/changmingxie/tcc-transaction
[https_github.com_liuyangming_ByteTCC _https_www.bytesoft.org]: https://github.com/liuyangming/ByteTCC%20%EF%BC%88https://www.bytesoft.org%EF%BC%89
[d36e5fc07e0e4c6394bec2f3105a67bb.png]: https://img-blog.csdnimg.cn/d36e5fc07e0e4c6394bec2f3105a67bb.png
[7390197efcb14476ab7a5ddf15375b1e.png]: https://img-blog.csdnimg.cn/7390197efcb14476ab7a5ddf15375b1e.png
[cf72add9805b44858d359a4d9703c5a5.png]: https://img-blog.csdnimg.cn/cf72add9805b44858d359a4d9703c5a5.png
[Seata]: https://blog.csdn.net/weixin_43333483/article/details/122908892?spm=1001.2014.3001.5502
[08610a566ada46cbafc52dfda120f958.png]: https://img-blog.csdnimg.cn/08610a566ada46cbafc52dfda120f958.png
[a0f38d7be7ee4e53950c3667a1689190.png]: https://img-blog.csdnimg.cn/a0f38d7be7ee4e53950c3667a1689190.png