十、发布确认高级(SpringBoot版)

在生产环境中由于一些不明原因，导致 RabbitMQ 重启，在 RabbitMQ 重启期间生产者消息投递失败， 导致消息丢失，需要手动处理和恢复。

于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？
image.png

1、ConfirmCallback 发布端到交换机的回调

ConfirmCallback 是一个回调接口，消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中。
我们需要在生产者的配置中添加下面配置，表示开启发布者确认。

  1. spring.rabbitmq.publisher-confirm-type=correlated     #新版本
  2. spring.rabbitmq.publisher-confirms=true                         #老版本

确认机制方案

5_发布确认高级 - 图2

代码架构图

5_发布确认高级 - 图3

配置文件

  1. spring.rabbitmq.host=*.*.*.*
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=admin
  4. spring.rabbitmq.password=123456
  5. spring.rabbitmq.publisher-confirm-type=correlated    #确认消息已发送到交换机(Exchange)
  • NONE 禁用发布确认模式，是默认值
  • CORRELATED 发布消息成功到交换器后会触发回调方法
  • SIMPLE 经测试有两种效果，
    • 其一效果和 CORRELATED 值一样会触发回调方法，
    • 其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker;

添加配置类

  1. package com.atguigu.rabbitmq.springbootrabbitmq.config;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. /**
  9.  * @author: like
  10.  * @Date: 2021/07/20 8:07
  11.  */
  12. @Configuration
  13. public class ConfirmConfig {
  15.     //交换机
  16.     public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
  17.     //队列
  18.     public static final String CONFIRM_QUEUE_NAME = "confirm.queue";
  20.     //RoutingKey
  21.     public static final String CONFIRM_ROUTING_KEY = "key1";
  24.     //声明交换机
  25.     @Bean("confirmExchange")
  26.     public DirectExchange confirmExchange() {
  27.         return new DirectExchange(CONFIRM_EXCHANGE_NAME);
  28.     }
  30.     //声明队列
  31.     @Bean("confirmQueue")
  32.     public Queue confirmQueue() {
  33.         return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
  34.     }
  36.     //绑定
  37.     @Bean
  38.     public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
  39.                                         @Qualifier("confirmExchange") DirectExchange confirmExchange) {
  40.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
  41.     }
  43. }

消息生产者

  1. package com.atguigu.rabbitmq.springbootrabbitmq.controller;
  3. import com.atguigu.rabbitmq.springbootrabbitmq.config.ConfirmConfig;
  4. import com.atguigu.rabbitmq.springbootrabbitmq.config.MyCallBack;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.amqp.rabbit.connection.CorrelationData;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.web.bind.annotation.GetMapping;
  10. import org.springframework.web.bind.annotation.PathVariable;
  11. import org.springframework.web.bind.annotation.RequestMapping;
  12. import org.springframework.web.bind.annotation.RestController;
  14. import javax.annotation.PostConstruct;
  16. /**
  17.  * @author: like
  18.  * @Date: 2021/07/20 8:20
  19.  */
  20. @Slf4j
  21. @RestController
  22. @RequestMapping("/confirm")
  23. public class ProducerController {
  25.     @Autowired
  26.     private RabbitTemplate rabbitTemplate;
  28.     @Autowired
  29.     private MyCallBack myCallBack;
  31.     //依赖注入 rabbitTemplate 之后再设置它的回调对象
  32.     @PostConstruct
  33.     public void init() {
  34.         //注入
  35.         rabbitTemplate.setConfirmCallback(myCallBack);
  36.     }
  39.     /**
  40.      * 发消息 消息回调和退回
  41.      *
  42.      * @param message
  43.      */
  44.     @GetMapping("/sendMessage/{message}")
  45.     public void sendMsg(@PathVariable String message) {
  46.         //指定消息 id 为 1
  47.         CorrelationData correlationData1 = new CorrelationData("1");
  48.         rabbitTemplate.convertAndSend(
  49.                 ConfirmConfig.CONFIRM_EXCHANGE_NAME,
  50.                 ConfirmConfig.CONFIRM_ROUTING_KEY,
  51.                 message + "key1",
  52.                 correlationData1);
  53.         log.info("发送消息内容:{}", message + "key1");
  55.         //指定消息 id 为 2
  56.         CorrelationData correlationData2 = new CorrelationData("2");
  57.         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,
  58.                 ConfirmConfig.CONFIRM_ROUTING_KEY + "2",    //故意写错RoutingKey
  59.                 message + "key2",
  60.                 correlationData2);
  61.         log.info("发送消息内容:{}", message + "key2");
  62.     }
  65. }

回调接口

