消息队列MQ（二）

---

前言

在用MQ实现异步调用时，我们必须尽可能确保MQ消息的可靠性，即消息应该至少被消费者处理1次，而消息从生产者到消费者每一步都可能导致消息丢失：

• 发送消息时丢失：
  1. 生产者发送消息时连接MQ失败。
  2. 生产者发送消息到达MQ后未找到Exchange。
  3. 生产者发送消息到达MQ的Exchange后，未找到合适的Queue。
  4. 消息到达MQ后，处理消息的进程发生异常。
• MQ导致消息丢失：
  • 消息到达MQ，保存到队列后，尚未进行下一步处理（比如消费）就突然宕机。
• 消费者的处理消息时：
  • 消息接收后尚未处理突然宕机。
  • 消息接收后处理过程中抛出异常。

---

一、发送者的可靠性

目的：确保生产者一定能把消息发到MQ。

1. 生产者重试机制

对于第一种情况，即生产者发送消息时，出现网络故障，导致连接MQ失败。

• 为了解决这个问题，SpringAMQP提供的消息发送时的重试机制。即：当RabbitTemplate与MQ连接超时后多次重试。
• 当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。
• 如果对于业务性能有要求，建议禁用重试机制。
• 如果一定要使用，需要合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。

2. 生产者确认机制

对于其它三种情况，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机制。
• 默认两种机制都是关闭状态，需要通过配置文件来开启。

3. 实现生产者确认

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，回调机制。

3.2. 定义ReturnCallback

每个RabbitTemplate只能配置一个ReturnCallback，因此可以在配置类中统一设置。

3.3. 定义ConfirmCallback

由于每个消息发送时的处理逻辑不一定相同，因此ConfirmCallback需要在每次发消息时定义。

开启生产者确认比较消耗MQ性能，一般不建议开启。而且大家思考一下触发确认的几种情况：

• 路由失败：一般是因为RoutingKey错误导致，往往是编程导致
• 交换机名称错误：同样是编程错误导致
• MQ内部故障：这种需要处理，但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启，而且仅仅需要开启ConfirmCallback处理nack就可以了。

二、MQ的可靠性

消息到达MQ以后，如果MQ不能及时保存，也会导致消息丢失，所以MQ的可靠性也非常重要。

1. 数据持久化

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

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

2. LazyQueue

在默认情况下，RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下，这会导致消息积压，比如：

• 消费者宕机或出现网络故障
• 消息发送量激增，超过了消费者处理速度
• 消费者处理业务发生阻塞

一旦出现消息堆积问题，RabbitMQ的内存占用就会越来越高，直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上，这个行为成为PageOut. PageOut会耗费一段时间，并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息，生产者的所有请求都会被阻塞。

为了解决这个问题，从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的模式，也就是惰性队列。惰性队列的特征如下：

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

---