RabbitMQ 中关于发送可靠性问题（失败重试、发送方确认机制）

偏执的太偏执、 2023-09-28 14:36 24阅读 0赞

### 一. RabbitMQ 消息发送机制 ###

1.  RabbitMQ 中的消息发送引入了 Exchange (交换机) 的概念。消息的发送首先到达交换机，然后再根据既定的路由规则，由交换机将消息路由到不用的 Queue (队列) 中，再由不同的消费者去消费。

![format_png][]

1.  大致的流程就是这样，所以要确保消息发送的可靠性，主要从两个方面去确认
    
     *  消息成功到达 Exchange
     *  消息成功到达 Queue
2.  如果能够确认这两步，那么我们就可以认为消息发送成功。
3.  如果这两步中任一步骤出现了问题，那么消息就是没有发送成功。此时我们可能要通过重试等方式去重新发送消息，多次重试之后，如果消息还是不能到达，则可能需要人工介入了。
4.  那么，要确保消息成功发送，我们需要做好三件事：
    
     *  确认消息到达 Exchange
     *  确认消息到达 Queue
     *  开启定时任务，定时投递那些发送失败的消息
5.  上面三个步骤，第三步需要我们自己去实现，前两步 RabbitMQ 都有提供解决的方案。那么，如果确保消息成功到达 RabbitMQ 呢？
    
    (1) 开启事务机制
    
    (2) 发送方确认机制
6.  注意：这是两种不同的方案，不可以同时开启，只能二选其一。如果同时开启，则会报错。

### 二. 开启事务机制 ###

1.  首先需要配置一个事务管理器
    
        /**
         * 事务管理器
         */
        @Configuration
        public class TxConfig {
            @Bean
            PlatformTransactionManager platformTransactionManager(ConnectionFactory connectionFactory) {
                return new RabbitTransactionManager(connectionFactory);
            }
        }
2.  然后在生产者上添加事务注解以及设置通信通道为事务模式。我这里还是单元测试中进行，代码是修改上面已经使用过的 `test01` 进行演示
    
        @Autowired
        	RabbitTemplate rabbitTemplate;
        
        	/**
        	 * 开启事务机制：
        	 * 1.配置事务管理器
        	 * 2.使用注解开启事务
        	 * 3.把 rabbit 中消息通道设置为事务模式
        	 * 
        	 * 	@Transactional	开启事务
        	 */
        	@Test
        	@Transactional
        	public void test01(){
        		//设置消息通道为事务模式
        		rabbitTemplate.setChannelTransacted(true);
        		rabbitTemplate.convertAndSend(RabbitConfig.MY_QUEUE_NAME, "Hello World!");
        		//手动设置一个异常
        		int i = 1 / 0;
        	}
3.  开启事务机制就三步：
    
     *  配置事务管理器
     *  使用 `@Transactional` 注解开启事务
     *  调用 `setChannelTransacted` 方法设置消息通道为事务模式，即设置为 true
4.  当我们开启事务模式之后，RabbitMQ 生产者发送消息会有这样几个步骤：
    
    (1) 客服端发出请求，将通信管道设置为事务模式
    
    (2) 服务端给出回复，同意将通信管道设置为事务模式
    
    (3) 客户端发送消息
    
    (4) 客户端提交事务
    
    (5) 服务端给出响应，确认事务提交
5.  上面那几个步骤中，除了第三步本来就有的，其他四个步骤都是因为开启事务模式后多出来的。所以事实上，事务模式其实效率是有点低的，并非是最佳解决方案。
6.  在实际开发的过程中，一般都是一些高并发的项目才会使用消息中间件，这个时候并发性能尤为重要。

### 三. 发送方确认机制(常用) ###

1.  首先我们需要在 `application.properties` 中配置开启发送方确认机制：
    
        # 消息到达转换器的确认回调
        spring.rabbitmq.publisher-confirm-type=correlated
        # 消息到达队列的回调
        spring.rabbitmq.publisher-returns=true
    
    其中，`spring.rabbitmq.publisher-confirm-type` 这个属性有三个取值：
    
    (1) none：表示禁用发布确认模式，默认就是这个
    
    (2) correlated：表示成功发布消息到交换器后会触发的回调方法
    
    (3) simple：类似 correlated ，并且支持 `waitForConfirms()` 和 `waitForConfirmsOrDie()` 方法的调用
