RabbtiMQ:发布确认高级

2022-6-20

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

发布确认springboot版

确认机制方案

代码架构图

配置文件

在配置文件当中需要添加
spring.rabbitmq.publisher-confirm-type=correlated
⚫ NONE
禁用发布确认模式，是默认值
⚫ CORRELATED
发布消息成功到交换器后会触发回调方法
⚫ SIMPLE

经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，
其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker

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

回退消息

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

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

配置文件中加入: spring.rabbitmq.publisher-returns=true

实现 在下方的   MyCallBack.java

代码

配置类:ConfirmConfig

@Configuration
public class ConfirmConfig {
      //交换机
    public  static final String confirm_exchange_name = "confirm_exchange";
    //队列
    public  static final String confirm_queue_name = "confirm_queue";
    //Routing
    public  static final String confirm_routing_name = "key1";
    //声明交换机
    @Bean("confirmExchange")
    public DirectExchange confirmExchange(){
        return new DirectExchange(confirm_exchange_name);
    }
    @Bean("confirmQueue")
    public Queue confirmQueue(){
        return QueueBuilder.durable(confirm_queue_name).build();
    }
    //绑定
    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue queue,
                                        @Qualifier("confirmExchange") DirectExchange confirmExchange){
        return BindingBuilder.bind(queue).to(confirmExchange).with(confirm_routing_name);
    }
}

消息生产者:ProducerController

@Slf4j
@RestController
@RequestMapping("/confirm")
public class ProducerController {
    @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_name,message);
        log.info("发送消息:{}",message);
    }
}

回调接口:MyCallBack

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
    @Autowired
    private  RabbitTemplate rabbitTemplate;
    //注入
    @PostConstruct
    public void init(){
         /**
        * 和配置文件spring.rabbitmq.publisher-returns=true 作用一样
        * true：
        * 交换机无法将消息进行路由时，会将该消息返回给生产者
        * false：
        * 如果发现消息无法进行路由，则直接丢弃
        */
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }
    /**
     * 交换机确认回调方法
     * 1.发消息 交换机接收到了 回调
     *   1.1 correlationData 保存回调消息的ID及相关信息
     *   1.2 交换机收到消息 ack=true
     *   1.3 cause null
     * 2.发消息 交换机接收失败 回调
     *   2.1 correlationData 保存回调消息的ID及相关信息
     *   2.2 交换机收到消息 ack=false
     *   2.3 cause 失败的原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData!=null?correlationData.getId():"";
        if(ack){
            log.info("交换机收到id为{}的消息",id);
        }else {
            log.info("交换机还未收到id为{}的信息，原因:{}",id,cause);
        }
    }

    //当消息传递过程中不可达目的地时将消息返回给生产者。
    //只有不可达到目的地时，才进行回退
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText,
                                String exchange, String routingKey) {
        log.error("消息{}，被交换机{}退回，原因：{}，路由ley：{}",
                  new  String(message.getBody()),exchange,replyText,routingKey);
    }
}

消息消费者:Consumer
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
    @Autowired
    private  RabbitTemplate rabbitTemplate;
    //注入
    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }
    /**
     * 交换机确认回调方法
     * 1.发消息 交换机接收到了 回调
     *   1.1 correlationData 保存回调消息的ID及相关信息
     *   1.2 交换机收到消息 ack=true
     *   1.3 cause null
     * 2.发消息 交换机接收失败 回调
     *   2.1 correlationData 保存回调消息的ID及相关信息
     *   2.2 交换机收到消息 ack=false
     *   2.3 cause 失败的原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData!=null?correlationData.getId():"";
        if(ack){
            log.info("交换机收到id为{}的消息",id);
        }else {
            log.info("交换机还未收到id为{}的信息，原因:{}",id,cause);
        }
    }

    //当消息传递过程中不可达目的地时将消息返回给生产者。
    //只有不可达到目的地时，才进行回退
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText,
                                String exchange, String routingKey) {
        log.error("消息{}，被交换机{}退回，原因：{}，路由ley：{}",
                  new  String(message.getBody()),exchange,replyText,routingKey);
    }
}

测试

修改上面 生产者中的代码，使其发送两条消息，routingKey为key1（存在），key12（不存在），结果:

 备份交换机

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

代码结构

修改配置类

@Configuration
public class ConfirmConfig {
      //交换机
    public  static final String confirm_exchange_name = "confirm_exchange";
    //队列
    public  static final String confirm_queue_name = "confirm_queue";
    //Routing
    public  static final String confirm_routing_name = "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("backupExchange")
    public FanoutExchange backupExchange(){
        return new FanoutExchange(backup_exchange_name);
    }
    @Bean("backupQueue")
    public Queue backupQueue(){
        return QueueBuilder.durable(backup_queue_name).build();
    }
    @Bean("warningQueue")
    public Queue warningQueue(){
        return QueueBuilder.durable(warning_queue_name).build();
    }
    //绑定
    @Bean
    public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue queue,
                                        @Qualifier("backupExchange") FanoutExchange backupExchange){
        return BindingBuilder.bind(queue).to(backupExchange);
    }
    //绑定
    @Bean
    public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue queue,
                                                     @Qualifier("backupExchange") FanoutExchange backupExchange){
        return BindingBuilder.bind(queue).to(backupExchange);
    }
    //////////////////////////////////////////////////////////////////
    //声明交换机
    @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(){
        return QueueBuilder.durable(confirm_queue_name).build();
    }
    //绑定
    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue queue,
                                        @Qualifier("confirmExchange") DirectExchange confirmExchange){
        return BindingBuilder.bind(queue).to(confirmExchange).with(confirm_routing_name);
    }
}

 报警消费者

@Component
@Slf4j
public class WarningConsumer {
    //接收报警消息
    @RabbitListener(queues = ConfirmConfig.warning_queue_name)
    public void receiveWarningMsg(Message message){
        String msg = new String(message.getBody());
        log.error("报警发现不可路由的消息:{}",message);
    }
}

测试

修改上面 生产者中的代码，使其发送两条消息，routingKey为key1（存在），key12（不存在），结果:

阅读剩余
版权声明：
作者：Tin
链接：http://www.tinstu.com/2061.html
文章版权归作者所有，未经允许请勿转载。
THE END
RabbitMQ
RabbitMQ:延迟队列(SpringBoot)
<<上一篇
RabbitMQ:幂等性/优先队列/惰性队列
下一篇>>
相关推荐
商城:RabbitMQ
RabbitMQ:集群
RabbitMQ:幂等性/优先队列/惰性队列
RabbtiMQ:发布确认高级

JAVA学习路茫茫，脚步不能停!

Copyright © 2025 Java学习笔记 CoreNext Powered by WordPress