消息可靠性

在发送消息到接收消息的过程中，每一步都会造成消息的丢失

丢失的原因：一.发送时丢失：
              1). 生产者发送的消息未送达exchange（交换机名称错误）
               2). 消息到达exchange后未到达queue（消息到达交换机，路由写错，没有到
              达队列）
           二.MQ宕机，queue将消息丢失
           三.consumer接收到消息后未消费就宕机
RabbitMQ分别给出了解决方案:
            一.- 生产者确认机制
            二.- mq持久化
            三.- 消费者确认机制
            四.- 失败重试机制

发送者消息确认

publisher confirm机制;作用:避免消息发送到MQ过程中丢失
原理：给每个消息指定一个唯一ID，消息发送到MQ以后，会返回一个结果给
发送者，表示消息是否处理成功。

返回结果的两种形式

publisher-confirm
ack:消息成功投递到交换机，返回ack
nack:消息未投递到交换机，返回nack

消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。

步骤

1.修改yml配置

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

参数说明

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

1.2定义ReturnConfirm 回调

每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目加载时
配置：
修改publisher服务，添加一个：

package cn.itcast.mq.config;
@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());
            // 如果有业务需要，可以重发消息
        });
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * @param correlationData  自定义的数据
             * @param ack  是否确认
             * @param cause  原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    if(ack){
                        // 3.1.ack，消息成功
                        log.debug("消息发送成功, ID:{}", correlationData.getId());
                    }else{
                        // 3.2.nack，消息失败
                        log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), cause);
                    }
            }
        });
    }
     @Bean
    public DirectExchange simpleExchange(){
        // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
        return new DirectExchange("simple.direct", false, false);
    }
    @Bean
    public Queue simpleQueue(){
        return new Queue("simple.queue",false);
    }
    @Bean
    public Binding binding(){
        return BindingBuilder.bind(simpleQueue()).to(simpleExchange()).with("simple");
    }
}

1.3发送消息测试

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());
    // 4.发送消息
    rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);
    // 休眠一会儿，等待ack回执
    Thread.sleep(2000);
}

发送者回执

- a.设置不存在的交换机尝试发送     交换机: task.direct         路由: task
- 结果:  发送确认回调返回false消息没有正确发送到MQ中
-           return回调未触发
- b.设置存在的交换机,不存在的路由尝试发送     交换机: simple.direct         路由: task
- 结果:  发送确认回调返回true消息已经发送到MQ中
-           return回调触发，返回了消息，并提示路由错误
- c.设置正确的交换机，正确的路由  交换机: simple.direct         路由: simple
- 结果: 发送确认回调返回true消息已经发送到MQ中
-            return回调未触发

通过发送确认 和 消息返还机制可以确保消息 一定能够投递到指定的队列中，如果消息没有投递成功 
或返还了 

也可以通过定时重新投递的方式进行补偿

消息持久化

如果在传输过程中，mq突然宕机，也会造成消息丢失
想要消息持久化保存，需要开启消息持久化机制
- 交换机持久化
- 队列持久化
- 消息持久化

交换机持久化

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

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

事实上，默认情况下，由SpringAMQP声明的交换机都是持久化的
RabbitMQ控制台看到持久化的交换机都会带上D的标示

队列持久化

RabbitMQ中队列如果设置成非持久化的，mq重启后就丢失。

SpringAMQP中可以通过代码指定交换机持久化：
    @Bean
    public Queue simpleQueue(){
        return new Queue("simple.queue",true);
    }

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

消息持久化

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

消费者消息确认

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假定消费者获取消息后会成功处理，因此消息投递后立即被删除

三种模式对比（一般默认auto）

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

none模式

1.在接收者yml文件中添加内容

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

2.修改监听类的方法 模拟一个消息处理异常：

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
    log.info("消费者接收到simple.queue的消息：【{}】", msg);
    // 模拟异常
    System.out.println(1 / 0);
    log.debug("消息处理完成！");
}

当消息处理抛异常时，消息依然被RabbitMQ删除了。

auto模式

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto # 关闭ack

消费失败重试机制

当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，
带来不必要的压力：

本地重试

我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。

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

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000ms # 初识的失败等待时长为1秒
          multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false
重试完成之后，消息又丢失了

失败策略

