RabbitMQ：如何保证消息的可靠性？

• RabbitMQ基础
  
• RabbitMQ支持的消息模子
  
• SpringBoot集成RabbitMQ
  

一、发送者的可靠性

消息从发送者发送消息，到消费者处置处罚消息，需要颠末的流程是如许的：

消息从生产者到消费者的每一步都有大概导致消息丢失：

• 发送消息时丢失：
  
  
  
  • 生产者发送消息时连接 MQ
    
  • 失败生产者发送消息到达 MQ 后未找到 Exchange
    
  • 生产者发送消息到达 MQ 的 Exchange 后，未找到合适的 Queue
    
  • 消息到达 MQ 后，处置处罚消息的进程发生异常。
    
  
  
• MQ导致丢失：
  
  
  
  • 消息到达 MQ，保存到队列后，尚未消费就突然宕机。
    
  
  
• 消费者处置处罚消费时丢失：
  
  
  
  • 消息接收后尚未处置处罚突然宕机
    
  • 消息接收后处置处罚过程中抛出异常
    
  
  

综上 我们要办理消息丢失问题，保证 MQ 的可靠性，就必须从3个方面入手：

• 确保生产者一定把消息发送到 MQ
  
• 确保 MQ 不会将消息弄丢
  
• 确保消费者一定要处置处罚消息
  

1.1、生产者重试机制

第一种环境：生产者发送消息时，出现了网络故障，导致与MQ的连接停止。为相识决这个问题，SpringAMQP 提供消息发送时重试机制。修改 publisher 模块的设置文件，添加内容如下：

1. spring:
2.   rabbitmq:
3.     host: 192.168.137.120
4.     port: 5672
5.     virtual-host: /
6.     username: admin
7.     password: 123456
8.     connection-timeout: 1s  # 设置MQ的连接超时时间
9.     template:
10.       retry:
11.         enabled: true  # 开启超时重试机制
12.         initial-interval: 1000s # 失败后的初始等待时间
13.         multiplier: 1 # 失败后下次等待时长倍数
14.         max-attempts: 3 # 最大重试次数

复制代码

停止rabbitMQ 服务

1. docker stop mq

复制代码

测试发送一条消息，查察结果是否有重试机制出现

   留意：当网络不稳定的时间，使用重试机制可以有效进步消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试期待的过程中，当前线程是被阻塞的。假如对于业务性能有要求，建议禁用重试机制。假如一定要使用，请合理设置期待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。
  1.2、生产者确认机制

一般环境下，只要生产者与MQ 之间的网路连接顺畅，根本不会出现发送消息丢失的环境，因此大多数环境下我们无需考虑这种问题。不过，在少数环境下，也会出现消息发送到MQ之后丢失的现象，好比：

• MQ 内部处置处罚消息的进程发生了异常
  
• 生产者发送消息到达 MQ 后未找到Exchange
  
• 生产者发送消息到达 MQ 的 Exchange 后，未找到合适的Queue，因此无法路由。
  

针对上述环境，RabbitMQ 提供了生产者消息确认机制，包罗Publisher Confirm和Publisher Return两种。在开启确认机制的环境下，当生产者发送消息给MQ后，MQ 会根据消息处置处罚的环境返回不同的回执。

总结如下：

• 当消息投递到 MQ，但是路由失败时，通过 Publisher Return 返回异常信息，同时返回 ACK 简直认信息，代表投递成功。
  
• 临时消息投递到了 MQ，并且入队成功，返回 ACK，告知投递成功。
  
• 持久消息投递到了MQ，并且入队完成持久化，返回 ACK ，告知投递成功。
  
• 其它环境都会返回 NACK，告知投递失败。
  

其中ack和 nack 属于 Publisher Confirm 机制，ack是投递成功；nack是投递失败。而return则属于 Publisher Return 机制。
默认两种机制都是关闭状态，需要通过设置文件来开启。
1.3、生产者确认机制实现

