RabbitMq---高级发布确认

谁借莪１个温暖的怀抱￠ 2023-09-24 14:17 17阅读 0赞

**目录**

一. 解决交换机消息丢失

二.解决队列消息丢失

三.备份队列与告警队列

--------------------

![77c345547dda47caacd7d035a969e3d9.png][]

在生产者发送消息后，可能会发生消息丢失。

（1）交换机因某种原因一直收不到消息，就导致了消息丢失。

（2）交换机收到消息了，但队列因某种原因一直收不到消息，就导致了消息丢失。

## 一. 解决交换机消息丢失 ##

< 1 > 配置文件

#确认消息已发送到交换机(Exchange)
    spring.rabbitmq.publisher-confirm-type=correlated

< 2 > rabbitmq配置类

< 3 > 告警配置类（交换机确认是否收到消息）

该类需要实现 RabbitTemplate.ConfirmCallback 接口，再实现 confirm 方法。

该方法可以通过ack判断交换机是否收到消息。

注意：因为该方法需要依赖 RabbitTemplate ，所以要用 @PostConstruct 这个注解注入。

import lombok.extern.slf4j.Slf4j;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    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;
    import java.nio.charset.StandardCharsets;
    
    /**
     * 回调接口  告警配置类
     **/
    @Component
    @Slf4j
    public class CallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    
        private static Logger logger = LoggerFactory.getLogger(CallBack.class);
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        //相当于 注入，在其他注入结束后才进行注入
        @PostConstruct
        private void init() {
            //注入
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnCallback(this);
        }
    
        /**
         * 交换机回调方法(针对于交换机是否成功接收消息)
         * 1.发消息 交换机接收到了 回调
         * 1.1 correlationData 保存回调消息的id及相关信息
         * 1.2 交换机收到消息 ack=true
         * 1.3 call null
         * 2. 发消息 交换机失败了 回调
         * 2.1 correlationData 保存回调消息的id及相关信息
         * 2.2 交换机收到消息 ack=false
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            String id = correlationData != null ? correlationData.getId() : "";
    
            if (ack) {
                logger.info("交换机接受消息成功了，消息id为：{}", id);
            } else {
    
                Message message = correlationData.getReturnedMessage();
    
                logger.info("交换机接受失败了，消息id为：{},失败原因：{}", id, cause);
                logger.info("消息是：{}",new String(message.getBody(), StandardCharsets.UTF_8));
            }
    
        }
    }

< 4 > 生产者

故意把交换机写错，来测试交换机异常

// 交换机异常
        @GetMapping("/sendMsgToBadExchange")
        public void sendMsgToBadExchange() {
            //消息id
            CorrelationData correlationData = new CorrelationData("1");
    
            MessageProperties messageProperties = new MessageProperties();
            Message message = new Message("高级发布确认-交换机异常".getBytes(), messageProperties);
    
            correlationData.setReturnedMessage( new Message("correlationData message".getBytes(), messageProperties));
    
            //发送消息
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE + "bad", ConfirmConfig.CONFIRM_ROUNTING_KEY,  message, correlationData);
            log.info("测试交换机异常，发送消息,消息的ID是：{}",correlationData);
        }

**测试：**

**测试结果已告知，交换机未收到消息，并已说明原因。**

![1298b79d2cc9488192911b4aeb3a1dc7.png][]

##  ##

## 二.解决队列消息丢失 ##

< 1 > 配置文件

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

< 2 > 告警类配置

在交换机告警类上接着写的。

该接口要实现 RabbitTemplate.ReturnCallback接口，再实现 returnedMessage方法。

由于该方法也依赖 RabbitTemplate ，所以要用 @PostConstruct这个注解注入。

import lombok.extern.slf4j.Slf4j;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    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;
    import java.nio.charset.StandardCharsets;
    
    /**
     * 回调接口  告警配置类
     **/
    @Component
    @Slf4j
    public class CallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    
        private static Logger logger = LoggerFactory.getLogger(CallBack.class);
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        //相当于 注入，在其他注入结束后才进行注入
        @PostConstruct
        private void init() {
            //注入
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnCallback(this);
        }
    
        /**
         * 交换机回调方法(针对于交换机是否成功接收消息)
         * 1.发消息 交换机接收到了 回调
         * 1.1 correlationData 保存回调消息的id及相关信息
         * 1.2 交换机收到消息 ack=true
         * 1.3 call null
         * 2. 发消息 交换机失败了 回调
         * 2.1 correlationData 保存回调消息的id及相关信息
         * 2.2 交换机收到消息 ack=false
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            String id = correlationData != null ? correlationData.getId() : "";
    
            if (ack) {
                logger.info("交换机接受消息成功了，消息id为：{}", id);
            } else {
    
                Message message = correlationData.getReturnedMessage();
    
                logger.info("交换机接受失败了，消息id为：{},失败原因：{}", id, cause);
                logger.info("消息是：{},消息id：{}",new String(message.getBody(), StandardCharsets.UTF_8),id);
            }
    
        }
    
        /**
         * 只有消息不可达到目的地时，返回给生产者
         *
         * 如果有备份交换机，则不执行
         *
         * @param message    消息
         * @param replayCode 失败码
         * @param replayText 失败原因
         * @param exchanges  交换机
         * @param routingKey 路由
         */
        @Override
        public void returnedMessage(Message message, int replayCode, String replayText, String exchanges, String routingKey) {
            System.out.println("队列接受失败了");
            System.out.println("消息：" + new String(message.getBody(), StandardCharsets.UTF_8) +"     消息id："+message.getMessageProperties().getMessageId()+ ";消息码：" + replayCode + ";原因：" + replayText + ";交换机：" + exchanges + ";路由：" + routingKey);
        }
    }

