RabbitMQ------发布确认高级（消息回调、回退、备份交换机）（八）

以你之姓@ 2024-03-31 11:39 57阅读 0赞

## RabbitMQ------发布确认高级（八） ##

可能由于某些意外情况，导致RabbitMQ重启，在RabbitMQ重启过程中，生产者投递消息失败，导致消息丢失。  
如果才能够保证RabbitMQ的消息可靠性呢？  
可能出现两种问题的情况。  
1.交换机接收不到请求  
2.队列接收不到请求  
解决逻辑：当交换机确认发送消息后，将消息从内存中删除，未能删除的消息，通过定时任务，再次发送给交换机

### 代码 ###

配置类：

生产者：

package com.example.springboot01.controller;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import java.util.Date;
    
    /**
     * 发送延迟
     * 生产者
     */
    @Slf4j
    @RestController
    @RequestMapping("/ttl")
    public class SendMsgController {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
        
        //开始发消息  发布确认
        @GetMapping("/sendConfirmMsg/{message}")
        public void sendConfirmMsg(@PathVariable String message){
        
            log.info("当前时间:{},发送一条消息队列：{}",new Date().toString(),message);
            /**
             * 交换机
             * routingkey
             * message
             */
            rabbitTemplate.convertAndSend("confirm_exchange","key1",message);
        }
    }

消费者：

package com.example.springboot01.consumer;
    
    import com.rabbitmq.client.Channel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    
    /**
     *  消费者
     */
    @Slf4j
    @Component
    public class ConfrimConsumer {
        
    
        //接收消息
        @RabbitListener(queues = "confirm_queue")
        public void receiveConfirm(Message message) throws Exception {
        
            String string = new String(message.getBody(),"UTF-8");
            log.info("当前时间:{},收到确认消息：{}",new Date().toString(),string);
        }
    }

经测试以上模板代码可行。

### 交换机收不到消息，实现回调接口 ###

使用回调接口，前提需要在配置文件中增加配置，spring.rabbitmq.publisher-confirm-type有三种参数设置：  
1.NONE 禁用发布确认模式，是默认值  
2.correlated，发布消息成功到交换机后会触发回调方法（交换机接收成功或失败都会回调）  
3.simple  
它有两种效果，第一种和correlated一样触发回调方法  
第二种：消息在发送成功后使用rabbitTemplete调用waitForConfirms或者waitForConfirmOrDie方法，等待broker节点返回发送结果，根据返回结果判定下一步的逻辑。

#### 注意：waitForConfirmOrDie方法如果返回false，则会关闭channel，则接下来无法发送消息到broker。 ####

spring.rabbitmq.host=182.92.234.71
    spring.rabbitmq.prot=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123
    spring.rabbitmq.publisher-confirm-type=correlated

将发不出去的消息保存下来等等自定义的处理逻辑，都可以通过回调接口自己实现。  
在RabbitTemplate中提供了这样的回调接口，  
![在这里插入图片描述][def7c4b57fe24558bade28b07947abc9.png]  
这个接口是一个函数式接口，接口中只有一个方法，可以用lamdba表达式  
![在这里插入图片描述][b401af1077b842f2b8d07103bac8920e.png]  
如果失败了收不到信息，var2就会为false，并且var将失败原因返回。var1是发送的内容  
如果成功了，var2将会为true。  
实现回调接口，需要创建新的类，对其实现

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    
    @Slf4j
    @Component
    public class MyCallBack implements RabbitTemplate.ConfirmCallback {
        
        //实现类虽然实例化,但是它集成的是内部接口,并且MyCallBack并不在RabbitTemplate里面
        //所以导致RabbitTemplate在掉用自身接口时,根本掉不到MyCallBack
        //因此需要将MyCallBack注入到RabbitTemplate
        @Autowired 
        private RabbitTemplate rabbitTemplate;
        //注入
        //PostConstruct注解会在Component、Autowired注解完成后再执行
        @PostConstruct
        public void init(){
        
            rabbitTemplate.setConfirmCallback(this);
        }
        
        
        /**
         * 交换机确认回调方法    成功和失败都会回调
         * @param correlationData 保存消息的id以及相关信息
         * @param ack  交换机收到为true   失败为false
         * @param cause  成功为null   失败为错误原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
            if(ack){
        
                log.info("交换机已收到消息,ID为{}的消息",correlationData == null ? "":correlationData.getId());
            }else {
        
                log.info("交换机未收到消息,ID为{}的消息,原因为:",correlationData.getId(),cause);
            }
        }
    }

correlationData消息的相关信息，本质上不是交换机的，是由生产者发送的。  
而生产者的发送消息方法：convertAndSend，有很多重载方法。其中就包括传递correlationData的方法。因此需要再生产发送消息时new出该对象

public void convertAndSend(String exchange, String routingKey, Object object, @Nullable CorrelationData correlationData) throws AmqpException {
        
            this.send(exchange, routingKey, this.convertMessageIfNecessary(object), correlationData);
        }

![在这里插入图片描述][756f668629f34323adf6930da34695d6.png]  
生产者代码：

@GetMapping("/sendConfirmMsg/{message}")
        public void sendConfirmMsg(@PathVariable String message){
        
            log.info("当前时间:{},发送一条消息队列：{}",new Date().toString(),message);
    
            /**
             * 交换机
             * routingkey
             * message
             */
            CorrelationData correlationData = new CorrelationData("1");
            rabbitTemplate.convertAndSend("confirm_exchange","key1",message,correlationData);
        }

