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

RabbitMQ确保消息可靠性

news 2025/6/24 10:11:43

消息丢失的可能性

在这里插入图片描述
支付服务先扣减余额和更新支付状态（这俩是同步调用），然后通过RabbitMq异步调用支付服务更新订单状态。但是有些情况下，可能订单已经支付 ，但是更新订单状态却失败了，这就出现了消息丢失。

  1. 发送者在发送的过程中出现了网络故障
  2. RabbitMQ在发送消息的过程中出现了问题
  3. 消费者在更新订单状态的时候出现了问题

发送者的可靠性

发送者确认机制需要与MQ进行通信和确认，会影响消息发送的效率且一般出现的概率极低，所以一般不用这个。

方法1. 发送者重连

确保发送者与MQ之间连接的可靠性。有的时候由于网络波动，可能出现发送者连接MQ失败的情况，这个配置是关闭的，可以开启连接失败后的重连机制：

spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制(默认是false)initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multipliermax-attempts: 3 # 最大重试次数

【注】：当网络不稳定时，利用重试机制可以提高消息发送的成功率，但是SpringAMQP提供的重试机制是阻塞式的重试，如果需要多次重试等待，当前线程被阻塞，会影响性能。
如果对业务性能有要求，建议禁用重试机制，如果一定要使用，要合理的配置等待时常和重试次数，或使用异步线程来执行发送消息的代码。

方法2. 发送者确认

确保消息发送的可靠性。SpringAMQP提供了Publisher Confirm和Publisher Return两种机制，开启确认机制后，当发送者发送消息给MQ后，MQ会返回确认结果给发送者，返回的结果有以下几种情况：

  • 消息投递到MQ，但是路由失败，此时通过PublisherReturn返回路由异常信息，然后返回ACK，告知投递成功。例如：
    • 消息发送给图中的exchange1，但是RoutingKey写错了，没有匹配到正确的队列，也会导致路由失败。
    • 消息发送给图中的exchange2，但是它底下没有绑定新的队列，就会导致路由失败。
  • 临时消息【不需要往磁盘做持久化的消息】投递到MQ，并入队成功，返回ACK，告知投递成功。
  • 持久消息投递到MQ，并入队完成持久化，返回ACK，告知投递成功。
  • 其他情况都会返回NACK，告知投递失败。
    在这里插入图片描述
    步骤：
  1. 在发送方publisher所在的微服务的application.yml中配置：
spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型publisher-returns: true # 开启publisher return机制

publisher-confirm-type有三种模式：

  • none：关闭confirm机制
  • simple：同步阻塞等待MQ回执消息
  • correlated：MQ异步回调方式返回回执消息（常用）
  1. 开启回调机制：每个RabbitTemplate只能配置一个ReturnCallback，在发送者publisher所在的项目启动时配置即可。
@Configuration
@RequiredArgsConstructor
@Slf4j
public class MqConfig {private final RabbitTemplate rabbitTemplate;@PostConstruct // 在Bean初始化完成后调用这个方法（只会调用一次）public void init() {// 返回ACK，但是此时路由失败，就会走这个方法rabbitTemplate.setReturnsCallback(returnedMessage -> {log.error("监听到了消息return callback");log.debug("exchange: {}", returnedMessage.getExchange());log.debug("routingKey: {}", returnedMessage.getRoutingKey());log.debug("message: {}", returnedMessage.getMessage());log.debug("replyCode: {}", returnedMessage.getReplyCode());log.debug("replyText: {}", returnedMessage.getReplyText());});}
}
  1. 开启消息确认机制：发送消息、指定消息ID、每次发送消息都需要配置一个ConfirmCallback
