RabbitMQ 消息丢失解决（高级发布确认、消息回退与重发、备份交换机）


server:
  port: 8888
spring:
  rabbitmq:
    host: 192.168.163.132
    port: 5672
    username: 2252631565
    password: 2252631565
#    高级发布确认 发布消息成功后将会触发回调方法
    publisher-confirm-type: correlated

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

添加配置类

声明交换机和队列，并且将交换机和队列进行绑定:


@Configuration
public class ConfirmConfig {
    public static final String CONFIRM_EXCHANGE="confirm.exchange";
    public static final String CONFIRM_QUEUE="confirm.queue";
    public static final String ROUTING_KEY="key1";
 
    @Bean
    public DirectExchange confirmExchange(){
        return new DirectExchange(CONFIRM_EXCHANGE,false,false);
    }
 
    @Bean
    public Queue confirmQueue(){
        return QueueBuilder.durable(CONFIRM_QUEUE).build();
    }
 
    @Bean
    public Binding EAndQBind(@Qualifier("confirmExchange") DirectExchange exchange,@Qualifier("confirmQueue")Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
}

消息生产者

也可以说是 Controller 层，在这里发送两条消息给两个交换机，其中一个交换机是我们设置好的，另一个交换机不存在；这样就可以清晰看出交换机应答效果。


@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);
        CorrelationData correlationData=new CorrelationData("1");
        correlationData.setReturnedMessage(new org.springframework.amqp.core.Message(message.getBytes()));
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY,message,correlationData);
        //向一个不存在的交换机发送消息
        log.info("发送一条时长为的消息给第一个队列内容是:{}",new Date().toString(),message);
        CorrelationData correlationData2=new CorrelationData("2");
        correlationData2.setReturnedMessage(new org.springframework.amqp.core.Message(message.getBytes()));
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE+123,ConfirmConfig.ROUTING_KEY,message,correlationData2);
    }
 
}

消息消费者

监听 confirm.queue 队列


@Slf4j
@Component
public class ConfirmLetterQueue {
    @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE)
    public void confirmConsumer(Message message, Channel channel){
        log.info("收到了消息：{}",new String(message.getBody()));
    }
}

消息生产者发布消息后的回调接口

只要生产者发布消息，交换机不管是否收到消息，都会调用该类的 confirm 方法