< 3 > 生产者

故意把路由Key写错，演示队列接受失败。

//路由Key异常
        @GetMapping("/sendMsgToBadRounting")
        public void sendMsgToBadRounting() {
            //消息id
            CorrelationData correlationData = new CorrelationData("123");
    
            MessageProperties messageProperties = new MessageProperties();
            //消息id
            messageProperties.setMessageId(correlationData.getId());
            Message message = new Message("高级发布确认-路由Key异常".getBytes(), messageProperties);
            //发送消息
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUNTING_KEY +"bad",  message, correlationData);
            log.info("测试路由Key异常，发送消息,消息Id是：{}",correlationData);
        }

测试：

测试结果，交换机接受成功了，队列接受失败了，并告知失败原因。

![ecad6677a9af433e93c482a24cdc672c.png][]

![5b3ec04897074f56824f312a5a2c3465.png][]

## 三.备份队列与告警队列 ##

什么是备份交换机呢？备份 交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时， 就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由 备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑 定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都 进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。

< 1 > 配置类

import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /*
     *   rabbitmq 发布高级确认
     *
     * */
    @Configuration
    public class ConfirmConfig {
        //队列
        public static final String CONFIRM_QUEUE = "CONFIRM_QUEUE";
        //交换机
        public static final String CONFIRM_EXCHANGE = "CONFIRM_EXCHANGE";
        //routingKey
        public static final String CONFIRM_ROUNTING_KEY = "CONFIRM_ROUNTING_KEY";
        //备份交换机
        public static final String BACKUP_EXCHANGE = "BACKUP_EXCHANGE";
        //备份队列
        public static final String BACKUP_QUEUE = "BACKUP_QUEUE";
        //告警队列
        public static final String WARNING_QUEUE = "WARNING_QUEUE";
    
    
        //声明交换机
        //这里因为用的是之前的交换机，所以需要删除原先的交换机才能生效
        @Bean
        public DirectExchange confirmEchange() {
            // 路由Key异常，走 消息退回
        //   return (DirectExchange) ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).durable(true).build();
    
            //添加了备份交换机 不走 消息退回
            return (DirectExchange) ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).durable(true).withArgument("alternate-exchange", BACKUP_EXCHANGE).build();
        }
    
        //声明队列
        @Bean
        public Queue confirmQueue() {
            //创建队列
            return new Queue(CONFIRM_QUEUE);
        }
    
    
        //绑定队列
        @Bean
        public Binding confirmBindingQueue(@Qualifier("confirmQueue") Queue confirmQueue, @Qualifier("confirmEchange") DirectExchange confirmEchange) {
            return BindingBuilder.bind(confirmQueue).to(confirmEchange).with(CONFIRM_ROUNTING_KEY);
        }
    
    //----------------------------------------------------------------------------------------------------------------------------------------------------------
    
        //声明备份交换机
        @Bean
        public FanoutExchange backupEchange() {
            return new FanoutExchange(BACKUP_EXCHANGE);
        }
    
    
        //声明备份队列
        @Bean
        public Queue backupQueue() {
            //创建队列
            return new Queue(BACKUP_QUEUE);
        }
    
        //绑定备份队列
        @Bean
        public Binding backupBindingQueue(@Qualifier("backupQueue") Queue backupQueue, @Qualifier("backupEchange") FanoutExchange backupEchange) {
            return BindingBuilder.bind(backupQueue).to(backupEchange);
        }
    
    //----------------------------------------------------------------------------------------------------------------------------------------------------------
    
        //声明报警队列
        @Bean
        public Queue warningQueue() {
            //创建队列
            return new Queue(WARNING_QUEUE);
        }
    
    
        //绑定报警队列
        @Bean
        public Binding warningBindingQueue(@Qualifier("warningQueue") Queue warningQueue, @Qualifier("backupEchange") FanoutExchange backupEchange) {
            return BindingBuilder.bind(warningQueue).to(backupEchange);
        }
    
    
    }

