RabbitMQ 进阶1（发送者和MQ的可靠性）

我们可以通过 MQ 异步调用，来使程序的性能更好和解耦合。但是如果 MQ 的消息没有成功的被对应的程序处理，那么这样不就会造成数据不一致的情况。因此，我们这里必须要尽可能的确保 MQ 消息的可靠性，即：消息应该至少被消费者处理一次。

那么问题来了：

我们该如何确保 MQ 消息的可靠性？

如果真的发送失败，有没有其它的兜底方案？

相信本篇博客能给你答案。

一、发送者的可靠性

首先，我们一起分析一下消息丢失的可能性有哪些。

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

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

发送消息时丢失：

生产者发送消息时连接 MQ 失败。

生产者发送消息到达 MQ 后未找到Exchange

生产者发送消息到达 MQ 的Exchange后，未找到合适的Queue

MQ 导致消息丢失：

消息到达 MQ，保存到队列后，尚未消费就突然宕机

消费者处理消息时：

消息接收后尚未处理突然宕机

消息接收后处理过程中抛出异常

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

确保生产者一定把消息发送到 MQ

确保 MQ 不会将消息弄丢

确保消费者一定要成功处理消息

我们先来看如何确保生产者一定能把消息发送到 MQ。

1.1 生产者重试机制：

首先第一种情况，就是生产者发送消息时，出现了网络故障，导致与 MQ 的连接中断。

为了解决这个问题，SpringAMQP 提供的消息发送时的重试机制。即：当RabbitTemplate与 MQ 连接超时后，多次重试。

修改publisher模块的application.yaml文件，添加下面的内容：

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

注意：当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过 SpringAMQP 提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的。

如果对于业务性能有要求，建议禁用重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码。

如果是断网的情况下，整个业务都会被影响，我们可以很容易的发现问题所在，并进行解决，不过断网一般是不太会出现。

1.2 生产者确认机制：

一般情况下，只要生产者与 MQ 之间的网路连接顺畅，基本不会出现发送消息丢失的情况，因此大多数情况下我们无需考虑这种问题。

不过，在少数情况下，也会出现消息发送到 MQ 之后丢失的现象，比如：

MQ 内部处理消息的进程发生了异常

生产者发送消息到达 MQ 后未找到Exchange

生产者发送消息到达 MQ 的Exchange后，未找到合适的Queue，因此无法路由

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

具体如图所示：

总结如下：

当消息投递到 MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回 ack 的确认信息（因为这里是程序员自己代码写错了，如果返回 NACK 后面生产者会继续投递该消息，但是代码错误，怎么投都不会成功），代表投递成功

临时消息投递到了 MQ，并且入队成功，返回 ACK，告知投递成功

持久消息投递到了 MQ，并且入队完成持久化，返回 ACK ，告知投递成功

其它情况都会返回 NACK，告知投递失败

其中ack和nack属于Publisher Confirm机制，ack是投递成功；nack是投递失败。而return则属于Publisher Return机制。

默认两种机制都是关闭状态，需要通过配置文件来开启。

两个机制一般配合是配合使用。

1.2.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.2.2 定义 ReturnCallback：

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

内容如下：

package com.itheima.publisher.config;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {
    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.error("触发return callback,");
                log.debug("exchange: {}", returned.getExchange());
                log.debug("routingKey: {}", returned.getRoutingKey());
                log.debug("message: {}", returned.getMessage());
                log.debug("replyCode: {}", returned.getReplyCode());
                log.debug("replyText: {}", returned.getReplyText());
            }
        });
    }
}

1.2.3 定义 ConfirmCallback：

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

这里的 CorrelationData 中包含两个核心的东西：

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

SettableListenableFuture：回执结果的 Future 对象

将来 MQ 的回执就会通过这个Future来返回，我们可以提前给CorrelationData中的Future添加回调函数来处理消息回执：

下面我们来演示一下：

@Test
    public void testConfirmRollback() throws InterruptedException {
        // 1.设置唯一 id
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
        // 2.添加回调函数
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                // Future 本身发生异常的时处理的逻辑，一般不会发生 
                log.debug("send message fail", ex);
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                if (result.isAck()) {
                    log.debug("发送消息成功，收到 ack！");
                } else {
                    log.error("发送消息失败，收到 nack，reason: {}", result.getReason());
                    // 进行消息的重发，这里没有实现
                }

            }
        });
        // 2.发送信息给 rabbit mq
        rabbitTemplate.convertAndSend("hmall.direct", "blue", "hello", cd);
    }

注意：

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

路由失败：一般是因为 RoutingKey 错误导致，往往是编程导致

交换机名称错误：同样是编程错误导致

MQ 内部故障：这种需要处理，但概率往往较低。因此只有对消息可靠性要求非常高的业务才需要开启，而且仅仅需要开启 ConfirmCallback 处理 nack 就可以了。

二、MQ 的可靠性

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

2.1 数据持久化：

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

交换机持久化

队列持久化

消息持久化

我们下面以 Java 代码来进行演示：

2.1.1 交换机持久化：

下面演示 FanoutExchange 其它的交换机也都是类似的。

使第二个参数 durable 为 true。

@Bean
public FanoutExchange fanoutExchange(){
    // 第一个参数 name 表示交换机名称
    // 第二个参数 durable 表示是否持久化
    // 第三个参数 autoDelete 表示是否自动删除
    return new FanoutExchange("hmall.fanout", true, false);
}

2.1.2 队列持久化：

使第二个参数 durable 为 true。

@Bean
public Queue fanoutQueue(){
    return new Queue("fanout.queue", true);
}

2.1.3 消息持久化：

对应的 API 如下：

// 发送持久化的消息给 rabbit mq
rabbitTemplate.convertAndSend("队列名", "路由名", "消息内容",
        message -> {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        });

说明：在开启持久化机制以后，如果同时还开启了生产者确认，那么 MQ 会在消息持久化以后才发送 ACK 回执，进一步确保消息的可靠性。

不过出于性能考虑，为了减少 IO 次数，发送到 MQ 的消息并不是逐条持久化到数据库的，而是每隔一段时间批量持久化。一般间隔在100 毫秒左右，这就会导致 ACK 有一定的延迟，因此建议生产者确认全部采用异步方式。

2.2 LazyQueue：

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

消费者宕机或出现网络故障

消息发送量激增，超过了消费者处理速度

消费者处理业务发生阻塞

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

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

接收到消息后直接存入磁盘而非内存

消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载）

支持数百万条的消息存储

而在 3.12 版本之后，LazyQueue 已经成为所有队列的默认格式。因此官方推荐升级 MQ 为 3.12 版本或者所有队列都设置为 LazyQueue 模式。

2.2.1 控制台配置 Lazy 模式：

在添加队列的时候，添加x-queue-mod=lazy参数即可设置队列为 Lazy 模式：

2.2.2 代码配置 Lazy 模式：

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

@Bean
public Queue lazyQueue(){
    return QueueBuilder
            .durable("lazy.queue") // 表示创建一个名为 lazy.queue 的持久化队列
            .lazy() // 开启Lazy模式
            .build();
}

这里是通过QueueBuilder的lazy()函数配置Lazy模式，底层源码如下：

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

@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){
    log.info("接收到 lazy.queue的消息：{}", msg);
}