@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
 
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @PostConstruct
    public void init(){
        //注入
        rabbitTemplate.setConfirmCallback(this::confirm);
    }
    /*交换机确认回调
    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) {
        if(ack){
            log.info("发送消息到交换机成功！消息体为：{}",new String(correlationData.getReturned().getMessage().getBody()));
        }else {
            log.info("发送消息到交换机失败！原因为：{}",cause.toString());
        }
    }
}

测试：

效果很明显，我们配置的交换机成功收到消息并转发给队列；不存在的交换机没有接受到消息并作出反应。

1.2回退消息并重发（队列的发布确认）

在配置文件中开启消息回退功能


server:
  port: 8888
spring:
  rabbitmq:
    host: 192.168.163.133
    port: 5672
    username: 2252631565
    password: 2252631565
#    高级发布确认 发布消息成功后将会触发回调方法
    publisher-confirm-type: correlated
    #    消息回退 当消息未路由至队列时触发
    publisher-returns: true

修改回调接口

实现 RabbitTemplate.ReturnsCallback 接口，并实现方法


@Slf4j
@Component
public class MyCallBack  implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{
 
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @PostConstruct
    public void init(){
        //注入
        rabbitTemplate.setConfirmCallback(this::confirm);
        rabbitTemplate.setReturnsCallback(this::returnedMessage);
    }
    /*交换机确认回调
    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) {
        if(ack){
            log.info("时间:{}发送消息到交换机成功！",new Date());
        }else {
            log.info("发送消息到交换机失败！原因为：{}",cause.toString());
        }
    }
 
 
    //当消息未路由到队列时触发 只有失败时才触发 若消息发送至延迟队列则一定会触发回退 记得根据交换机名称排除延迟队列
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.error("消息：'{}',被交换机：{}回退，回退原因为：{}，routingKey为：{}"
                ,new String(returned.getMessage().getBody())
                ,returned.getExchange()
                ,returned.getReplyText()
                ,returned.getRoutingKey());
        //10s后消息重发
        try {
            Thread.sleep(10000);
            log.info("时间:{}，生产者重新发消息",new Date());
            rabbitTemplate.convertAndSend(returned.getExchange(),ConfirmConfig.ROUTING_KEY,new String(returned.getMessage().getBody()));
        }catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

生产者：

向交换机中发送消息，指定错误的routingkey,触发队列回退消息并重发消息。


    //高级发布确认模式
    @GetMapping("/sendConfirmMsg/{message}")
    public void sendConfirmMsg(@PathVariable String message){
        //向一个不存在的队列发送消息
        log.info("时间:{}生产者发送一条的消息给第一个队列内容是:{}",new Date().toString(),message);
        CorrelationData correlationData2=new CorrelationData("2");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY+123,message,correlationData2);
    }

测试：

回退未进入队列的消息并重新发送消息。

二、备份交换机

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

实战

需要一个备份交换机 backup.exchange，类型为 fanout，该交换机发送消息到队列 backup.queue 和 warning.queue。

修改高级确认发布配置类


@Configuration
public class ConfirmConfig {
    public static final String CONFIRM_EXCHANGE="confirm.exchange";
    public static final String CONFIRM_QUEUE="confirm.queue";
    public static final String ROUTING_KEY="key1";
 
    //备份交换机
    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 confirmExchange(){
        //绑定确认交换机与备份交换机
        Map argument=new HashMap<>();
        argument.put("alternate-exchange",BACKUP_EXCHANGE);
        return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).withArguments(argument).build();
    }
 
    //备份交换机
    @Bean
    public FanoutExchange backupExchange(){
        return new FanoutExchange(BACKUP_EXCHANGE);
    }
    @Bean
    public Queue confirmQueue(){
        return QueueBuilder.durable(CONFIRM_QUEUE).build();
    }
    //备份队列
    @Bean
    public Queue backupQueue(){
        return QueueBuilder.durable(BACKUP_QUEUE).build();
    }
    //警告队列
    @Bean
    public Queue warningQueue(){
        return QueueBuilder.durable(WARNING_QUEUE).build();
    }
    @Bean
    public Binding EAndQBind(@Qualifier("confirmExchange") DirectExchange exchange,@Qualifier("confirmQueue")Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
    //绑定备份交换机与两个队列
    @Bean
    public Binding BAndBBing(@Qualifier("backupExchange") FanoutExchange exchange,@Qualifier("backupQueue")Queue queue){
        return BindingBuilder.bind(queue).to(exchange);
    }
    @Bean
    public Binding BAndWBing(@Qualifier("backupExchange") FanoutExchange exchange,@Qualifier("warningQueue")Queue queue){
        return BindingBuilder.bind(queue).to(exchange);
    }
}

生产者

生产者发送两条消息一个配置正确的路由，另一个是错误的路由。预期目标是正确路由正常接收消息，错误路由传输的信息由警告队列接收。


    //高级发布确认模式
    @GetMapping("/sendConfirmMsg/{message}")
    public void sendConfirmMsg(@PathVariable String message){
        log.info("时间:{}生产者发送两条消息队列内容是:{}",new Date().toString(),message);
        CorrelationData correlationData=new CorrelationData("1");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY,message,correlationData);
        //向一个不存在的队列发送消息
        CorrelationData correlationData2=new CorrelationData("2");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE,ConfirmConfig.ROUTING_KEY+123,message,correlationData2);
    }

报警消费者：

接收不可路由的消息


/**
 * 报警消费者
 */
@Slf4j
@Component
public class WarningConsumer {
    //监听报警消息
    @RabbitListener(queues = ConfirmConfig.WARNING_QUEUE)
    public void receiveWarningMsg(Message message){
        String msg=new String(message.getBody());
        log.info("报警发现不可路由消息：{},重发消息",msg);
    }
}

测试：

生产者发送两条消息一个配置正确的路由，另一个是错误的路由。预期目标是正确路由正常接收消息，错误路由传输的信息由警告队列接收。

在此案例中，也设置了消息回退的配置，但是没有触发消息回退。由此得出：备份交换机的优先级更高。

相关阅读:
4009. Problem D：走台阶问题
 【Git】window下大小写不敏感问题处理
 物理备份xtrabackup
SwiftUI Spacer() onTapGesture 无法触发
 k8s-9.部署elk日志收集
 在 Ubuntu 20.04 上，将 Subversion 从默认的 1.13.0 降级到 1.9.7。解决svn无法保存密码问题
 牛客网：NC26 括号生成
 家居服务小程序发展指南
 一起看看Python中的迭代器&生成器
 streamlit报错：AxiosError: Request failed with status code 403
原文地址：https://blog.csdn.net/weixin_52536274/article/details/134456459

RabbitMQ 消息丢失解决 （高级发布确认、消息回退与重发、备份交换机）

一、发布确认SpringBoot版本

确认机制图例：

代码实战：

代码架构图：

1.1交换机的发布确认