1.3.1、生产者开启确认机制

在 publisher模块的 application.yaml中添加设置：

1. spring:
2.   rabbitmq:
3.     publisher-confirm-type: correlated # 开启publisher confirm 机制，并设置confirm 类型
4.     publisher-returns: true # 开启publisher return 机制

复制代码

这里publisher-confirm-type 有三种模式可选：

• **none：**关闭 confirm 机制。
  
• simple： 同步阻塞期待 MQ 的回执。
  
• correlated： MQ异步回调返回回执。
  

一般选择 correlated 回调机制。
1.3.2、定义ReturnCallback

每个RabbitTemplate 只能设置一个ReturnCallback，因此我们可以在设置类中统一设置。我们在publisher 模块定义一个设置类，内容如下：

1. package com.itheima.publisher.config;
3. import lombok.extern.slf4j.Slf4j;
4. import org.springframework.amqp.core.ReturnedMessage;
5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
6. import org.springframework.beans.BeansException;
7. import org.springframework.context.ApplicationContext;
8. import org.springframework.context.ApplicationContextAware;
9. import org.springframework.context.annotation.Configuration;
11. /**
12. * @author db
13. * @version 1.0
14. * @description MqConfig
15. * @since 2023/12/7
16. */
17. @Configuration
18. @Slf4j
19. public class MqConfig implements ApplicationContextAware {
20.     @Override
21.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
22.         RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
23.         rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
24.             @Override
25.             public void returnedMessage(ReturnedMessage returnedMessage) {
26.                 log.debug("触发return callback,");
27.                 log.debug("exchange: {}", returnedMessage.getExchange());
28.                 log.debug("routingKey: {}", returnedMessage.getRoutingKey());
29.                 log.debug("message: {}", returnedMessage.getMessage());
30.                 log.debug("replyCode: {}", returnedMessage.getReplyCode());
31.                 log.debug("replyText: {}", returnedMessage.getReplyText());
32.             }
33.         });
34.     }
35. }

复制代码

1.3.3、定义ConfirmCallback

由于每个消息发送时的处置处罚逻辑不一定相同，因此ConfirmCallback需要在每次发消息时定义。具体来说，是在调用RabbitTemplate中的convertAndSend 方法时，多通报一个参数CorrelationData这里的CorrelationData中包含两个核心的东西：

• id： 消息的唯一标示，MQ 对不同的消息的回执以此做判断，制止肴杂。
  
• SettableListenableFuture： 回执结果的 Future 对象。
  

新建一个测试，向系统自带的交换机发送消息，并且添加ConfirmCallback：

1.     @Test
2.     public void testPublisherConfirm() throws InterruptedException {
3.         // 创建CorrelationData
4.         CorrelationData correlationData = new CorrelationData();
5.         // 给future添加confirmCallback
6.         correlationData.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
7.             @Override
8.             public void onFailure(Throwable ex) {
9.                 // 2.1.Future发生异常时的处理逻辑，基本不会触发
10.                 log.error("send message fail", ex);
11.             }
13.             @Override
14.             public void onSuccess(CorrelationData.Confirm result) {
15.                 // 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容
16.                 if(result.isAck()){ // result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执
17.                     log.debug("发送消息成功，收到 ack!");
18.                 }else{ // result.getReason()，String类型，返回nack时的异常描述
19.                     log.error("发送消息失败，收到 nack, reason : {}", result.getReason());
20.                 }
21.             }
22.         });
24.         // 发送消息
25.         rabbitTemplate.convertAndSend("logs","info","hello",correlationData);
27.         Thread.sleep(2000);
28.     }

复制代码

测试结果：

改变交换机，使用错误的交换机，由于通报的RoutingKey 是错误的，路由失败后，触发了return callback，同时也收到了ack。当我们修改为正确的RoutingKey以后，就不会触发return callback了，只收到ack。而假如连交换机都是错误的，则只会收到nack。

   留意：开启生产者确认比较消耗MQ性能，一般不建议开启。
  二、MQ的可靠性