2.  接着需要创建一个配置类，对两个监听进行配置，即 消息是否到达转换器的监听 和 消息是否到达队列的监听
    
        @Configuration
        public class RabbitTemplateConfig implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{
        
            @Autowired
            RabbitTemplate rabbitTemplate;
        
            /**
             * @PostConstruct   当 bean 完成初始化的时候，这个方法就会被调用
             */
            @PostConstruct
            public void init() {
                rabbitTemplate.setConfirmCallback(this);
                rabbitTemplate.setReturnsCallback(this);
            }
        
            /**
             * 如果消息到达了或者没有到达交换机，都会触发该方法
             *
             * @param correlationData
             * @param ack   如果 ack 为 true，表示消息到达了交换机，反之则没有到达
             * @param cause
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack) {
                    System.out.println("成功！消息到达了交换机");
                } else {
                    System.out.println("失败！消息未到达交换机");
                }
            }
            
            /**
             * 消息未到达队列，会触发该方法
             *
             * @param returnedMessage
             */
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                System.out.println("消息未到达队列");
            }
        }
    
    在该配置类中：
    
    (1) `RabbitTemplate.ConfirmCallback` 这个接口是用来确定消息是否到达交换器的
    
    (2) `RabbitTemplate.ReturnsCallback` 这个则是用来确定消息是否到达队列的，未到达队列时会被调用
    
    (3) `init()` 这个方法上加了 `@PostConstruct` 这个注解，即在 bean 完成初始化的时候调用该方法，完成对 RabbitTemplate 的配置
    
    (4) 上面的几步可以理解为： RabbitTemplate 这个类只需要加依赖就可以直接注入进来调用，但这个类的方法还不能够满足我的需求，所以在 RabbitTemplate 中配置两个 Callback
3.  我这里还是在单元测试中进行，也是上面已经使用过的代码 `test03`
    
    (1) 第一次测试：发送一个不存在的交换机
    
        @Autowired
        	RabbitTemplate rabbitTemplate;
        
        	/**
        	 * 在 DirectConfig.MY_DIRECT_EXCHANGE_NAME 交换机的名字中加上双引号，使该参数变成一个字符串
        	 */
        	@Test
        	public void test03(){
        	    rabbitTemplate.convertAndSend("DirectConfig.MY_DIRECT_EXCHANGE_NAME",DirectConfig.MY_DIRECT_QUEUE_NAME_01,"Hello Queue01");
        	}
    
    注意：第一个参数是字符串，不是变量
    
    运行之后会有这样一句日志：

![format_png 1][]

(2) 第二次测试：发送一个不存在的队列

@Autowired
    	RabbitTemplate rabbitTemplate;
    
    	/**
    	 * 在 DirectConfig.MY_DIRECT_QUEUE_NAME_01 队列的名字中加上双引号，使该参数变成一个字符串
    	 */
    	@Test
    	public void test03(){
    	    rabbitTemplate.convertAndSend(DirectConfig.MY_DIRECT_EXCHANGE_NAME,"DirectConfig.MY_DIRECT_QUEUE_NAME_01","Hello Queue01");
    	}

注意：此时的第二个参数是字符串，不是变量

运行之后会有这样一句日志：

![format_png 2][]

1.  如果消息是批量处理，发送成功的回调与监听都是一样的，这里就不演示了。
2.  相比于事务机制，发送方确认机制下的消息吞吐量会得到极大的提升

### 四. 失败重试 ###

#### I. 自带重试机制 ####

1.  在前面的 `事务机制` 和 `发送方确认机制` 都是发送方确认消息发送成功的办法。那么，如果说发送方从一开始就连不上 MQ，那么 Spring Boot 中也有相应的重试机制。
2.  但是呢，这个重试机制就和 MQ 本身是没有关系的，这是利用 Spring 中的 retry 机制来完成的
3.  在 `application.properties` 中进行如下配置：
    
        # 开启重试机制
        spring.rabbitmq.template.retry.enabled=true
        # 最大重试间隔时间
        spring.rabbitmq.template.retry.max-interval=1000ms
        # 最大重试次数
        spring.rabbitmq.template.retry.max-attempts=5
        # 间隔乘数
        spring.rabbitmq.template.retry.multiplier=1.2
        # 初始化的时间间隔
        spring.rabbitmq.template.retry.initial-interval=1000ms
4.  配置完成后，关闭 RabbitMQ ,然后再次启动上面的 `test03`
    
    (1) 这里是 RabbitMQ 的关闭，后面再 `docker start some-rabbit` 启动即可(注意：some-rabbit 是你安装时候取的名字)
    
    ![format_png 3][]
    
    (2) 启动后，就可以看到后台重试打印的日志。因为在配置中设置了最大重试次数为 5 ，所以这里重试了 5 次之后就结束重试了，抛出了异常。
    
    ![format_png 4][]

#### II. 业务重试 ####

1.  业务重试主要是针对消息没有到达交换机的情况
2.  如果消息没有到达交换机，正如上面介绍的，会触发 `RabbitTemplate.ConfirmCallback` 这个方法的回调，那么我们就可以再这个回调中进行处理了。
3.  这种情况大多需要结合自己的业务去处理，这里后面再结合具体的业务去说明吧。

![be0284c7f4804dcba20e10dee7ee9514.jpeg][]

[format_png]: https://img-blog.csdnimg.cn/img_convert/8054a74b6d9045b705e73c2e9977fdd3.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/fa876c7430ba64ca53c527330d93db99.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/cb4b78c4d75e177fe3e0a63eb5becd24.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/3b25b1bb98c61b116c1702e61a98a783.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/5f1f131ca3d85a4001facf19aaa92575.webp?x-oss-process=image/format,png
[be0284c7f4804dcba20e10dee7ee9514.jpeg]: https://img-blog.csdnimg.cn/be0284c7f4804dcba20e10dee7ee9514.jpeg