Rabbitmq解决分布式事务思路

Posted 2023-05-03

参考技术A 上一篇 <<< Rabbitmq通过死信队列实现过期监听
下一篇 >>> Rabbitmq解决分布式事务demo

推荐阅读：
<<< 消息中间件的核心思想
<<< 消息中间件常见问题汇总
<<< 基于Netty简单手写消息中间件思路
<<< 消息队列常用名词与中间件对比
<<< Rabbitmq基础知识
<<< Rabbitmq示例之点对点简单队列
<<< Rabbitmq示例之工作(公平)队列
<<< Rabbitmq示例之发布订阅模式
<<< Rabbitmq示例之路由模式Routing
<<< Rabbitmq示例之通配符模式Topics
<<< Rabbitmq示例之RPC模式
<<< Rabbitmq队列模式总结
<<< Rabbitmq如何保证消息不丢失
<<< Springboot利用AmqpTemplate整合Rabbitmq
<<< Rabbitmq如何保证幂等性
<<< Rabbitmq的重试策略
<<< Rabbitmq通过死信队列实现过期监听
<<< Rabbitmq解决分布式事务demo
<<< Rabbitmq环境安装
<<< Kafka中的专业术语都有哪些
<<< Kafka的设计原理介绍
<<< Kafka集群如何实现相互感知
<<< Kafka如何实现分区及指定分区消费
<<< Kafka如何保证消息顺序消费
<<< Kafka如何保证高吞吐量
<<< Kafka集群环境搭建
<<< RocketMQ架构原理
<<< RocketMQ、RabbitMQ和Kafka的对比
<<< SpringBoot整合RocketMQ示例
<<< RocketMQ保证顺序消费demo
<<< RocketMQ如何动态扩容和缩容
<<< RocketMQ如何解决分布式事务
<<< RocketMQ单机版本安装
<<< RocketMQ集群环境程序启用相关知识点
<<< RocketMQ单机做主备实操
<<< RocketMQ所有配置说明

漫画使用消息队列解决分布式事务

RabbitMQ的核心概念

 

重要组成部分说明如下：

Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。（邮局部门/快递公司）

Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。（邮局的快递员/快递公司快递员）

Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。（收件人家的信箱/菜鸟驿站）

Producer：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。（寄信人/寄货人）

Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。（收信人/收货人）

 -----发送消息-----

1、生产者和Broker建立TCP连接。

2、生产者和Broker建立通道。

3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。

4、Exchange将消息转发到指定的Queue（队列）

----接收消息-----

1、消费者和Broker建立TCP连接

2、消费者和Broker建立通道

3、消费者监听指定的Queue（队列）

4、当有消息到达Queue时Broker默认将消息推送给消费者。

5、消费者接收到消息。

工作流程

分布式事务的产生

多个系统相互配合工作，产生数据一致性问题。

例如外卖场景中，下单中心，运单中心两个系统要配合工作，必须保证两个系统数据一致性。

使用API接口调用，下单中心插入数据，调用运单中心的API接口处理数据，并启动事务回滚。

咋一看这场景没有什么问题，毕竟有事务回滚，一起成功一起失败。但其实存在API调用超时的情况，此时下单中心以为调用失败回滚，而运单中心只是超时仍会继续执行程序，从而造成两个系统数据不一致。

假设API调用成功，也有可能是在订单中心提交事务时失败了，此时订单中心回滚，而API已经调用，下单中心的数据已经产生，数据不一致。

使用消息队列解决分布式事务

先看一张图：

可靠生产：下单中心处理数据和状态表更改应该保证事务一致。生产者往消息队列发送数据时，在本地建立一张状态表，看是否成功发送给队列。利用RabbitMQ的确认机制看是否重发还是定时扫描状态表重发，保证可靠生产。兜底方案还是定时扫描状态表。

代码：

 /**    * 分单处理队列    */ publicstaticfinal String QUEUE_NAME_TRANSACTION = "xucheng.distribute.queue";    /**     * 分单处理交换机     */    publicstaticfinal String EXCHANGE_NAME_TRANSACTION =   "xucheng.distribute.exchange";    /**     * 消息队列服务进程，此进程包括两个部分：Exchange和Queue。     */    publicstaticfinal String ROUTE_NAME_TRANSACTION =  "xucheng.distribute.route";      /**     * 补单队列     */     //public static final StringCREATE_QUEUE_NAME_TRANSACTION = "xucheng.order.reCreate.queue";     /**     * 1、交换机绑定到分单队列     * @return     */     @Bean     public DirectExchange transExchange() {         returnnew DirectExchange(EXCHANGE_NAME_TRANSACTION);     }    /**     *2、分单队列     *     * @return     */    @Bean    public Queue scoreQueue() {        returnnew Queue(QUEUE_NAME_TRANSACTION, true);    }     /**      * 3、Binding      * @return      */    @Bean    public Binding bindingExchangeOrderReceiverQueue(){    //通过Binding将Exchange与Queue关联起来。　　        return BindingBuilder.bind(scoreQueue()).to(transExchange()).with(ROUTE_NAME_TRANSACTION);}

---

可靠消费：消费端开启手动ACK，给MQ队列发送确认信息。对每条数据进行记录，一旦有异常记录可以试着去重试重新要求MQ再发数据，但不能重试太多次。可以将数据主键插入数据库，这样同一数据就不会执行两次，或者使用redis记录数据操作。

代码：

● 【漫画】

点击在看，驱动原创