交换机如果收不到消息，可以进行回调接口

  1. package com.atguigu.rabbitmq.springbootrabbitmq.config;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.rabbit.connection.CorrelationData;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.stereotype.Component;
  9. @Component
  10. @Slf4j
  11. public class MyCallBack implements RabbitTemplate.ConfirmCallback {
  13.     /**
  14.      * 交换机不管是否收到消息 都回调
  15.      *
  16.      * @param correlationData 保存回调消息的ID及相关信息
  17.      * @param ack             交换机是否收到消息 true false
  18.      * @param cause           未收到消息的原因
  19.      */
  20.     @Override
  21.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
  22.         String id = correlationData != null ? correlationData.getId() : "";
  24.         if (ack) {
  25.             log.info("交换机已经收到ID为：{}的消息", id);
  26.         } else {
  27.             log.info("交换机还未收到ID为：{}的消息，由于原因：{}", id, cause);
  28.         }
  30.     }
  32. }

消息消费者

  1. package com.atguigu.rabbitmq.springbootrabbitmq.consumer;
  3. import com.atguigu.rabbitmq.springbootrabbitmq.config.ConfirmConfig;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  9. /**
  10.  * @author: like
  11.  * @Date: 2021/07/20 8:23
  12.  */
  13. @Component
  14. @Slf4j
  15. public class ConfirmConsumer {
  17.     @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
  18.     public void receiveConfirmMessage(Message message) {
  20.         String msg = new String(message.getBody());
  21.         log.info("接收到队列confirm.queue消息：{}", msg);
  22.     }
  25. }

访问： http://localhost:8080/confirm/sendMessage/大家好

结果分析

5_发布确认高级 - 图4

可以看到，发送了两条消息，第一条消息的 RoutingKey 为 “key1”，第二条消息的 RoutingKey 为 “key2”，两条消息都成功被交换机接收，也收到了交换机的确认回调，

但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。

丢弃的消息交换机是不知道的，需要解决告诉生产者消息传送失败

2、回退消息 ReturnCallback 交换机到队列的回调

通过实现 ReturnCallback 接口，启动消息失败返回，此接口是在交换器路由不到队列时触发回调，该方法可以不使用，因为交换器和队列是在代码里绑定的，如果消息成功投递到 Broker 后几乎不存在绑定队列失败，除非你代码写错了。

使用此接口需要在生产者配置中加入一下配置，表示发布者返回。

  1. spring.rabbitmq.publisher-returns=true  #确认消息已发送到队列(Queue)

Mandatory 参数

  1. rabbitTemplate.setMandatory(true);
  2. rabbitTemplate.setReturnsCallback(myCallBack);

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。

那么如何让无法被路由的消息帮我想办法处理一下？最起码通知我一声，我好自己处理啊。

通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

修改配置

  1. #消息退回
  2. spring.rabbitmq.publisher-returns=true

修改回调接口

  1. package com.atguigu.rabbitmq.springbootrabbitmq.config;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.rabbit.connection.CorrelationData;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.stereotype.Component;
  10. @Component
  11. @Slf4j
  12. public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
  14.     /**
  15.      * 交换机确认回调方法
  16.      * <p>
  17.      * 不管是否收到消息 都回调
  18.      *
  19.      * @param correlationData 保存回调消息的ID及相关信息
  20.      * @param ack             交换机是否收到消息 true false
  21.      * @param cause           未收到消息的原因
  22.      */
  23.     @Override
  24.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
  25.         String id = correlationData != null ? correlationData.getId() : "";
  27.         if (ack) {
  28.             log.info("交换机已经收到ID为：{}的消息", id);
  29.         } else {
  30.             log.info("交换机还未收到ID为：{}的消息，由于原因：{}", id, cause);
  31.         }
  33.     }
  35.     //当消息无法路由的时候的回调方法
  36.     //只有不可达目的地的时候 才进行回退
  37.     @Override
  38.     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
  39.         log.error("消息{}，被交换机{}退回，退回原因：{}，路由key为：{}",
  40.                 new String(message.getBody()),
  41.                 exchange,
  42.                 replyText,
  43.                 routingKey);
  44.     }
  47. }

修改消息生产者

  1. //依赖注入 rabbitTemplate 之后再设置它的回调对象
  2. @PostConstruct
  3. public void init() {
  4.     //注入
  5.     rabbitTemplate.setConfirmCallback(myCallBack);
  6.     /**
  7.      * true：交换机无法将消息进行路由时，会将该消息返回给生产者
  8.      * false：如果发现消息无法进行路由，则直接丢弃
  9.      */
  10.     rabbitTemplate.setMandatory(true);
  11.     //设置回退消息交给谁处理
  12.     rabbitTemplate.setReturnCallback(myCallBack);
  13. }

