一、前言
可靠消息最终一致性方案主要适用于消息数据能够独立存储：
能够降低系统之间耦合度
业务对数据一致性的时间敏感度高
此方案需要实现的服务模式：
可查询操作：提供查询自身事务状态的接口。
幂等操作：只要参数相同，无论调用多少次接口，都应该和第一次调用产生的结果相同。
那么什么时候回查？
事务发送端执行本地事务时（已经发送了 Half 消息了），这时候发送端宕机了或者超时了，就需要回查
（1）实现方案
实现方案有两种：
1.基于本地消息
绑定了具体的业务场景，耦合性太高，不可公用和扩展。
消息数据与业务数据在同一数据库，占用了业务系统的扩展。
消息数据可能会受到数据库并发性的影响。
2.基于消息队列中间件
消息数据能够独立存储，与具体的业务数据库解耦。
消息的并发性和吞吐量优于本地消息表方案。
发送一次消息需要完成两次网络交互：1.消息的发送 ; 2. 消息的提交或回滚。
需要实现消息的回查接口，增加了开发成本。
（2）注意的问题
1、事务发送方本地事务与消息发送的原子性问题：
// 原子性：事务 + 消息确认（回滚）@Override@Transactional(rollbackFor = Exception.class)public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object obj) { try{ TxMessage txMessage = this.getTxMessage(msg); // 1. 执行本地事务 orderService.submitOrderAndSaveTxNo(txMessage); // 2. 提交事务 return RocketMQLocalTransactionState.COMMIT; } catch (Exception e){ // 异常回滚事务 return RocketMQLocalTransactionState.ROLLBACK; }}
复制代码
2、事务参与方接收消息的可靠性问题：
3、事务参与方接收消息的幂等性问题：
另外还整理一些 C++后台开发架构师 相关学习资料，面试题，教学视频，以及学习路线图，免费分享有需要的可以自行添加：Q群：720209036 点击加入~ 群文件共享
（3）实战
通过 RocketMQ 消息中间件实现可靠消息最终一致性分布式事务，模拟电商业务中的下单扣减库存场景。 涉及服务有：
整体流程如下：
第一步：订单服务向 RocketMQ 发送 Half 消息。
第二步：RocketMQ 向订单服务响应 Half 消息发送成功。
第三步：订单服务执行本地事务，向本地数据库中插入、更新、删除数据。
第四步：订单服务向 RocketMQ 发送提交事务或者回滚事务的消息。
第五步：如果库存服务未收到消息，或者执行事务失败，且 RocketMQ 未删除保存的消息数据，RocketMQ 会回查订单服务的接口，查询事务状态，以此确认是再次提交事务还是回滚事务。
第六步：订单服务查询本地数据库，确认事务是否执行成功。
第七步：订单服务根据查询出的事务状态，向 RocketMQ 发送提交事务或者回滚事务的消息。
第八步：如果第七步中订单服务向 RocketMQ 发送的是提交事务的消息，则 RocketMQ 会向库存服务投递消息。
第九步：如果第七步中订单服务向 RocketMQ 发送的是回滚事务的消息，则 RocketMQ 不会向库存微服务投递消息，并且会删除内部存储的消息数据。
第十步：如果 RocketMQ 向库存服务投递的是执行本地事务的消息，则库存服务会执行本地事务，向本地数据库中插入、更新、删除数据。
第十一步：如果 RocketMQ 向库存服务投递的是查询本地事务状态的消息，则库存服务会查询本地数据库中事务的执行状态。
二、实战实验
涉及服务有：
实验准备：
MySQL：8.0.20
RocketMQ 消息中间件：rocketmq-all-4.5.0-bin-release
RocketMQ 客户端：rocketmq-spring-boot-starter 2.0.2
Spring Boot 版本：2.2.6.RELEASE
订单服务重点相关代码：
1.发送 Half 消息：
@Slf4j@Servicepublic class OrderServiceImpl implements OrderService { @Override public void submitOrder(Long productId, Integer payCount) { // 1. 生成全局分布式序列号 String txNo = UUID.randomUUID().toString(); 。。。 。。。 // 2. 封装消息 Message<String> message = MessageBuilder.withPayload(jsonObject.toJSONString()).build(); // 3. 发送一条事务消息 rocketMQTemplate.sendMessageInTransaction("tx_order_group", "topic_txmsg", message, null); }}
复制代码
2.处理本地事务
@Slf4j@Component@RocketMQTransactionListener(txProducerGroup = "tx_order_group")public class OrderTxMessageListener implements RocketMQLocalTransactionListener { @Autowired private OrderService orderService; @Autowired private OrderMapper orderMapper; @Override @Transactional(rollbackFor = Exception.class) public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object obj) { try{ // 1. 获取消息并解析消息 TxMessage txMessage = this.getTxMessage(msg); // 2. 提交订单 并且 保存事务日志 orderService.submitOrderAndSaveTxNo(txMessage); // 3. 事务状态为提交 return RocketMQLocalTransactionState.COMMIT; }catch (Exception e){ // 发生异常 // 事务状态为回滚 return RocketMQLocalTransactionState.ROLLBACK; } } @Override public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { // 1. 获取消息并解析消息 TxMessage txMessage = this.getTxMessage(msg); // 2. 查询订单是否存在 Integer exists = orderMapper.isExistsTx(txMessage.getTxNo()); if(exists != null){ // 订单存在：事务状态为提交 return RocketMQLocalTransactionState.COMMIT; } // 订单不存在：事务状态为未知 // 这里需要再次调用：处理本地事务嘛？ return RocketMQLocalTransactionState.UNKNOWN; }}
复制代码
库存服务重点相关代码：
@Slf4j@Component@RocketMQMessageListener(consumerGroup = "tx_stock_group", topic = "topic_txmsg")public class StockTxMessageConsumer implements RocketMQListener<String> { @Autowired private StockService stockService; @Override public void onMessage(String message) { // 监听到对应消息 // 获取消息并解析 TxMessage txMessage = this.getTxMessage(message); stockService.decreaseStock(txMessage); }}
复制代码
服务：
订单服务端口：8080
库存服务端口：8081
RocketMQ: 9876
数据准备：
USE tx_msg_stock;INSERT INTO stock (id, product_id, total_count) VALUES (1, 1001, 10000);SELECT * FROM stock;+----|------------|-------------+| id | product_id | total_count |+----|------------|-------------+| 1 | 1001 | 10000 |+----|------------|-------------+
复制代码
（1）正常流程
1.请求下单接口：调用 订单服务
$ curl "http://localhost:8080/order/submit_order?productId=1&payCount=1"下单成功
复制代码
订单服务日志：
2022-05-09 14:19:05.197 c.d.t.message.OrderTxMessageListener : 订单微服务执行本地事务2022-05-09 14:19:05.233 c.d.t.message.OrderTxMessageListener : 订单微服务提交事务2022-05-09 14:21:05.090 INFO 19423 --- [lientSelector_1] RocketmqRemoting : closeChannel: close the connection to remote address[172.17.0.3:10909] result: true
复制代码
2.库存服务日志：
2022-05-09 14:22:59.956 [MessageThread_5] c.d.t.message.StockTxMessageConsumer : 库存微服务开始消费事务消息:{"txMessage":{"payCount":1,"productId":1,"txNo":"3fcd4e8d-1f5b-448d-ad3d-693c335f994e"}}2022-05-09 14:22:59.956 [MessageThread_5] c.d.t.service.impl.StockServiceImpl : 库存微服务执行本地事务,商品id:1, 购买数量:1
复制代码
-- order 库下SELECT * FROM `order`;+---------------|---------------------|---------------|------------|-----------+| id | create_time | order_no | product_id | pay_count |+---------------|---------------------|---------------|------------|-----------+| 1652077145222 | 2022-05-09 14:19:05 | 1652077145224 | 1 | 1 |+---------------|---------------------|---------------|------------|-----------+SELECT * FROM tx_log;+--------------------------------------|---------------------+| tx_no | create_time |+--------------------------------------|---------------------+| 3fcd4e8d-1f5b-448d-ad3d-693c335f994e | 2022-05-09 06:19:05 |+--------------------------------------|---------------------+-- stock 库下SELECT * FROM tx_log;+--------------------------------------|---------------------+| tx_no | create_time |+--------------------------------------|---------------------+| 3fcd4e8d-1f5b-448d-ad3d-693c335f994e | 2022-05-09 06:23:00 |+--------------------------------------|---------------------+SELECT * FROM stock;+----|------------|-------------+| id | product_id | total_count |+----|------------|-------------+| 1 | 1 | 9999 |+----|------------|-------------+
复制代码
（2）异常流程：消息中间件宕机
1. 步骤一出现异常
即：还没开始下单，消息中间件宕机了，那么立刻下单失败。 如图：
2. 步骤四出现异常
即：此步骤还在下单事务中，当提交中间请求失败，本地事务不会回滚。
实验步骤：
在执行本地事务之后，睡眠 30s
此期间，消息中间件宕机：broker 关闭。
日志如下：
2022-05-09 15:27:14.980 c.d.t.message.OrderTxMessageListener : 订单微服务执行本地事务2022-05-09 15:27:15.015 c.d.t.message.OrderTxMessageListener : 订单微服务提交事务2022-05-09 15:27:15.015 c.d.t.message.OrderTxMessageListener : 尝试睡 30s 2022-05-09 15:27:39.025 INFO 22815 --- [lientSelector_1] RocketmqRemoting : closeChannel: close the connection to remote address[] result: true2022-05-09 15:27:39.025 INFO 22815 --- [lientSelector_1] RocketmqRemoting : closeChannel: close the connection to remote address[] result: true2022-05-09 15:27:45.015 c.d.t.message.OrderTxMessageListener : 睡醒，起来干活了2022-05-09 15:27:48.041 INFO 22815 --- [lientSelector_1] RocketmqRemoting : closeChannel: close the connection to remote address[] result: true2022-05-09 15:28:09.017 INFO 22815 --- [lientSelector_1] RocketmqRemoting : closeChannel: close the connection to remote address[] result: true2022-05-09 15:28:09.019 INFO 22815 --- [lientSelector_1] RocketmqRemoting : closeChannel: close the connection to remote address[] result: true。。。。。。2022-05-09 15:35:47.481 INFO 22815 --- [pool-1-thread-1] c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务
复制代码
（3）异常流程：事务发送方执行本地事务失败
即：这时候捕捉到异常：
订单服务给 RocketMQ 发送回滚消息：RocketMQLocalTransactionState.ROLLBACK
RocketMQ 接收到消息后，会回查
回查，发现不存在这个订单，订单服务向 RocketMQ 发送 未知消息：RocketMQLocalTransactionState.UNKNOWN UNKNOWN 未知状态：可能是事务正在执行中出异常等，这种情况下消息系统不知道该如何处理，当前的逻辑是会直接丢弃掉，等待后续检查逻辑来处理。
日志如下：
2022-05-09 15:45:11.829 c.d.t.message.OrderTxMessageListener : 订单微服务执行本地事务2022-05-09 15:45:11.861 c.d.t.message.OrderTxMessageListener : 订单微服务回滚事务2022-05-09 15:45:47.490 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:46:47.484 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:47:47.489 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:48:47.491 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:49:47.492 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:50:47.494 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:51:47.496 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:52:47.497 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:53:47.498 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:54:47.501 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:55:47.501 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:56:47.503 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:57:47.504 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:58:47.506 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务2022-05-09 15:59:47.510 c.d.t.message.OrderTxMessageListener : 订单微服务查询本地事务
复制代码
可看到 RocketMQ 回查了 15 次。
（4）异常流程：事务接收方宕机
即：RocketMQ 无法推送消息给消息接收方。
当库存服务再次上线后：会接收到消息。
订单服务日志：
2022-05-09 16:47:53.142 c.d.t.message.OrderTxMessageListener : 订单微服务执行本地事务2022-05-09 16:47:53.174 c.d.t.message.OrderTxMessageListener : 订单微服务提交事务
复制代码
库存服务日志：
2022-05-09 16:48:10.283 c.d.t.message.StockTxMessageConsumer : 库存微服务开始消费事务消息:{"txMessage":{"payCount":1,"productId":1,"txNo":"d5d86dae-76c2-4b56-9597-a12dae14325a"}}2022-05-09 16:48:10.284 c.d.t.service.impl.StockServiceImpl : 库存微服务执行本地事务,商品id:1, 购买数量:1
复制代码
（5）异常流程：事务接收方执行本地事务失败
即：当接收到消息后，执行本地事务失败，RocketMQ 会不断发送消费消息。
事务接收方执行本地事务失败，措施有：
记录日志，人工介入处理。
重试，再出错，则人工介入。
实验步骤：
下单错误商品，订单下单成功。
库存中没有此商品数据，向上抛错。
日志：
2022-05-09 14:19:59.930 WARN 19542 --- [MessageThread_3] a.r.s.s.DefaultRocketMQListenerContainer : consume message failed. messageExt:MessageExt [queueId=0, storeSize=480, queueOffset=1, sysFlag=8, bornTimestamp=1652077145061, bornHost=/172.17.0.1:36848, storeTimestamp=1652077199916, storeHost=/172.17.0.3:10911, msgId=AC11000300002A9F0000000000000D1B, commitLogOffset=3355, bodyCRjava.lang.NullPointerException: null at com.donald.txmsgstock.service.impl.StockServiceImpl.decreaseStock(StockServiceImpl.java:35) ~[classes/:na] at com.donald.txmsgstock.message.StockTxMessageConsumer.onMessage(StockTxMessageConsumer.java:27) ~[classes/:na] at com.donald.txmsgstock.message.StockTxMessageConsumer.onMessage(StockTxMessageConsumer.java:16) ~[classes/:na] at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer$DefaultMessageListenerConcurrently.consumeMessage(DefaultRocketMQListenerContainer.java:308) ~[rocketmq-spring-boot-2.0.2.jar:2.0.2] at org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:417) [rocketmq-client-4.4.0.jar:4.4.0] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_162] at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_162] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_162] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_162] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_162]
复制代码
参考资料
推荐一个零声教育 C/C++后台开发的免费公开课程，个人觉得老师讲得不错，分享给大家：C/C++后台开发高级架构师，内容包括Linux，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK等技术内容，立即学习
原文地址：可靠消息最终一致性分布式事务 - 掘金
评论