@toc

引言

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

1、发布确认(SpringBoot版本)

1.1 代码架构图

1.2 配置文件

spring.rabbitmq.host=192.168.159.33
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
# 发布消息成功到交换器后会触发回调方法
spring.rabbitmq.publisher-confirm-type=correlated

在配置文件中添加

spring.rabbitmq.publisher-confirm-type=correlated

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

1.3 添加配置类

/**
 * 配置类  发布确认(高级)
 */
@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("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 confirmQueue,
                                        @Qualifier("confirmExchange") DirectExchange confirmExchange){
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
    }
    
}

1.4 消息生产者

这里发送两条消息，但故意将第二条消息的交换机名称写错

@Api(description = "发布确认高级控制器")
@Slf4j
@RestController
@RequestMapping("/confirm")
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    //发消息
    @ApiModelProperty("发消息")
    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message){

        CorrelationData correlationData1=new CorrelationData("1");

         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,
                ConfirmConfig.CONFIRM_ROUTING_KEY,message+"key1",correlationData1);
        log.info("发送消息内容:{}",message+"key1");
        
         CorrelationData correlationData2=new CorrelationData("2");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME+"121",
                ConfirmConfig.CONFIRM_ROUTING_KEY,message+"key2",correlationData2);
        log.info("发送消息内容:{}",message+"key2");
    }

}

1.5 回调接口

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        //注入
        rabbitTemplate.setConfirmCallback(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);
        }
    }
}

1.6 消息消费者

/**
 * 接收消息
 */
@Slf4j
@Component
public class Consumer {

    @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
    public void receiveConfirmMessage(Message message){
        String msg = new String(message.getBody());
        log.info("接受到队列confirm.queue消息:{}",msg);
    }
}

1.7 测试结果

swagger发送测试请求

  可以看到，发送了两条消息，两条消息都被发送，但是交换机只接收到了其中的一条，错误的那一条由于找不到交换机被丢弃掉。

2、回退消息

2.1 Mandatory参数

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

2.2 回调接口

@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("消息{},被交换机{}退回，退回的原因：{},路由Key:{}",new String(message.getBody()),
                exchange,replyText,routingKey);
    }
}

2.3 消息生产者代码

这回发送两条消息，一条正常的，另一条故意将routingKey写错，看消息是否会被回退

@Api(description = "发布确认高级控制器")
@Slf4j
@RestController
@RequestMapping("/confirm")
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    //发消息
    @ApiModelProperty("发消息")
    @GetMapping("/sendMessage/{message}")
    public void sendMessage(@PathVariable String message){

        CorrelationData correlationData1=new CorrelationData("1");

        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,
                ConfirmConfig.CONFIRM_ROUTING_KEY,message+"key1",correlationData1);
        log.info("发送消息内容:{}",message+"key1");
        

        CorrelationData correlationData2=new CorrelationData("2");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,
                ConfirmConfig.CONFIRM_ROUTING_KEY+"2",message+"key12",correlationData2);
        log.info("发送消息内容:{}",message+"key12");
    }

}

2.4 配置

spring.rabbitmq.host=192.168.159.33
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
# 发布消息成功到交换器后会触发回调方法
spring.rabbitmq.publisher-confirm-type=correlated
#路由不出去会回退消息给生产者
spring.rabbitmq.publisher-returns=true

2.5回退测试

  可以看到，routing为Key1的消息被队列正常接收，另一条由于routingKey是乱写的，消息会被退回给生产者。