在默认环境下，RabbitMQ会将接收到的信息保存在内存中以低落消息收发的延伸。如许会导致两个问题:

• 一旦MQ宕机，内存中的消息会丢失。
  
• 内存空间有限，当消费者故障或处置处罚过慢时，会导致消息积压，引发MQ阻塞。
  

2.1、数据持久化

为了提拔性能，默认环境下MQ 的数据都是在内存存储的临时数据，重启后就会消失。为了保证数据的可靠性，必须设置数据持久化，包罗：

• 交换机持久化
  
• 队列持久化
  
• 消息持久化
  

交换机的持久化：添加交换机时，设置交换机的Durability参数：

队列的持久化，添加队列时，设置交换机的Durability参数：

队列的持久化，控制台发送消息的时间，可以添加很多参数，而消息的持久化要设置一个properties：

说明：
在开启持久化机制以后，假犹如时还开启了生产者确认，那么MQ会在消息持久化以后才发送ACK 回执，进一步确保消息的可靠性。不过出于性能考虑，为了淘汰IO 次数，发送到MQ 的消息并不是逐条持久化到数据库的，而是每隔一段时间批量持久化。一般间隔在100毫秒左右，这就会导致ACK 有一定的延伸，因此建议生产者确认全部采取异步方式。
2.2、LazyQueue

在默认环境下，RabbitMQ会将接收到的信息保存在内存中以低落消息收发的延伸。但在某些特殊环境下，这会导致消息积压，好比：

• 消费者宕机或出现网络故障
  
• 消息发送量激增，凌驾了消费者处置处罚速率
  
• 消费者处置处罚业务发生阻塞
  

一旦出现消息堆积问题，RabbitMQ 的内存占用就会越来越高，直到触发内存预警上限。此时RabbitMQ 会将内存消息刷到磁盘上，这个举动成为PageOut，PageOut会耗费一段时间，并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处置处罚新的消息，生产者的全部请求都会被阻塞为相识决这个问题，从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues 的模式，也就是惰性队列。惰性队列的特征如下：

• 接收到消息后直接存入磁盘而非内存
  
• 消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载）
  
• 支持数百万条的消息存储
  

控制台设置Lazy模式
在添加队列的时间，添加x-queue-mod=lazy参数即可设置队列为Lazy模式!

代码设置Lazy模式
使用SpringAMQP声明队列的时间，添加x-queue-mod=lazy参数也可设置队列为Lazy模式：

1.     @Bean
2.     public Queue lazyQueue(){
3.         return QueueBuilder
4.                 .durable("lazy.queue")
5.                 .lazy() // 开启Lazy模式
6.                 .build();
7.     }

复制代码

也可以基于注解来声明队列并设置为Lazy模式：

1.     @RabbitListener(bindings = @QueueBinding(value = @Queue, // 创建临时队列
2.             key = {"user.#"},  //路由key
3.             exchange = @Exchange(name="topics",type="topic"),  // 绑定交换机
4.             arguments = @Argument(name="x-queue-mod",value = "lazy")
5.     ))
6.     public void listenLazyQueueMessage(String msg) throws InterruptedException {
7.         System.out.println("消费者2接收到消息：【" + msg + "】");
8.     }

复制代码

三、消费者的可靠性

3.1、消费者确认机制

为了确认消费者是否成功处置处罚消息，RabbitMQ提供了消费者确认机制（Consumer Acknowledgement）。即：当消费者处置处罚消息竣事后，应该向RabbitMQ 发送一个回执，告知RabbitMQ 自己消息处置处罚状态。回执有三种可选值：

• ack： 成功处置处罚消息，RabbitMQ 从队列中删除该消息。
  
• nack： 消息处置处罚失败，RabbitMQ 需要再次投递消息。
  