模拟交换机接受不到消息，可以将交换机的名字改错，  
结论：**将生产者发送消息的队列名改错，就能够再控制台看到，接收失败时的回调函数处理方式**。  
但是这种，方式只能解决交换机接收不到消息的情况。

### 队列收不到消息，回退消息 ###

#### Mandatory参数 ####

**在仅开启生产者确认机制的情况下，交换机接收到消息后，会直接将消息生产发送确认消息，如果发现发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃的事件的。**  
通过设置mandatory参数可以当消息传递过程中不可达目的时将消息返回给生产者。  
同样也有退回消息的配置。

spring.rabbitmq.publisher-returns=true

开启配置后，修改config代码，实现回退接口ReturnCallback  
**同时需要注入**

@Slf4j
    @Component
    public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
        
        //实现类虽然实例化,但是它集成的是内部接口,并且MyCallBack并不在RabbitTemplate里面
        //所以导致RabbitTemplate在掉用自身接口时,根本掉不到MyCallBack
        //因此需要将MyCallBack注入到RabbitTemplate
        @Autowired
        private RabbitTemplate rabbitTemplate;
        //注入
        //PostConstruct注解会在Component、Autowired注解完成后再执行
        @PostConstruct
        public void init(){
        
            rabbitTemplate.setConfirmCallback(this);
            //注入
            rabbitTemplate.setReturnCallback(this);
        }
    
        
    
        /**
         * 交换机确认回调方法    成功和失败都会回调
         * @param correlationData 保存消息的id以及相关信息
         * @param ack  交换机收到为true   失败为false
         * @param cause  成功为null   失败为错误原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
            if(ack){
        
                log.info("交换机已收到消息,ID为{}的消息",correlationData == null ? "":correlationData.getId());
            }else {
        
                log.info("交换机未收到消息,ID为{}的消息,原因为:",correlationData.getId(),cause);
            }
        }
    
        /**
         * 队列回退接口  当消息不能到目的地时会返回给生产者
         * 只有不可达目的地时,才进行回退
         * @param message   消息主体
         * @param replyCode  退回码值
         * @param replyText  退回原因
         * @param exchange   交换机
         * @param routingKey  路由key
         */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        
            log.error("消息{},被交换机{}退回,退回原因{},路由key{}",new String(message.getBody()),exchange,replyText,routingKey);
        }
    }

**修改生产者代码，将routingkey改为key2，队列接收不到，触发回退接口，打印出日志。**

### 备份交换机 ###

当交换机接收到一条不可路由的消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为Fanout，这样就能把所有消息都投递到预期绑定的队列。当然也可以建立一个报警队列，用独立的消费者来进行监测和报警。  
示例代码：消费者，主交换机，主队列，备份交换机，备份队列，报警队列，  
修改配置文件  
**1.声明备份交换机，fanout类型  
2.声明备份队列  
3.声明报警队列  
4.备份队列与备份交换机绑定  
5.报警队列与备份交换机绑定  
6.修改主交换机，配置参数，主交换机发送消息失败后，将消息转发到备份交换机，并且需要设置持久化**  
ExchangeBuilder.directExchange(CONFIRM\_EXCHANGE\_NAME).durable(true)  
.withArgument(“alternate-exchange”,BACKUP\_EXCHANGE\_NAME).build();