public void testConfirmCallback() {CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {// 【几乎不可能发生】Future发生异常时的处理逻辑@Overridepublic void onFailure(Throwable ex) {log.error("spring ampq处理确认结果异常", ex);}// 成功，拿到MQ结果，判断是ACK还是NACK@Overridepublic void onSuccess(CorrelationData.Confirm result) {if(result.isAck()) {// ACKlog.debug("收到ACK，消息发送成功");}else {// NACKlog.debug("收到NACK，消息发送失败，失败原因：{}",result.getReason());}}});rabbitTemplate.convertAndSend("hmall.direct", "blue", "hello world", cd); // 发送消息
}

MQ的可靠性

RabbitMQ一般会将收到的信息保存到内存（速度快）中，降低消息收发的延迟，这样会导致：

  1. MQ宕机，内存中的消息会丢失。
  2. 内存空间有限，消费者故障或处理过慢，会导致消息积压，引发MQ阻塞。
    在这里插入图片描述

【案例】发送者往MQ发消息，MQ会把数据保存到内存中，如果内存满了，MQ就会把一部分数据迁移到磁盘中暂时进行持久化存储，移动到磁盘的这段时间发送者发送的消息就会产生丢失。

方法1. 数据持久化

数据持久化就是把数据持久化到磁盘，但是不是向上边那个案例，等满了再去持久化（被动），而是提前进行持久化。

  1. 交换机的持久化（默认开启的）
    在这里插入图片描述
  2. 队列的持久化（默认开启的）
    在这里插入图片描述
  3. 消息持久化（默认是非持久的）
    在发送消息的时候设定的
    在这里插入图片描述
    【案例】：比较一下持久化和非持久化的性能。
    发100w条消息给MQ：
    这是非持久化的方式：使用纯内存的方式存储，每次内存满之后，MQ就会把消息写到磁盘中，此时就会出现阻塞状态，处理速度降低到0
    在这里插入图片描述
    【问题】可能出现消息丢失和MQ阻塞
    【解决办法】使用持久化的方式：
public void testSendPersistentMsg() {// 自定义构建消息Message msg = MessageBuilder.withBody("hello world".getBytes(StandardCharsets.UTF_8)) // 消息体.setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 投递模式（持久化）.build();for (int i = 0; i < 100000; ++i) {rabbitTemplate.convertAndSend("simple.queue", msg);}
}

Mq并没有阻塞，每发一条消息就赶紧把它存到磁盘中，和纯内存方式相比，不会有个中断的过程。
在这里插入图片描述

方法2. Lazy Queue（推荐）

【问题】：由于使用了消息持久化的方式，发到MQ的消息不仅要到内存，还要在磁盘中写一份，这会导致整体的并发能力下降
【特征】：

  • 接收到消息后直接入磁盘，不再存储到内存
  • 在写磁盘的时候也对写入磁盘的操作进行一些优化，比传统的写操作高很多
  • 消费者要消费消息时，才会从磁盘中读取并加载到内存
    • 【问题】：可能会影响消费者处理消息的速度
    • 【解决】：可以提前缓存部分消息到内存，最多2048条

控制台声明Lazy Queue队列

在这里插入图片描述

Java代码添加

声明Bean
@Bean
public Queue lazyQueue(){return QueueBuilder.durable("lazy.queue").lazy() // 开启Lazy模式.build();
}
@RabbitListener注解
@RabbitListener(queuesToDeclare = @Queue(name = "lazy.queue",durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy") // 开启Lazy模式
))
public void listenLazyQueue(String msg){log.info("接收到 lazy.queue的消息：{}", msg);
}

消费者的可靠性

消费者确认机制

为了确认消费者是否成功处理消息，当消费者处理消息结束后，应该向MQ发送一个回执，告知MQ自己的消息处理状态。有如下几种消息处理状态：

  • ack：处理消息成功，RabbitMQ从队列中删除该消息
  • nack：消息处理失败，RabbitMQ需要再次投递消息
  • reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息【在处理的过程中，发现消息的内容有问题，没有重试的必要，直接拒绝就行】

【注意】：不管是哪种情况，都应该等消息处理完后得到结果再返回，不要一拿到消息就返回
返回消息处理状态的过程，类似于处理事务，事务处理成功，返回ACK；处理失败，返回NACK

SpringAMQP允许通过在消费者的配置文件选择ACK的处理方式，有三种：

  • none：不处理，消息投递给消费者后立刻ack，消息会立刻从MQ中删除，别用
  • manual：手动模式，需要在业务代码中调用api，发送ack或reject，存在业务入侵，但是更灵活。
  • auto：自动模式，利用AOP对消息处理逻辑进行了环绕增强
    • 业务处理正常：自动返回ack
    • 业务处理异常：自动返回nack
    • 消息处理或校验异常【MessageConversionException】：自动返回reject
spring:rabbitmq:listener:simple:acknowledge-mode: auto# 不做处理

失败重试策略

在消费者出现异常时，利用本地重试，而不是无限的重新入队到mq，可以在消费者的yaml文件中添加配置来开启重试机制。

spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数（达到最大重试次数后，MQ会把消息丢弃）stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

【问题】：在开启重试模式后，重试次数耗尽，如果消息仍然失败，默认会把消息进行丢弃。
【解决】：因此需要有MessageRecoverer接口来处理，包含三种不同的实现：

  • RejectAndDontRequeueRecoverer（默认）：重试耗尽后，直接reject，丢弃消息。
  • ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队。
  • RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机 。
    在这里插入图片描述

修改失败重试策略为RepublishMessageRecoverer

  1. 定义接收失败的交换机、队列、
  2. 定义RepublishMessageRecoverer
@Configuration
public class ErrorMessageConfiguration {// 定义接收失败的交换机@Beanpublic DirectExchange errorExchange() {return new DirectExchange("error.direct");}// 定义接收失败的队列@Beanpublic Queue errorQueue() {return new Queue("error.queue");}// 定义绑定关系@Beanpublic Binding errorQueueBinding() {return BindingBuilder.bind(errorQueue()).to(errorExchange()).with("error");}// 定义失败处理策略@Beanpublic MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

业务幂等性

f(x) = f(f(x))，指同一个业务，执行一次或多次对业务状态的影响是一致的。

  • 幂等业务：查询业务、删除业务
  • 非幂等业务：用户下单需要扣减库存、用户退款业务需要恢复余额

方案1. 唯一消息id

给每个消息设置一个唯一id，利用id区分是否是重复消息：

  • 每条消息都生成一个唯一id，与消息一起投递给消费者
  • 消费者接收到消息后处理自己的业务，业务处理成功后将消息id保存到数据库中
  • 如果下次又收到相同消息，去数据库查询判断是否存在，存在则视为重复消息放弃处理
  1. 在发送方配置Bean用来自动创建消息id
@Configuration
public class MqConfig {@Beanpublic MessageConverter messageConverter() {Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();converter.setCreateMessageIds(true); // 配置自动创建消息idreturn converter;}
}
  1. 在接收方接收消息id
@RabbitListener(queues = "simple.queue")public void listenSimpleQueue(Message msg) { // 使用字符串发送，就用字符串接收log.info("监听到simple.queue的消息：{}", msg);log.info("消息id：{}", msg.getMessageProperties().getMessageId());// throw new RuntimeException("故意的");}

在这里插入图片描述

方案2. 业务判断（常用）

结合业务逻辑，基于业务本身做判断。
【案例】：当用户下单成功后，通过MQ通知交易服务来修改订单状态为已支付（这里记作消息1），修改成功后交易服务返回ACK给MQ，此时出现了网络的故障，MQ没有收到交易服务发送的ACK，MQ认为交易服务宕机，消息又重新入队。
就在此刻，用户点击了申请退款，直接向交易服务修改订单状态为退款中（这个操作没有走MQ，此时订单状态是退款中，但是消息1还在消息队列中）。
此时网络恢复了，MQ又将消息1发送给交易服务，此时交易服务又把订单状态标记为已支付（订单申请退款中的状态又被覆盖了）。
在这里插入图片描述
【解决】：通知来的时候，先判断订单的状态，再进行操作。

@Component
@RequiredArgsConstructor
public class PayStatusListener {private final IOrderService orderService;@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "trade.pay.success.queue", durable = "true"),exchange = @Exchange(name = "pay.direct"),key = "pay.success"))public void listenPaySuccess(Long orderId) {// 1.查询订单Order order = orderService.getById(orderId);// 2.判断订单状态是否为未支付if(order == null || order.getStatus() != 1) {// 不做处理return;}// 3.标记订单状态为已支付orderService.markOrderPaySuccess(orderId);}
}

在这里插入图片描述

延迟消息

延迟消息：发送者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。
延迟任务：设置在一定时间后才执行的任务。
在这里插入图片描述

方案1. 死信交换机

当一个队列中的消息满足下列情况之一的，就会成为死信：

  • 消费者使用basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false。
  • 消息是一个过期消息（达到队列设置的过期时间 或 消息本身设置的过期时间），超时无人消费。
  • 要投递的队列消息堆积满了，最早的消息可能成为死信。

队列通过dead-letter-exchange属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中，这个交换机就叫做死信交换机（DLX）。
在这里插入图片描述

  1. 声明死信队列、死信交换机、它们之间的绑定关系：
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "dlx.queue", durable = "true"), // 死信队列exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT), // 死信交换机key = {"hi"} ))public void listenDlxQueue(String msg) {log.info("消费者监听到dlx.queue的消息: " + msg);}
  1. 声明普通队列、普通交换机、它们之间的绑定关系，并把队列绑定到死信交换机上（此时就不需要把它绑定消费者了）：
@Configuration
public class NormalConfiguration {@Beanpublic DirectExchange normalExchange() { // 普通交换机return ExchangeBuilder.directExchange("normal.direct").build();}@Beanpublic Queue normalQueue() { // 普通队列return QueueBuilder.durable("normal.direct") // 队列名字.deadLetterExchange("dlx.direct") // 死信交换机名字.build();}@Beanpublic Binding normalQueueBinding(Queue normalQueue, DirectExchange normalExchange) { // 绑定关系// 把队列绑定到交换机return BindingBuilder.bind(normalQueue) // 队列.to(normalExchange) // 交换机.with("hi");// 这里绑定关系要和普通队列的绑定关系保持一致}
}
  1. 发送延迟消息：
@Test
public void testSendDelayMsg() {rabbitTemplate.convertAndSend("normal.direct", "hi", "hello world", message -> {// 当消息被转成Message对象后，还可以进一步做加工message.getMessageProperties().setExpiration("10000"); // 设置消息过期时间(10s)return message;});
}

【注】：normal.direct和normal.queue之间绑定的BindingKey 与 dlx.direct和dlx.queue之间绑定的BindingKey要一致

方案2. 延迟消息插件DelayExchange（推荐）

这个插件可以将普通交换机改造为支持延迟消息功能的交换机，当消息投递到交换机后，可以暂存一段时间，到后期再投递到队列。

一、安装插件

  1. 插件下载地址：DelayExchange
  2. 需要把插件放在RabbitMQ插件目录对应的数据卷下
docker volume inspect mq-plugins

在这里插入图片描述
在这里插入图片描述

  1. 执行命令，安装插件
docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

二、使用插件

  1. 声明延迟交换机：只要设置delay的属性为true即可
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "delay.queue", durable = "true"),exchange = @Exchange(name = "delay.direct", delayed = "true", type = ExchangeTypes.DIRECT), // 只要设置一个delayed属性为true即可key = {"hi"}
))
public void listenDelayQueue(String msg) {log.info("消费者监听到delay.queue的消息: " + msg);
}
  1. 发送延迟消息：通过消息头x-delay来设置过期时间
@Test
public void testSendDelayMsgByPlugin() {rabbitTemplate.convertAndSend("delay.direct", "delay", "hello world", message -> {message.getMessageProperties().setDelay(10000);// 添加延迟消息属性return message;});
}

延迟消息的实现需要记录消息的过期时间，计时的时钟需要依赖cpu，是个cpu密集型任务。因此使用延迟消息时，需要避免同一时刻在mq里存在大量的延迟消息（尽可能地让延迟消息的延迟时间不要太长）。

http://www.lryc.cn/news/521119.html

相关文章：

  • 前端常见的设计模式之【单例模式】
  • 【React】脚手架进阶
  • win32汇编环境,窗口程序中单选框的一般操作示例
  • 如何移除git中被跟踪的commit文件
  • 结合night compute分析 利用tensor core 优化K值较大的矩阵乘（超过cublas50%）
  • Docker 部署 Typecho
  • 【大数据】机器学习-----模型的评估方法
  • 【Excel笔记_3】execl的单元格是#DIV/0!，判断如果是这个，则该单元格等于空
  • FPGA EDA软件的位流验证
  • 信号与系统初识---信号的分类
  • 信号量机制之苹果-橘子问题
  • 三相无刷电机控制｜FOC理论04 - 克拉克变换 + 帕克变换的最终目标
  • Nacos: 一个动态服务发现与配置管理平台
  • 认识机器学习中的结构风险最小化准则
  • 计算机网络 （35）TCP报文段的首部格式
  • ubuntu24.04安装docker显卡工具包nvidia-container-toolkit
  • rknn环境搭建之docker篇
  • OpenCV相机标定与3D重建(56)估计物体姿态（即旋转和平移）的函数solvePnPRansac()的使用
  • vue倒计时组件封装,根据每个循环项的倒计时是否结束添加新类名。
  • 缩放 对内外参的影响
  • SQL面试题2：留存率问题
  • 晨辉面试抽签和评分管理系统之九：随机编排考生的分组(以教师资格考试面试为例)
  • 【EtherCATBridge】- KRTS C++示例精讲（9）
  • C++实现设计模式--- 观察者模式 (Observer)
  • iOS 解决两个tableView.嵌套滚动手势冲突
  • Lianwei 安全周报|2025.1.13
  • rtthread学习笔记系列(2) -- 宏
  • 美摄科技PC端视频编辑解决方案，为企业打造专属的高效创作平台
  • 服务端开发模式-thinkphp-重新整理workman
  • HTB：Access[WriteUP]