• reject： 消息处置处罚失败并拒绝该消息，RabbitMQ 从队列中删除该消息。
  

一般reject 方式用的较少，除非是消息格式有问题，那就是开发问题了。因此大多数环境下我们需要将消息处置处罚的代码通过try catch 机制捕捉，消息处置处罚成功时返回ack，处置处罚失败时返回nack。由于消息回执的处置处罚代码比较统一，因此SpringAMQP 帮我们实现了消息确认。并允许我们通过设置文件设置ACK处置处罚方式，有三种模式：

• none：不处置处罚。即消息投递给消费者后立刻 ack，消息会立刻从 MQ 删除。非常不安全，不建议使用。
  
• manual：手动模式。需要自己在业务代码中调用 api ，发送 ack 或 reject，存在业务入侵，但更机动。
  
• auto：自动模式。SpringAMQP 使用 AOP 对我们的消息处置处罚逻辑做了环绕增强，当业务正常执行时则自动返回 ack。当业务出现异常时，根据异常判断返回不同结果：
  
  
  
  • 假如是业务异常，会自动返回 nack；
    
  • 假如是消息处置处罚或校验异常，自动返回 reject;
    
  
  

修改SpringAMQP的ACK处置处罚方式：

1. spring:
2.   rabbitmq:
3.     host: 192.168.137.120
4.     port: 5672
5.     username: admin
6.     password: 123456
7.     virtual-host: /
8.     connection-timeout: 600000
9.     listener:
10.       simple:
11.         acknowledge-mode: none  # 不做处理

复制代码

修改consumer服务的SpringRabbitListener类中的方法，模拟一个消息处置处罚的异常：

1. @RabbitListener(queues = "simple.queue")
2. public void listenSimpleQueueMessage(String msg) throws InterruptedException {
3.     log.info("spring 消费者接收到消息：【" + msg + "】");
4.     if (true) {
5.         throw new MessageConversionException("故意的");
6.     }
7.     log.info("消息处理完成");
8. }

复制代码

测试可以发现：当消息处置处罚发生异常时，消息依然被RabbitMQ删除了。确认机制修改为auto：

1. spring:
2.   rabbitmq:
3.     host: 192.168.137.120
4.     port: 5672
5.     username: admin
6.     password: 123456
7.     virtual-host: /
8.     connection-timeout: 600000
9.     listener:
10.       simple:
11.         acknowledge-mode: auto  # 自动ack

复制代码

在异常位置打断点，再次发送消息，程序卡在断点时，可以发现此时消息状态为 unacked（未确定状态）放行以后，由于抛出的是消息转换异常，因此 Spring 会自动返回 reject，以是消息依然会被删除：将异常改为 RuntimeException 范例：

1. @RabbitListener(queues = "object.queue")
2. public void listenObjectMessage(Map<String,Object> msg) throws InterruptedException {
3.     System.out.println("消费者接收到消息：【" + msg + "】");
4.     throw new RuntimeException("故意的");
5. }

复制代码

在异常位置打断点，然后再次发送消息测试，程序卡在断点时，可以发现此时消息状态为 unacked（未确定状态）：放行以后，由于抛出的是业务异常，以是 Spring返回 ack，最终消息规复至 Ready 状态，并且没有被 RabbitMQ删除：当我们把设置改为 auto时，消息处置处罚失败后，会回到 RabbitMQ ，并重新投递到消费者。
3.2、失败重试机制

当消费者出现异常后，消息会不停requeue（重入队）到队列，再重新发送给消费者。假如消费者再次执行依然出错，消息会再次requeue到队列，再次投递，直到消息处置处罚成功为止。极度环境就是消费者一直无法执行成功，那么消息requeue就会无穷循环，导致MQ的消息处置处罚飙升，带来不必要的压力
失败重试机制：在消费者出现异常时使用本地重试，而不是无穷制的requeue到MQ队列。
修改consumer服务的application.yml文件，添加内容：

