a1. 消息可靠性

消息队列在使用过程中，面临着很多实际问题需要思考：

消息从发送，到消费者接收，会经理多个过程：

其中的每一步都可能导致消息丢失，常见的丢失原因包括：

发送时丢失：
- 生产者发送的消息未送达 exchange
- 消息到达 exchange 后未到达 queue
MQ 宕机，queue 将消息丢失
consumer 接收到消息后未消费就宕机

针对这些问题，RabbitMQ 分别给出了解决方案：

生产者确认机制
mq 持久化
消费者确认机制
失败重试机制

下面我们就通过案例来演示每一个步骤。

首先，导入课前资料提供的 demo 工程：

项目结构如下：

1.1.生产者消息确认

RabbitMQ 提供了 publisher confirm 机制来避免消息发送到 MQ 过程中丢失。这种机制必须给每个消息指定一个唯一 ID。消息发送到 MQ 以后，会返回一个结果给发送者，表示消息是否处理成功。

返回结果有两种方式：

publisher-confirm，发送者确认
- 消息成功投递到交换机，返回 ack
- 消息未投递到交换机，返回 nack
publisher-return，发送者回执
- 消息投递到交换机了，但是没有路由到队列。返回 ACK，及路由失败原因。

注意：

1.1.1.修改配置

首先，修改 publisher 服务中的 application.yml 文件，添加下面的内容：

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

[](data:image/svg+xml;base64,PHN2ZyBkYXRhLXYtNWRjMzE1NjggeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIiB3aWR0aD0iMjQiIGhlaWdodD0iMjQiIHZpZXdib3g9IjAgMCAyNCAyNCIgY2xhc3M9ImhvdmVyIiBzdHlsZT0iYm90dG9tOiA3LjVweDsiPjxwYXRoIGRhdGEtdi01ZGMzMTU2OCBmaWxsPSJub25lIiBkPSJNMCAwaDI0djI0SDB6IiAvPjxwYXRoIGRhdGEtdi01ZGMzMTU2OCBmaWxsPSIjMjdiMWZmIiBkPSJNMTYgMUg0Yy0xLjEgMC0yIC45LTIgMnYxNGgyVjNoMTJWMXptLTEgNGw2IDZ2MTBjMCAxLjEtLjkgMi0yIDJINy45OUM2Ljg5IDIzIDYgMjIuMSA2IDIxbC4wMS0xNGMwLTEuMS44OS0yIDEuOTktMmg3em0tMSA3aDUuNUwxNCA2LjVWMTJ6IiAvPjwvc3ZnPg==)

说明：

publish-confirm-type：开启 publisher-confirm，这里支持两种类型：
- simple：同步等待 confirm 结果，直到超时
- correlated：异步回调，定义 ConfirmCallback，MQ 返回结果时会回调这个 ConfirmCallback
publish-returns：开启 publish-return 功能，同样是基于 callback 机制，不过是定义 ReturnCallback
template.mandatory：定义消息路由失败时的策略。true，则调用 ReturnCallback；false：则直接丢弃消息

1.1.2.定义 Return 回调

每个 RabbitTemplate 只能配置一个 ReturnCallback，因此需要在项目加载时配置：

修改 publisher 服务，添加一个：

package cn.itcast.mq.config;

import lombok.extern.slf4j.Slf4j;
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;

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 获取RabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        // 设置ReturnCallback
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            // 投递失败，记录日志
            log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
                     replyCode, replyText, exchange, routingKey, message.toString());
            // 如果有业务需要，可以重发消息
        });
    }
}

1.1.3.定义 ConfirmCallback

ConfirmCallback 可以在发送消息时指定，因为每个业务处理 confirm 成功或失败的逻辑不一定相同。

在 publisher 服务的 cn.itcast.mq.spring.SpringAmqpTest 类中，定义一个单元测试方法：

public void testSendMessage2SimpleQueue() throws InterruptedException {
    // 1.消息体
    String message = "hello, spring amqp!";
    // 2.全局唯一的消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 3.添加callback
    correlationData.getFuture().addCallback(
        result -> {
            if(result.isAck()){
                // 3.1.ack，消息成功
                log.debug("消息发送成功, ID:{}", correlationData.getId());
            }else{
                // 3.2.nack，消息失败
                log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
            }
        },
        ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 4.发送消息
    rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);

    // 休眠一会儿，等待ack回执
    Thread.sleep(2000);
}

1.2.消息持久化

生产者确认可以确保消息投递到 RabbitMQ 的队列中，但是消息发送到 RabbitMQ 以后，如果突然宕机，也可能导致消息丢失。

要想确保消息在 RabbitMQ 中安全保存，必须开启消息持久化机制。

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

1.2.1.交换机持久化

RabbitMQ 中交换机默认是非持久化的，mq 重启后就丢失。

SpringAMQP 中可以通过代码指定交换机持久化：

@Bean
public DirectExchange simpleExchange(){
    // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
    return new DirectExchange("simple.direct", true, false);
}

事实上，默认情况下，由 SpringAMQP 声明的交换机都是持久化的。

可以在 RabbitMQ 控制台看到持久化的交换机都会带上D的标示：

1.2.2.队列持久化

RabbitMQ 中队列默认是非持久化的，mq 重启后就丢失。

SpringAMQP 中可以通过代码指定交换机持久化：

@Bean
public Queue simpleQueue(){
    // 使用QueueBuilder构建队列，durable就是持久化的
    return QueueBuilder.durable("simple.queue").build();
}

事实上，默认情况下，由 SpringAMQP 声明的队列都是持久化的。

可以在 RabbitMQ 控制台看到持久化的队列都会带上D的标示：

1.2.3.消息持久化

利用 SpringAMQP 发送消息时，可以设置消息的属性（MessageProperties），指定 delivery-mode：

1：非持久化
2：持久化

用 java 代码指定：

默认情况下，SpringAMQP 发出的任何消息都是持久化的，不用特意指定。

1.3.消费者消息确认

RabbitMQ 是阅后即焚机制，RabbitMQ 确认消息被消费者消费后会立刻删除。

而 RabbitMQ 是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向 RabbitMQ 发送 ACK 回执，表明自己已经处理消息。

设想这样的场景：

1）RabbitMQ 投递消息给消费者
2）消费者获取消息后，返回 ACK 给 RabbitMQ
3）RabbitMQ 删除消息
4）消费者宕机，消息尚未处理

这样，消息就丢失了。因此消费者返回 ACK 的时机非常重要。

而 SpringAMQP 则允许配置三种确认模式：

manual：手动 ack，需要在业务代码结束后，调用 api 发送 ack。
auto：自动 ack，由 spring 监测 listener 代码是否出现异常，没有异常则返回 ack；抛出异常则返回 nack
none：关闭 ack，MQ 假定消费者获取消息后会成功处理，因此消息投递后立即被删除

由此可知：

none 模式下，消息投递是不可靠的，可能丢失
auto 模式类似事务机制，出现异常时返回 nack，消息回滚到 mq；没有异常，返回 ack
manual：自己根据业务情况，判断什么时候该 ack

一般，我们都是使用默认的 auto 即可。

1.3.1.演示 none 模式

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

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none 关闭ack