• RabbitMQ(八)发布确认高级


    在生产环境中由于一些不明原因,导致 rabbitmq重启,在RabbitMQ重启期间生产者消息投递失败,导致消息丢失,需要手动处理和恢复。于是,我们开始思考,如何才能进行RabbitMQ.的消息可靠投递呢?特别是在这样比较极端的情况,RabbitMQ集群不可用的时候,无法投递的消息该如何处理呢?

    8.1发布确认springboot版本

    8.1.1确认机制方案

    image-20220823141204810

    如果交换机宕机了怎么办,后面的队列也无法使用,所以我们需要放在缓存中,用来解决这个问题

    8.1.2代码架构图

    image-20220823141328223

    8.1.3配置类
    package com.example.demo.config;
    
    import com.rabbitmq.client.AMQP;
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:15
     * 发布确认高级
     */
    @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";
    
        //创建交换机
        @Bean
        public DirectExchange confirmExchange(){
            return new DirectExchange(CONFIRM_EXCHANGE_NAME);
        }
    
        @Bean
        public Queue confirmQueue(){
            return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
        }
    
        @Bean
        public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
                                            @Qualifier("confirmExchange") DirectExchange confirmExchange){
            return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
        }
    
    
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    8.1.4生产者
    package com.example.demo.controller;
    
    import com.example.demo.config.ConfirmConfig;
    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;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:22
     */
    @RestController
    @RequestMapping("/confirm")
    @Slf4j
    public class ProduceController {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        @GetMapping("/sendMessage/{message}")
        public void sendMessage(@PathVariable String message){
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY,message);
            log.info("消息内容为{}",message);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    8.1.5消费者
    package com.example.demo.consumer;
    
    import com.example.demo.config.ConfirmConfig;
    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;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:25
     * 发布确认高级
     */
    @Component
    @Slf4j
    public class Consumer {
    
        @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
        public void receiveConfirmMessage(Message message, Channel channel){
            String msg = new String(message.getBody());
            log.info("接收到的队列信息{}",msg);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    完成之后启动项目,进行消息的发送,http://localhost:8080/confirm/sendMessage/a,控制台信息

    消息内容为a 
    接收到的队列信息a
    
    • 1
    • 2

    可见,这就是一个普通的消息发送和接收的程序,但是如果在发消息过程中,交换机宕机了,怎么办?

    8.1.6回调接口

    ​ 当生产者发送消息后,交换机无法回应,那么生产者应该知道自己的消息无法被确认,应该有所反应,所以就需要一个回调接口

    在配置文件中添加:

    spring:
      rabbitmq:
    	publisher-confirm-type: correlated  
    #NONE 禁用发布确认模式 默认值, 
    #correlated 发布消息成功到交换机后触发回调方法,
    #SIMPLE 有两种效果:
    #第一个和correlated一样,
    #第二个在发布消息成功后使用rabbitTemplate,调用waitForConfirms.或 waitForConfirmsOrDie方法等待broker节点返回发送结果,根据返回结果来判定下一步的逻辑,要注意的点是waitForConfirmsOrDie.方法如果返回false则会关闭channel,则接下来无法发送消息到 broker
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    回调接口:

    package com.example.demo.config;
    
    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;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:45
     * 回调接口
     */
    @Slf4j
    @Component
    public class MyCallBack implements RabbitTemplate.ConfirmCallback {
    
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
    
        /*
           交换机确认回调方法
            1.发消息 交换机接收到了消息 进行回调
                1.1correlationData      保存回调消息的ID及相关信息
                1.2b--->ack             交换机收到消息 true
                1.3cause                null
            2.发消息 交换机接收失败进行回调
                2.2correlationData      保存回调消息的ID及相关信息
                2.2b--->ack             交换机收到消息 false   所以可以通过ack判断是否成功
                2.3cause                失败的原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause ) {
            String id = correlationData != null ? correlationData.getId() : null;
            if(ack){
                log.info("交换机已经收到了消息ID为:{}的消息",id);
            }else{
                log.info("交换机还未收到id为:{}的消息,由于原因:{}",id,cause );
            }
        }
    
    
        //回调之后注入
        @PostConstruct
        public  void init(){
            rabbitTemplate.setConfirmCallback(this);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 生产者代码修改
    package com.example.demo.controller;
    
    import com.example.demo.config.ConfirmConfig;
    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.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;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:22
     */
    @RestController
    @RequestMapping("/confirm")
    @Slf4j
    public class ProduceController {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        @GetMapping("/sendMessage/{message}")
        public void sendMessage(@PathVariable String message){
            //回调函数使用后,需要在发送端声明
            CorrelationData correlationData = new CorrelationData("1");
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY,message,correlationData);
            log.info("消息内容为{}",message);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33

    测试,发送信息http://localhost:8080/confirm/sendMessage/d,控制台:

    消息内容为d
    接收到的队列信息d
    交换机已经收到了消息ID为:1的消息
    
    • 1
    • 2
    • 3

    此时是没有问题的,那么下面开始模拟有问题的情况:

    • 将生产者发送消息的交换机名字修改成不存在的交换机名
     rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME+"123",ConfirmConfig.CONFIRM_ROUTING_KEY,message,correlationData);
    
    • 1
    • 重启后再次发送,控制台代码:
    消息内容为d
    交换机还未收到id为:1的消息,由于原因:channel error; protocol method: #method(reply-code=404, reply-text=NOT_FOUND - no exchange 'confirm_exchange123' in vhost '/', class-id=60, method-id=40)
    
    • 1
    • 2

    可见已经回调了函数,失败了,因为该交换机不存在。

    该方法只适用于交换机出问题,当队列或routingKey出现问题导致消费者无法获取消息时,回调函数就不再适用了,我们可以使用下面的方法。

    8.2回退消息

    8.2.1Mandatory参数

    在仅开启了生产者确认机制的情况下,交换机接收到消息后,会直接给消息生产者发送确认消息,如果发现该消息不可路由(消息队列失效、不存在等),那么消息会被直接丢弃,此时生产者是不知道消息被丢弃这个事件的。那么如何让无法被路由的消息帮我想办法处理一下?最起码通知我一声,我好自己处理啊。通过设置mandatory参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

    8.2.2消息生产者代码
    • 配置文件
    spring:
      rabbitmq:
    	publisher-returns: true
    
    • 1
    • 2
    • 3
    • MycallBack方法
    package com.example.demo.config;
    
    import lombok.extern.slf4j.Slf4j;
    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;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:45
     * 回调接口
     */
    @Slf4j
    @Component
    public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
    
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
    
        /*
           交换机确认回调方法
            1.发消息 交换机接收到了消息 进行回调
                1.1correlationData      保存回调消息的ID及相关信息
                1.2b--->ack             交换机收到消息 true
                1.3cause                null
            2.发消息 交换机接收失败进行回调
                2.2correlationData      保存回调消息的ID及相关信息
                2.2b--->ack             交换机收到消息 false   所以可以通过ack判断是否成功
                2.3cause                失败的原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause ) {
            String id = correlationData != null ? correlationData.getId() : null;
            if(ack){
                log.info("交换机已经收到了消息ID为:{}的消息",id);
            }else{
                log.info("交换机还未收到id为:{}的消息,由于原因:{}",id,cause );
            }
        }
    
    
        //回调之后注入
        @PostConstruct
        public  void init(){
            rabbitTemplate.setConfirmCallback(this);
            //注入回退
            rabbitTemplate.setReturnCallback(this);
        }
    
         //当消息传递过程中出问题无法获取到时,将消息退回给生产者
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            log.error("消息:{},被交换机:{}退回,退回的原因:{},路由key:{}",new String(message.getBody()),exchange,replyText,routingKey);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • ProduceController
    package com.example.demo.controller;
    
    import com.example.demo.config.ConfirmConfig;
    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.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;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:22
     */
    @RestController
    @RequestMapping("/confirm")
    @Slf4j
    public class ProduceController {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        @GetMapping("/sendMessage/{message}")
        public void sendMessage(@PathVariable String message){
            //回调函数使用后,需要在发送端声明
            CorrelationData correlationData = new CorrelationData("1");
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY,message,correlationData);
    
            CorrelationData correlationDataT = new CorrelationData("2");
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY+"error",message,correlationDataT);
            log.info("消息内容为{}",message);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 测试,发送http://localhost:8080/confirm/sendMessage/test,控制台消息
    接收到的队列信息test
    交换机已经收到了消息ID为:1的消息
    消息内容为test
    消息:test,被交换机:confirm_exchange退回,退回的原因:NO_ROUTE,路由key:key1error
    交换机已经收到了消息ID为:2的消息
    
    • 1
    • 2
    • 3
    • 4
    • 5

    8.3备份交换机

    ​ 有了mandatory 参数和回退消息,我们获得了对无法投递消息的感知能力,有机会在生产者的消息无法被投递时发现并处理。但有时候,我们并不知道该如何处理这些无法路由的消息,最多打个日志,然后触发报警,再来手动处理。而通过日志来处理这些无法路由的消息是很不优雅的做法,特别是当生产者所在的服务有多台机器的时候,手动复制日志会更加麻烦而且容易出错。而且设置mandatory参数会增加生产者的复杂性,需要添加处理这些被退回的消息的逻辑。如果既不想丢失消息,又不想增加生产者的复杂性,该怎么做呢?前面在设置死信队列的文章中,我们提到,可以为队列设置死信交换机来存储那些处理失败的消息,可是这些不可路由消息根本没有机会进入到队列,因此无法使用死信队列来保存消息。在RabbitMQ中,有一种备份交换机的机制存在,可以很好的应对这个问题。什么是备份交换机呢?备份交换机可以理解为RabbitMQ中交换机的“备胎”,当我们为某一个交换机声明一个对应的备份交换机时,就是为它创建一个备胎,当交换机接收到一条不可路由消息时,将会把这条消息转发到备份交换机中,由备份交换机来进行转发和处理,通常备份交换机的类型为Fanout,这样就能把所有消息都投递到与其绑定的队列中,然后我们在备份交换机下绑定一个队列,这样所有那些原交换机无法被路由的消息,就会都进入这个队列了。当然,我们还可以建立一个报警队列,用独立的消费者来进行监测和报警。

    8.3.1代码架构图

    image-20220825211957804

    8.3.2修改配置类
    package com.example.demo.config;
    
    import com.rabbitmq.client.AMQP;
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:15
     * 发布确认高级
     */
    @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 WARNING_QUEUE_NAME = "warning_queue";
    
        //创建交换机
        @Bean
        public DirectExchange confirmExchange(){
            //备份交换机和主交换机构建链接
            return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
                    .withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
        }
    
        @Bean
        public Queue confirmQueue(){
            return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
        }
    
        @Bean
        public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
                                            @Qualifier("confirmExchange") DirectExchange confirmExchange){
            return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
        }
    
        //备份交换机,type=fanout
        @Bean
        public FanoutExchange backupExchange(){
            return new FanoutExchange(BACKUP_EXCHANGE_NAME);
        }
    
        //备份队列
        @Bean
        public Queue backQueue(){
            return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
        }
    
        //报警队列
        @Bean
        public Queue warningQueue(){
            return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
        }
    
        //备份交换机和队列的绑定
        @Bean
        public Binding backupQueueBindingBackupExchange(@Qualifier("backQueue") Queue backQueue,
                                            @Qualifier("backupExchange") FanoutExchange backupExchange){
            return BindingBuilder.bind(backQueue).to(backupExchange);
        }
    
        @Bean
        public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue warningQueue,
                                            @Qualifier("backupExchange") FanoutExchange backupExchange){
            return BindingBuilder.bind(warningQueue).to(backupExchange);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 只测试警报队列
    • WarningConsumer
    package com.example.demo.consumer;
    
    import com.example.demo.config.ConfirmConfig;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/25 21:34
     * 报警队列
     */
    @Component
    @Slf4j
    public class WarningConsumer {
    
        //接收报警消息
        @RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)
        public void receiveWarningMessage(Message message){
            String msg = new String(message.getBody());
            log.error("报警发现不可路由消息:{}",msg);
        }
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • Produce
    package com.example.demo.controller;
    
    import com.example.demo.config.ConfirmConfig;
    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.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;
    
    /**
     * @author 我见青山多妩媚
     * @date Create on 2022/8/23 14:22
     */
    @RestController
    @RequestMapping("/confirm")
    @Slf4j
    public class ProduceController {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        @GetMapping("/sendMessage/{message}")
        public void sendMessage(@PathVariable String message){
            //回调函数使用后,需要在发送端声明
            CorrelationData correlationData = new CorrelationData("1");
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY,message+"test1",correlationData);
    
            CorrelationData correlationDataT = new CorrelationData("2");
            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY+"error",message+"test2",correlationDataT);
            log.info("消息内容为{}",message);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 测试,发送消息http://localhost:8080/confirm/sendMessage/demo,控制台结果如下
    接收到的队列信息demotest1
    交换机已经收到了消息ID为:1的消息
    消息内容为demo
    报警发现不可路由消息:demotest2
    交换机已经收到了消息ID为:2的消息
    
    • 1
    • 2
    • 3
    • 4
    • 5

    我们可以看到,回退消息此时也在配置中,那么回退消息和备份交换机一起使用时,由测试结果可以看到,备份交换机的优先级更高

  • 相关阅读:
    【Git】深入了解Git及其常用命令
    TS+elementUI的表格做form校验写法(手机/邮箱号验证)
    18.客户端会话技术Cookie
    尚好房 02_用户角色管理
    2023年美国代理IP推荐,哪家IP代理好用?
    DataFrame在指定位置插入行和列
    element 周选择器el-date-picker
    Spring Security实现基于RBAC的权限表达式动态访问控制
    高数 | 【数一】 多元函数积分学预备知识 —— 总复习框架总结
    CSS 定位布局
  • 原文地址:https://blog.csdn.net/YSecret_Y/article/details/126776509