在之前的测试中，达到最大重试次数后，消息会被丢弃，这是由Spring内部机制决定的。

在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecovery接口来处理，
它包含三种不同的实现：
- RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式
- ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
- RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机
比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，
专门存放异常消息的队列，后续由人工集中处理。

1）在consumer服务中定义处理失败消息的交换机和队列

@Bean
public DirectExchange errorMessageExchange(){
    return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){
    return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
    return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

2）定义一个RepublishMessageRecoverer，关联队列和交换机

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

可以看到失败的消息转发到erro.queue上
如何保证RabbitMQ消息的可靠性？

如何确保RabbitMQ消息的可靠性？

- 开启生产者确认机制，确保生产者的消息能到达队列
- 开启持久化功能，确保消息未消费前在队列中不会丢失
- 开启消费者确认机制为auto，由spring确认消息处理成功后完成ack
- 开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工
处理

死信交换机

概念

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

- 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
- 消息是一个过期消息，超时无人消费
- 要投递的队列消息满了，无法投递

如果这个包含死信的队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，检查DLX）。


另外，队列将死信投递给死信交换机时，必须知道两个信息：

• 死信交换机名称
• 死信交换机与死信队列绑定的RoutingKey

这样才能确保投递的消息能到达死信交换机，并且正确的路由到死信队列。

利用死信交换机接收死信（拓展）

在失败重试策略中，默认的RejectAndDontRequeueRecoverer会在本地重试次数耗尽后，发送reject给RabbitMQ，消息变成死信，被丢弃。
1.在consumer中CommonConfig 修改消息策略

      // 修改 失败消息策略
    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
//        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
        return new RejectAndDontRequeueRecoverer();
    }

我们可以给simple.queue添加一个死信交换机，给死信交换机绑定一个队列。这样消息变成死信后也不会丢弃，而是最终投递到死信交换机，路由到与死信交换机绑定的队列。

2.我们在producer服务CommonConfig中，定义一组死信交换机、死信队列：

    @Bean
    public Queue simpleQueue(){
        return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化
                .deadLetterExchange("dl.direct") // 指定死信交换机
                .build();
    }
    // 声明死信交换机 dl.direct
    @Bean
    public DirectExchange dlExchange(){
        return new DirectExchange("dl.direct", true, false);
    }
    // 声明存储死信的队列 dl.queue
    @Bean
    public Queue dlQueue(){
        return new Queue("dl.queue", true);
    }
    // 将死信队列 与 死信交换机绑定
    @Bean
    public Binding dlBinding(){
        return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("dl");
    }

在创建时会报错，因为有同名队列
总结：

什么样的消息会成为死信？

- 消息被消费者reject或者返回nack
- 消息超时未消费
- 队列满了

死信交换机的使用场景是什么？

- 如果队列绑定了死信交换机，死信会投递到死信交换机；
- 可以利用死信交换机收集所有消费者处理失败的消息（死信），交由人工处理，进一步提高消息队列的可靠性。

TTL

消息超时的两种情况

• 消息所在的队列设置了超时时间
• 消息本身设置了超时时间

接收超时死信的死信交换机

在consumer服务的SpringRabbitListener中，定义一个新的消费者，并且声明 死信交换机、死信队列：

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "dl.ttl.queue", durable = "true"),
    exchange = @Exchange(name = "dl.ttl.direct"),
    key = "ttl"
))
public void listenDlQueue(String msg){
    log.info("接收到 dl.ttl.queue的延迟消息：{}", msg);
}

声明一个队列，并且指定TTL

@Bean
public Queue ttlQueue(){
    return QueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化
        .ttl(10000) // 设置队列的超时时间，10秒
        .deadLetterExchange("dl.ttl.direct") // 指定死信交换机
        .build();
}

注意，这个队列设定了死信交换机为dl.ttl.direct

声明交换机，将ttl与交换机绑定：

@Bean
public DirectExchange ttlExchange(){
    return new DirectExchange("ttl.direct");
}
@Bean
public Binding ttlBinding(){
    return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

发送消息，但是不要指定TTL：

@Test
public void testTTLQueue() {
    // 创建消息
    String message = "hello, ttl queue";
    // 消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 发送消息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    // 记录日志
    log.debug("发送消息成功");
}