访问： http://localhost:8080/confirm/sendMessage/大家好

结果分析

5_发布确认高级 - 图5

3、备份交换机

有了 mandatory 参数和回退消息，我们获得了对无法投递消息的感知能力，有机会在生产者的消息无法被投递时发现并处理。

但有时候，我们并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。

而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。

而且设置 mandatory 参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。

如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢？

前面在设置死信队列的文章中，我们提到，可以为队列设置死信交换机来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。

在 RabbitMQ 中，有一种备份交换机的机制存在，可以很好的应对这个问题。

什么是备份交换机呢？备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，

当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，

通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。

当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。

代码架构图

5_发布确认高级 - 图6

修改配置类

  1. package com.atguigu.rabbitmq.springbootrabbitmq.config;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. /**
  9.  * @author: like
  10.  * @Date: 2021/07/20 8:07
  11.  */
  12. @Configuration
  13. public class ConfirmConfig {
  15.     //交换机
  16.     public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";
  17.     //队列
  18.     public static final String CONFIRM_QUEUE_NAME = "confirm.queue";
  19.     //RoutingKey
  20.     public static final String CONFIRM_ROUTING_KEY = "key1";
  22.     ///////////////////////////关于备份的
  24.     //备份交换机
  25.     public static final String BACKUP_EXCHANGE_NAME = "backup.exchange";
  26.     //备份队列
  27.     public static final String BACKUP_QUEUE_NAME = "backup.queue";
  28.     //报警队列
  29.     public static final String WARNING_QUEUE_NAME = "warning.queue";
  33. //    //声明交换机 Exchange
  34. //    @Bean("confirmExchange")
  35. //    public DirectExchange confirmExchange() {
  36. //        return new DirectExchange(CONFIRM_EXCHANGE_NAME);
  37. //    }
  39.     //声明确认 Exchange 交换机的备份交换机
  40.     @Bean("confirmExchange")
  41.     public DirectExchange confirmExchange() {
  42.       return  ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME)
  43.                 .durable(true)
  44.                 //设置该交换机的备份交换机
  45.                 .withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME)
  46.                 .build();
  47.     }
  49.     //声明确认队列
  50.     @Bean("confirmQueue")
  51.     public Queue confirmQueue() {
  52.         return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
  53.     }
  55.     //声明确认队列绑定关系
  56.     @Bean
  57.     public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
  58.                                         @Qualifier("confirmExchange") DirectExchange confirmExchange) {
  59.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
  60.     }
  62.     //************************以下是关于备份的******************************
  64.     //备份交换机
  65.     @Bean("backupExchange")
  66.     public FanoutExchange backupExchange() {
  67.         return new FanoutExchange(BACKUP_EXCHANGE_NAME);
  68.     }
  70.     //备份队列
  71.     @Bean("backupQueue")
  72.     public Queue backupQueue() {
  73.         return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
  74.     }
  76.     // 声明备份队列绑定关系
  77.     @Bean
  78.     public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
  79.                                                     @Qualifier("backupExchange") FanoutExchange backupExchange) {
  80.         return BindingBuilder.bind(backupQueue).to(backupExchange);
  81.     }
  83.     //报警队列
  84.     @Bean("warningQueue")
  85.     public Queue warningQueue() {
  86.         return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
  87.     }
  89.     // 声明报警队列绑定关系
  90.     @Bean
  91.     public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue warningQueue,
  92.                                                      @Qualifier("backupExchange") FanoutExchange backupExchange) {
  93.         return BindingBuilder.bind(warningQueue).to(backupExchange);
  94.     }
  96. }

报警消费者

  1. package com.atguigu.rabbitmq.springbootrabbitmq.consumer;
  3. import com.atguigu.rabbitmq.springbootrabbitmq.config.ConfirmConfig;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  9. /**
  10.  * @author: like
  11.  * @Date: 2021/07/21 8:14
  12.  */
  13. @Component
  14. @Slf4j
  15. public class WarningConsumer {
  17.     //接收报警消息
  18.     @RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)
  19.     public void receiveWarningMsg(Message message){
  20.         String msg = new String(message.getBody());
  21.         log.error("报警发现不可路由消息：{}",msg);
  22.     }
  24. }

测试注意事项

之前已写过 confirm.exchange 交换机，由于更改配置，需要删掉，不然会报错

5_发布确认高级 - 图7

结果分析

访问： http://localhost:8080/confirm/sendMessage/大家好

5_发布确认高级 - 图8

mandatory 参数（回退消息）与备份交换机可以一起使用的时候，如果两者同时开启，消息究竟何去何从，谁优先级高？

经过上面结果显示答案是备份交换机优先级高。