当前位置：首页 > news >正文

【RabbitMQ——消息应答机制——分布式事务解决方式】

news 2025/7/7 15:14:04

1. RabbitMQ高级-消息确认机制的配置

NONE值是禁用发布确认模式，是默认值

CORRELATED值是发布消息成功到交换器后会触发回调方法，如1示例
SIMPLE值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;

# 服务端口
server:port: 8080
# 配置rabbitmq服务
spring:rabbitmq:username: adminpassword: adminvirtual-host: /host: 47.104.141.27port: 5672publisher-confirm-type: correlated

创建一个自己的消息确认类

package com.xuexiangban.rabbitmq.springbootorderrabbitmqproducer.callback;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;public class MessageConfirmCallback implements RabbitTemplate.ConfirmCallback {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if(ack){System.out.println("消息确认成功!!!!");}else{System.out.println("消息确认失败!!!!");}}
}

public void makeOrderTopic(String userId,String productId,int num){// 1: 根据商品id查询库存是否充足// 2: 保存订单String orderId = UUID.randomUUID().toString();System.out.println("保存订单成功：id是：" + orderId);// 3: 发送消息//com.#  duanxin//#.email.* email//#.sms.# sms// 设置消息确认机制rabbitTemplate.setConfirmCallback(new MessageConfirmCallback());rabbitTemplate.convertAndSend("topic_order_ex","com.email.sms.xxx",orderId);
}

2. 分布式事务的解决方式

2.1 两阶段提交

2.1.1 概念

两阶段提交（Two-phase Commit，2PC），通过引入协调者（Coordinator）来协调参与者的行为，并最终决定这些参与者是否要真正执行事务。

准备阶段
- 协调者询问参与者事务是否执行成功，参与者发回事务执行结果。
提交阶段
- 如果事务在每个参与者上都执行成功，事务协调者发送通知让参与者提交事务；否则，协调者发送通知让参与者回滚事务。
- 需要注意的是，在准备阶段，参与者执行了事务，但是还未提交。只有在提交阶段接收到协调者发来的通知后，才进行提交或者回滚。

2.1.2 优缺点

同步阻塞
- 所有事物参与者在等待其他参与者响应的时候都处于同步阻塞状态，无法进行其他操作。
单点问题
- 协调者在2PC中起到非常大的作用，发生故障将会造成很大影响。特别是在阶段二发生故障，所有参与者会一直等待状态，无法完成其它操作。
数据不一致
- 在阶段二，如果协调者只发送了部分 Commit 消息，此时网络发生异常，那么只有部分参与者接收到 Commit 消息，也就是说只有部分参与者提交了事务，使得系统数据不一致。
太过保守
- 任意一个节点失败就会导致整个事务失败，没有完善的容错机制

2.2 补偿事务（TCC）

2.2.1 概念

TCC 其实就是采用的补偿机制，其核心思想是：针对每个操作，都要注册一个与其对应的确认和补偿（撤销）操作。它分为三个阶段：

Try阶段：主要是对业务系统做检测及资源预留
Confirm阶段：主要是对业务系统做确认提交，Try阶段执行成功并开始执行 Confirm阶段时，默认 - - - Confirm阶段是不会出错的。即：只要Try成功，Confirm一定成功。
Cancel阶段：主要是在业务执行错误，需要回滚的状态下执行的业务取消，预留资源释放。

举个例子，假入 Bob 要向 Smith 转账，思路大概是：我们有一个本地方法，里面依次调用 1：首先在 Try 阶段，要先调用远程接口把 Smith 和 Bob 的钱给冻结起来。 2：在 Confirm 阶段，执行远程调用的转账的操作，转账成功进行解冻。 3：如果第2步执行成功，那么转账成功，如果第二步执行失败，则调用远程冻结接口对应的解冻方法 (Cancel)。

2.2.2 优缺点

优点：跟2PC比起来，实现以及流程相对简单了一些，但数据的一致性比2PC也要差一些
缺点：缺点还是比较明显的，在2,3步中都有可能失败。TCC属于应用层的一种补偿方式，所以需要程序员在实现的时候多写很多补偿的代码，在一些场景中，一些业务流程可能用TCC不太好定义及处理。

2.3 本地消息表（异步确认）

2.3.1 概念

本地消息表与业务数据表处于同一个数据库中，这样就能利用本地事务来保证在对这两个表的操作满足事务特性，并且使用了消息队列来保证最终一致性。

在分布式事务操作的一方完成写业务数据的操作之后向本地消息表发送一个消息，本地事务能保证这个消息一定会被写入本地消息表中。
之后将本地消息表中的消息转发到 Kafka 等消息队列中，如果转发成功则将消息从本地消息表中删除，否则继续重新转发。
在分布式事务操作的另一方从消息队列中读取一个消息，并执行消息中的操作。

2.3.2 优缺点

优点：一种非常经典的实现，避免了分布式事务，实现了最终一致性。
缺点：消息表会耦合到业务系统中，如果没有封装好的解决方案，会有很多杂活需要处理。

2.4 MQ 事务消息

2.4.1 概念

异步场景，通用性较强，拓展性较高
有一些第三方的MQ是支持事务消息的，比如RocketMQ，他们支持事务消息的方式也是类似于采用的二阶段提交，但是市面上一些主流的MQ都是不支持事务消息的，比如 Kafka 不支持。

以阿里的 RabbitMQ 中间件为例，其思路大致为：

第一阶段Prepared消息，会拿到消息的地址。第二阶段执行本地事务，第三阶段通过第一阶段拿到的地址去访问消息，并修改状态。
也就是说在业务方法内要想消息队列提交两次请求，一次发送消息和一次确认消息。如果确认消息发送失败了RabbitMQ会定期扫描消息集群中的事务消息，这时候发现了Prepared消息，它会向消息发送者确认，所以生产方需要实现一个check接口，RabbitMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。

2.4.2 优缺点

优点：实现了最终一致性，不需要依赖本地数据库事务。
缺点：实现难度大，主流MQ不支持，RocketMQ事务消息部分代码也未开源。

3. 具体实现

3.1 可靠的生产流程

通过冗余的本地消息，等待确认机制修改本地消息的标志位。定期循环扫描未被发送的消息。

在这里插入图片描述

3.2 基于MQ的分布式事务消息的可靠消费