< 2 > 消费者（备份）

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /**
     *  备份 消费者
     **/
    @Slf4j
    @Component
    public class BackupConsume {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        //接收消息
        @RabbitListener(queues = ConfirmConfig.BACKUP_QUEUE)
        public void receiveDelay(Message message) {
            String messageId = message.getMessageProperties().getMessageId();
            String s = new String(message.getBody());
            log.info("走的是备份交换机消息：{} ,消费ID为：{}" ,s,messageId);
            
        }
    }

< 3 > 消费者（告警）

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
      告警消费者
     **/
    @Slf4j
    @Component
    public class WarningConsume {
        //接收消息
        @RabbitListener(queues = ConfirmConfig.WARNING_QUEUE)
        public void receiveDelay(Message msg) {
    
            String message = new String(msg.getBody());
            log.info("报警发现不可路由的消息：{},消息id：" + message,msg.getMessageProperties().getMessageId());
        }
    }

< 4 > 生产者

备份消费者和告警消费者是在消息不可达的时候才会会消费。因此，故意写错路由Key。

测试：

测试结果，交换机接受正常，备份队列和告警队列都接受到了消息。

因为添加了备份交换机，消息退回就不执行了。

![99d2e14038874f94a4636d74e3bad35e.png][]

[77c345547dda47caacd7d035a969e3d9.png]: https://img-blog.csdnimg.cn/77c345547dda47caacd7d035a969e3d9.png
[1298b79d2cc9488192911b4aeb3a1dc7.png]: https://img-blog.csdnimg.cn/1298b79d2cc9488192911b4aeb3a1dc7.png
[ecad6677a9af433e93c482a24cdc672c.png]: https://img-blog.csdnimg.cn/ecad6677a9af433e93c482a24cdc672c.png
[5b3ec04897074f56824f312a5a2c3465.png]: https://img-blog.csdnimg.cn/5b3ec04897074f56824f312a5a2c3465.png
[99d2e14038874f94a4636d74e3bad35e.png]: https://img-blog.csdnimg.cn/99d2e14038874f94a4636d74e3bad35e.png