1. spring:
2.   rabbitmq:
3.     host: 192.168.137.120
4.     port: 5672
5.     username: admin
6.     password: 123456
7.     virtual-host: /
8.     connection-timeout: 600000
9.     listener:
10.       simple:
11.         acknowledge-mode: auto  # 不做处理
12.         retry:
13.           enabled: true # 开启消费者失败重试
14.           initial-interval: 1000ms # 初识的失败等待时长为1秒
15.           multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
16.           max-attempts: 3 # 最大重试次数
17.           stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

复制代码

重启 consumer 服务，重复之前的测试。可以发现：

• 消费者在失败后消息没有重新回到 MQ 无穷重新投递，而是在本地重试了3次
  
• 本地重试3次以后，抛出了 AmqpRejectAndDontRequeueException 异常。查察 RabbitMQ 控制台，发现消息被删除了，说明最后 SpringAMQP 返回的是 reject
  

结论：

• 开启本地重试时，消息处置处罚过程中抛出异常，不会 requeue 到队列，而是在消费者本地重试。
  
• 重试到达最大次数后，Spring 会返回 reject，消息会被丢弃。
  

3.3、失败处置处罚计谋

在之前的测试中，本地测试到达最大重试次数后，消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下，显然不太合适了。因此 Spring 允许我们自定义重试次数耗尽后的消息处置处罚计谋，这个计谋是由MessageRecovery 接口来定义的，它有3个不同实现：

• RejectAndDontRequeueRecoverer： 重试耗尽后，直接 reject，丢弃消息。默认就是这种方式。
  
• ImmediateRequeueMessageRecoverer： 重试耗尽后，返回nack，消息重新入队。
  
• RepublishMessageRecoverer： 重试耗尽后，将失败消息投递到指定的交换机。
  

比较优雅的一种处置处罚方案是 RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处置处罚。
1）在 consumer 服务中定义处置处罚失败消息的交换机和队列。
2）定义一个RepublishMessageRecoverer，关联队列和交换机。

1. package com.itheima.consumer.config;
3. import org.springframework.amqp.core.Binding;
4. import org.springframework.amqp.core.BindingBuilder;
5. import org.springframework.amqp.core.DirectExchange;
6. import org.springframework.amqp.core.Queue;
7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
8. import org.springframework.amqp.rabbit.retry.MessageRecoverer;
9. import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
10. import org.springframework.context.annotation.Bean;
12. @Configuration
13. @ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
14. public class ErrorMessageConfig {
15.     @Bean
16.     public DirectExchange errorMessageExchange(){
17.         return new DirectExchange("error.direct");
18.     }
19.     @Bean
20.     public Queue errorQueue(){
21.         return new Queue("error.queue", true);
22.     }
23.     @Bean
24.     public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
25.         return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
26.     }
28.     @Bean
29.     public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
30.         return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
31.     }
32. }

复制代码

消费者如何保证消息一定被消费？
开启消费者确认机制为auto，由Spring确认消息处置处罚成功后返回ack，异常时返回nack开启消费者失败重试机制，并设置MessageRecoverer多次重试失败后将消息投递给异常交换机，交由人工处置处罚。
代码地址：https://gitee.com/duan138/practice-code/tree/master/mq-demo
四、总结

要想保证消息的可靠性，就要先知道什么时间消息会丢失？会有三方面分别是：消息到MQ的过程丢失、MQ 自己丢失、MQ 到消费过程中丢失。文章中提到从三方面去保证消息的可靠性。分别是生产者到 RabbitMQ使用事务机制或者Confirm机制；RabbitMQ自身持久化、集群等；RabbitMQ到消费者采取basicAck机制、死信队列、消息赔偿机制。
参考：https://www.bilibili.com/video/BV1mN4y1Z7t9

---

改变你能改变的，担当你不能改变的，关注公众号：程序员康康，一起发展，共同进步。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。