SpringCloud基于RocketMQ实现分布式事务

曾经终败给现在 2023-09-28 17:52 25阅读 0赞

### 前言 ###

分布式事务是在微服务开发中经常会遇到的一个问题，之前的文章中我们已经实现了利用Seata来实现强一致性事务，其实还有一种广为人知的方案就是利用消息队列来实现分布式事务，保证数据的最终一致性，也就是我们常说的柔性事务。

#### 消息队列实现分布式事务原理 ####

首先让我们来看一下基于消息队列实现分布式事务的原理方案。

![97ceb0ef4c154e7094c4a045980a56be.png][]

柔性事务

发送消息的服务有个OUTBOX数据表，在进行INSERT、UPDATE、DELETE 业务操作时也会给OUTBOX数据表INSERT一条消息记录，这样可以保证原子性，因为这是基于本地的ACID事务。

OUTBOX表充当临时消息队列，然后我们在引入一个消息中继（MessageRelay）的服务，由他从OUTBOX表中读取数据并发布消息到消息组件。

消息中继的实现可以很简单，只需要通过定时任务定期从OUTBOX表中拉取最新未发布的数据，获取到数据后将数据发送给消息组件，最后将完成发送的消息从OUTBOX表中删除即可，对于失败的消息可以根据业务规则进行重试。

#### RocketMQ的事务消息 ####

RocketMQ本身已经支持事务消息，如果你们项目使用了RocketMQ，可以直接借助RocketMQ的事务消息实现分布式事务，我们先看一下RocketMQ事务消息的原理然后再借助RocketMQ来实现分布式事务。

RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示。

![ddb0d8f54dc7462a9c36b541484417ed.png][]

RocketMQ实现事务消息主要分为两个阶段：正常事务的发送及提交、事务信息的补偿流程

**整体流程为：**

*  正常事务发送与提交阶段  
    1、生产者发送一个半消息给MQServer（半消息是指消费者暂时不能消费的消息）  
    2、服务端响应消息写入结果，半消息发送成功  
    3、开始执行本地事务  
    4、根据本地事务的执行状态执行Commit或者Rollback操作
 *  事务信息的补偿流程  
    1、如果MQServer长时间没收到本地事务的执行状态会向生产者发起一个确认回查的操作请求  
    2、生产者收到确认回查请求后，检查本地事务的执行状态  
    3、根据检查后的结果执行Commit或者Rollback操作  
    补偿阶段主要是用于解决生产者在发送Commit或者Rollback操作时发生超时或失败的情况。

**RocketMQ事务流程关键**

1.  事务消息在一阶段对用户不可见  
    事务消息相对普通消息最大的特点就是一阶段发送的消息对用户是不可见的，也就是说消费者不能直接消费。这里RocketMQ的实现方法是原消息的主题与消息消费队列，然后把主题改成 `RMQ_SYS_TRANS_HALF_TOPIC` ，这样由于消费者没有订阅这个主题，所以不会被消费。
2.  如何处理第二阶段的失败消息？  
    在本地事务执行完成后会向MQServer发送Commit或Rollback操作，此时如果在发送消息的时候生产者出故障了，那么要保证这条消息最终被消费，MQServer会像服务端发送回查请求，确认本地事务的执行状态。  
    当然了rocketmq并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，RocketMQ默认回滚该消息。
3.  消息状态 事务消息有三种状态：**TransactionStatus.CommitTransaction**：提交事务消息，消费者可以消费此消息  
    **TransactionStatus.RollbackTransaction**：

[97ceb0ef4c154e7094c4a045980a56be.png]: https://img-blog.csdnimg.cn/97ceb0ef4c154e7094c4a045980a56be.png
[ddb0d8f54dc7462a9c36b541484417ed.png]: https://img-blog.csdnimg.cn/ddb0d8f54dc7462a9c36b541484417ed.png