springboot手动提交事务_SpringCloud异步场景分布式事务怎样做？试试RocketMQ

作者简介&＃xff1a;

陶陶老师

10年后端工作经验&＃xff0c;

专注Java、SpringBoot、SpringCloud、分布式系统/微服务、中间件等领域。

掘金专栏&＃xff1a;zlt2000 的个人主页 - 掘金

一、背景

在微服务架构中&＃xff0c;我们常常使用异步化的手段来提升系统的 吞吐量 和 解耦 上下游&＃xff0c;而构建异步架构最常用的手段就是使用 消息队列(MQ)&＃xff0c;那异步架构怎样才能实现数据一致性呢&＃xff1f;本文主要介绍如何使用RocketMQ的事务消息来解决一致性问题。

RocketMQ 是阿里巴巴开源的分布式消息中间件&＃xff0c;目前已成为 Apache 的顶级项目。历经多次天猫双十一海量消息考验&＃xff0c;具有高性能、低延时和高可靠等特性

PS&＃xff1a;同步场景怎样保证一致性&＃xff1f;请看文章《Spring Cloud同步场景分布式事务怎样做&＃xff1f;试试Seata》

二、MQ选型

可以看到在 业务处理 方面来说 RocketMQ 优于其他对手&＃xff0c;而且原生支持 事务消息

三、什么是事务消息

例如下图的场景&＃xff1a;生成订单记录 -> MQ -> 增加积分

我们是应该先 创建订单记录&＃xff0c;还是先 发送MQ消息 呢&＃xff1f;

1. 先发送MQ消息&＃xff1a;这个明显是不行的&＃xff0c;因为如果消息发送成功&＃xff0c;而订单创建失败的话是没办法把消息收回来的
2. 先创建订单记录&＃xff1a;如果订单创建成功后MQ消息发送失败 抛出异常&＃xff0c;因为两个操作都在本地事务中所以订单数据是可以 回滚 的

上面的 方式二 看似没问题&＃xff0c;但是 网络是不可靠的&＃xff01;如果 MQ 的响应因为网络原因没有收到&＃xff0c;所以在面对不确定的结果只好进行回滚&＃xff1b;但是 MQ 端又确实是收到了这条消息的&＃xff0c;只是回给客户端的 响应丢失 了&＃xff01;
所以 事务消息 就是用来保证 本地事务 与 MQ消息发送 的原子性&＃xff01;

四、RocketMQ事务消息原理

主要的逻辑分为两个流程&＃xff1a;

• 事务消息发送及提交&＃xff1a;

1. 发送 half消息
2. MQ服务端 响应消息写入结果
3. 根据发送结果执行 本地事务&＃xff08;如果写入失败&＃xff0c;此时half消息对业务 不可见&＃xff0c;本地逻辑不执行&＃xff09;
4. 根据本地事务状态执行 Commit 或者 Rollback&＃xff08;Commit操作生成消息索引&＃xff0c;消息对消费者 可见&＃xff09;

• 回查流程&＃xff1a;

1. 对于长时间没有 Commit/Rollback 的事务消息&＃xff08;pending 状态的消息&＃xff09;&＃xff0c;从服务端发起一次 回查
2. Producer 收到回查消息&＃xff0c;检查回查消息对应的 本地事务状态
3. 根据本地事务状态&＃xff0c;重新 Commit 或者 Rollback

逻辑时序图

五、异步架构一致性实现思路

从上面的原理可以发现 事务消息 仅仅只是保证本地事务和MQ消息发送形成整体的 原子性&＃xff0c;而投递到MQ服务器后&＃xff0c;并无法保证消费者一定能消费成功&＃xff01;
如果 消费端消费失败 后的处理方式&＃xff0c;建议是记录异常信息然后 人工处理&＃xff0c;并不建议回滚上游服务的数据(因为两者是 解耦 的&＃xff0c;而且 复杂度 太高)
我们可以利用 MQ 的两个特性 重试 和 死信队列 来协助消费端处理&＃xff1a;

1. 消费失败后进行一定次数的 重试
2. 重试后也失败的话该消息丢进 死信队列 里
3. 另外起一个线程监听消费 死信队列 里的消息&＃xff0c;记录日志并且预警&＃xff01;

因为有 重试 所以消费者需要实现幂等性

六、分布式事务场景样例

下面就用刚刚提到的场景&＃xff1a;生成订单记录 -> MQ -> 增加积分&＃xff1b;来简单讲一下 Spring Cloud 中应该怎么做&＃xff0c;详细代码请 下载demo 查看。
PS&＃xff1a;怎样安装部署RocketMQ可以参考《Apache RocketMQ 消息队列部署与可视化界面安装》

6.1. 引入依赖

使用 spring-cloud-stream 框架来访问 RocketMQ

Spring Cloud Stream 是一个构建消息驱动的框架&＃xff0c;通过抽象的定义实现应用与MQ消息队列之间的解耦&＃xff0c;目前支持 RabbitMQ、kafka 和 RocketMQ

6.2. 开启事务消息

消息生产者需要添加 transactional: true 开启 事务消息

6.3. 订单服务发送half消息

因为开启了 事务消息 所以这里发送的是 half消息 对于消费端是 不可见 的

6.4. 订单服务监听half消息

使用 &＃64;RocketMQTransactionListener 注解监听 半消息&＃xff0c;并实现 RocketMQLocalTransactionListener 接口&＃xff0c;该接口有两个方法

• executeLocalTransaction&＃xff1a;用于提交本地事务
• checkLocalTransaction&＃xff1a;用于事务回查

如果提交事务消息失败&＃xff0c;需等待约1分钟左右 事务回查 方法才会被调用

6.5. 积分服务消费消息

注意&＃xff1a;因为有重试 这里如果是真实的业务需要自行实现 幂等性

6.6. 消费死信队列预警

监听并消费死信队列中的消息&＃xff0c;用于记录错误日志&＃xff0c;并且预警通知运维人员等

6.7. 测试用例

demo中提供了3个接口分别测试不同的场景&＃xff1a;

• 事务成功
  http://localhost:11002/success
  流程如下&＃xff1a;

1. 订单创建 成功
2. 提交事务消息 成功
3. 消费消息增加积分 成功

• 订单创建成功但提交事务消息失败
  http://localhost:11002/produceError
  流程如下&＃xff1a;

1. 订单创建 成功
2. 提交事务消息 失败
3. 事务回查(等待1分钟左右) 成功
4. 提交事务消息 成功
5. 消费消息增加积分 成功

• 消费消息失败
  http://localhost:11002/consumeError
  流程如下&＃xff1a;

1. 订单创建 成功
2. 提交事务消息 成功
3. 消费消息增加积分 失败
4. 重试消费消息 失败
5. 进入死信队列 成功
6. 消费死信队列的消息 成功
7. 记录日志并发出预警 成功

七、demo下载地址

https://gitee.com/zlt2000/microservices-platform/tree/master/zlt-demo/rocketmq-demo/rocketmq-transactional

作者简介&＃xff1a;

陶陶老师

10年后端工作经验&＃xff0c;

专注Java、SpringBoot、SpringCloud、分布式系统/微服务、中间件等领域。

掘金专栏&＃xff1a;zlt2000 的个人主页 - 掘金

本文已经获得陶陶老师授权转发&＃xff0c;其他人若有兴趣转载&＃xff0c;请直接联系作者授权。

更多的学习资料&＃xff0c;请看这里&＃xff1a;

http://www.jnshu.com/login/1/36856070?source&＃61;zhihu-article-taotao