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

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

一、发送者的可靠性

消息从发送者发送消息，到消费者处理消息，需要经过的流程是这样的：

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

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

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

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

1.1、生产者重试机制

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

spring:rabbitmq:host: 192.168.137.120port: 5672virtual-host: /username: adminpassword: 123456connection-timeout: 1s  # 设置MQ的连接超时时间template:retry:enabled: true  # 开启超时重试机制initial-interval: 1000s # 失败后的初始等待时间multiplier: 1 # 失败后下次等待时长倍数max-attempts: 3 # 最大重试次数

停止rabbitMQ 服务

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中添加配置：

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

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

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

一般选择 correlated 回调机制。

1.3.2、定义`ReturnCallback`

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

package com.itheima.publisher.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;/*** @author db* @version 1.0* @description MqConfig* @since 2023/12/7*/
@Configuration
@Slf4j
public class MqConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.debug("触发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.3.3、定义`ConfirmCallback`

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

id： 消息的唯一标示，MQ 对不同的消息的回执以此做判断，避免混淆。
SettableListenableFuture： 回执结果的 Future 对象。

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

    @Testpublic void testPublisherConfirm() throws InterruptedException {// 创建CorrelationDataCorrelationData correlationData = new CorrelationData();// 给future添加confirmCallbackcorrelationData.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable ex) {// 2.1.Future发生异常时的处理逻辑，基本不会触发log.error("send message fail", ex);}@Overridepublic void onSuccess(CorrelationData.Confirm result) {// 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容if(result.isAck()){ // result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执log.debug("发送消息成功，收到 ack!");}else{ // result.getReason()，String类型，返回nack时的异常描述log.error("发送消息失败，收到 nack, reason : {}", result.getReason());}}});// 发送消息rabbitTemplate.convertAndSend("logs","info","hello",correlationData);Thread.sleep(2000);}

测试结果：

改变交换机，使用错误的交换机，由于传递的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模式：

    @Beanpublic Queue lazyQueue(){return QueueBuilder.durable("lazy.queue").lazy() // 开启Lazy模式.build();}

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

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

三、消费者的可靠性

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处理方式：

spring:rabbitmq:host: 192.168.137.120port: 5672username: adminpassword: 123456virtual-host: /connection-timeout: 600000listener:simple:acknowledge-mode: none  # 不做处理

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

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

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

spring:rabbitmq:host: 192.168.137.120port: 5672username: adminpassword: 123456virtual-host: /connection-timeout: 600000listener:simple:acknowledge-mode: auto  # 自动ack

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

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

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

3.2、失败重试机制

当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者。如果消费者再次执行依然出错，消息会再次requeue到队列，再次投递，直到消息处理成功为止。极端情况就是消费者一直无法执行成功，那么消息requeue就会无限循环，导致MQ的消息处理飙升，带来不必要的压力

失败重试机制：在消费者出现异常时利用本地重试，而不是无限制的requeue到MQ队列。

修改consumer服务的application.yml文件，添加内容：

spring:rabbitmq:host: 192.168.137.120port: 5672username: adminpassword: 123456virtual-host: /connection-timeout: 600000listener:simple:acknowledge-mode: auto  # 不做处理retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数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，关联队列和交换机。

package com.itheima.consumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
public class ErrorMessageConfig {@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");}@Beanpublic Queue errorQueue(){return new Queue("error.queue", true);}@Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

消费者如何保证消息一定被消费？

开启消费者确认机制为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

改变你能改变的，接受你不能改变的，关注公众号：程序员康康，一起成长，共同进步。