/**
     * 配置类 发布确认 高级
     */
    @Configuration
    public class ConfirmConfig {
        
    
        //交换机
        public static final  String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
        //队列
        public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
        //Routingkey
        public static final String CONFIRM_ROUTING_KEY = "key1";
    
        //备份交换机
        public static final  String BACKUP_EXCHANGE_NAME = "backup_exchange";
        //备份队列
        public static final String BACKUP_QUEUE_NAME = "backup_queue";
        //报警队列
        public static final String WARING_QUEUE_NAME = "warning_queue";
    
        //声明交换机
        @Bean("confirmExchange")
        public DirectExchange confirmExchange(){
        
            return  ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
                    .withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
        }
    
        //声明队列
        @Bean("confirmQueue")
        public Queue confirmQueue(){
        
            //创建队列的两种方式
            //QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
            return  new Queue(CONFIRM_QUEUE_NAME);
        }
    
        //绑定
        @Bean
        public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue ,
                                            @Qualifier("confirmExchange") DirectExchange confirmExchange){
        
            //将队列与交换机绑定
            return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
        }
    
        //声明交换机
        @Bean("backupExchange")
        public FanoutExchange backupExchange(){
        
            return  new FanoutExchange(BACKUP_EXCHANGE_NAME);
        }
    
        //声明备份队列
        @Bean("backupQueue")
        public Queue backupQueue(){
        
            //创建队列的两种方式
            //QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
            return  new Queue(BACKUP_QUEUE_NAME);
        }
    
        //声明报警队列
        @Bean("warningQueue")
        public Queue warningQueue(){
        
            //创建队列的两种方式
            //QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
            return  new Queue(WARING_QUEUE_NAME);
        }
        //绑定  扇出类型不写routingkey
        @Bean
        public Binding backupQueueBindingExchange(@Qualifier("backupQueue") Queue backupQueue ,
                                            @Qualifier("backupExchange") FanoutExchange backupExchange){
        
            //将队列与交换机绑定
            return BindingBuilder.bind(backupQueue).to(backupExchange);
        }
        //绑定  扇出类型不写routingkey
        @Bean
        public Binding warningQueueBindingExchange(@Qualifier("warningQueue") Queue warningQueue ,
                                                  @Qualifier("backupExchange") FanoutExchange backupExchange){
        
            //将队列与交换机绑定
            return BindingBuilder.bind(warningQueue).to(backupExchange);
        }
    
    }

消费者

/**
     * 队列TTL 消费者
     */
    @Slf4j
    @Component
    public class DeadLetterConsumer {
        
    
        //接收消息
        @RabbitListener(queues = "QD")
        public void receiveD(Message message, Channel channel) throws Exception {
        
            String string = new String(message.getBody(),"UTF-8");
            log.info("当前时间:{},收到死信队列的消息：{}",new Date().toString(),string);
    
        }
    
        //接收消息
        @RabbitListener(queues = "confirm_queue")
        public void receiveConfirm(Message message) throws Exception {
        
            String string = new String(message.getBody(),"UTF-8");
            log.info("当前时间:{},收到确认消息：{}",new Date().toString(),string);
        }
    
    
        //接收消息  报警消息
        @RabbitListener(queues = "warning_queue")
        public void Confirm(Message message) throws Exception {
        
            String string = new String(message.getBody(),"UTF-8");
            log.info("当前时间:{},收到报警消息：{}",new Date().toString(),string);
        }
    }

mandatory参数与备份交换机可以一起使用，如果两者同时开启，消息会去哪里？  
**备份交换机的优先级高！**

[def7c4b57fe24558bade28b07947abc9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/1fa2becf8a6f4c6b9fe66bf8f8957b97.png
[b401af1077b842f2b8d07103bac8920e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/e78fb75ac5444924b461184958dfe58d.png
[756f668629f34323adf6930da34695d6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/049701729bfc467086f40d35442b1c33.png