RabbitMQ 学习（七）-- 高级发布确认

高级发布确认（生产者角度）

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

1、发布确认机制方案

• 应答机制传回消息是 broker, 确认机制是消息到队列后，传回确定消息给 producer，现在 broker 坏了，无法进行消息确认，所以考虑现在这种方法解决，这里针对的 producer 的角度。

• 架构：当生产者投递消息失败了，就把消息存入到缓存当中，然后使用定时任务重新发送，交换机接收到消息了就删除缓存。

• 确认模式：消息从 生产者 到 交换机 则会返回一个 ConfirmCallback 。

• 回退模式：消息从 交换机 到 队列 投递失败则会返回一个 ReturnCallback 。

确认模式: (生产者和交换机的问题)

步骚:

1、配置文件当中需要添加 spring.rabbitmq.publisher-confirm-type=correlated

2．实现 RabbitTemplate.ConfirmCallBack 接口

回退模式: （交换机和队列的问题）

​ 当消息发送给 Exchange 后，Exchange 路由到 Queue 失败是才会执行 ReturnCallBack

步骤:

1、配置文件当中需要添加 spring.rabbitmq.publisher-returns=true

2、实现 RabbitTemplate.ReturnCallback 接口

---

基于 SpringBoot 实战

2、消息确认

2.1、配置文件

配置文件中添加

spring.rabbitmq.publisher-confirm-type=correlated 
1

• NONE

  禁用发布确认模式，是默认值

• CORRELATED

  发布消息成功到交换器后会触发回调方法

• SIMPLE

  在发布消息成功后使用 RabbitTemplate 调用 waitForConfirms() 或 waitForConfirmsOrDie() 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie() 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broke。

2.2、配置类代码

/**
 * @desc
 * @auth llp
 * @date 2022年08月10日 23:47
 */
@Configuration
public class ConfirmConfig {
    public static final String CONFIRM_EXCHANGE = "confirm.exchange";
    public static final String CONFIRM_QUEUE = "confirm.queue";
    public static final String CONFIRM_ROUTING_KEY = "confirm.routing.key";

    @Bean
    public DirectExchange confirmExchange(){
        return new DirectExchange(CONFIRM_EXCHANGE);
    }

    @Bean
    public Queue confirmQueue(){
        return new Queue(CONFIRM_QUEUE);
    }

    @Bean
    public Binding queueBingExchange(
            @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

2.3、回调接口

@Slf4j
@Component  // 第一步注入
public class ConfirmCallBack implements RabbitTemplate.ConfirmCallback {

    @Resource   // 第二步注入
    private RabbitTemplate rabbitTemplate;

    // @PostConstruct 注解,在对象加载完依赖注入后执行它通常都是一些初始化的操作，
    // 但初始化可能依赖于注入的其他组件，所以要等依赖全部加载完再执行
    // 第三步注入
    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
    }

    /**
     * @desc 交换机确认回调方法， 发消息 交换机接收到消息触发回调函数
     * @auth llp
     * @date 2022/8/11 22:40
     * @param correlationData 保存回调消息的 ID 及相关信息
     * @param ack 交换机是否收到消息
     * @param cause 失败的原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack){
            log.info("交换机已收到ID为：{} 的消息, cause:{}", correlationData.getId(), cause);
        }else {
            log.info("交换机未收到ID为：{} 的消息, cause:{}", correlationData.getId(), cause);
        }
    }
}
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

2.4、生产者代码

@Slf4j
@RestController
public class ConfirmController {
    @Resource
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/confirm/sendMsg/{message}")
    public void sendMessage(@PathVariable String message) throws InterruptedException {
        // 交换机和路由键都正确
        CorrelationData correlationData1 = new CorrelationData("1");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUTING_KEY, message, correlationData1);
        log.info("发送的消息内容为：{}", message);
        TimeUnit.SECONDS.sleep(1);
        
		// 交换机不正确，路由键正确
        CorrelationData correlationData2 = new CorrelationData("2");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE+"ddd", ConfirmConfig.CONFIRM_ROUTING_KEY, message, correlationData2);
        log.info("发送的消息内容为：{}", message);
        TimeUnit.SECONDS.sleep(1);
        
		// 交换机正确，路由键不正确（找不到队列）
        CorrelationData correlationData3 = new CorrelationData("3");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUTING_KEY+"ddd", message, correlationData3);
        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

2.5、消费者代码

@Slf4j
@Component
public class ConfirmConsumer {
    @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE)
    public void receiveQueueDead(Message message, Channel channel){
        String msg = new String(message.getBody());
        log.info("当前时间：{}, 接收到消息为：{}", new Date(), msg);
    }
}
1
2
3
4
5
6
7
8
9

2.6、运行测试与问题

第三条消息会被直接丢弃，生产者不能得知，如何解决？消息回退模式

3、消息回退

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

3.1、配置文件

spring:
  rabbitmq:
   publisher-returns: true
1
2
3

3.2、回调接口

@Slf4j
@Component  // 第一步注入
public class ConfirmCallBack implements RabbitTemplate.ReturnCallback {

    @Resource   // 第二步注入
    private RabbitTemplate rabbitTemplate;

    // @PostConstruct 注解,在对象加载完依赖注入后执行它通常都是一些初始化的操作，
    // 但初始化可能依赖于注入的其他组件，所以要等依赖全部加载完再执行
    // 第三步注入
    @PostConstruct
    public void init(){
        rabbitTemplate.setReturnCallback(this);
    }

    // 只有消息不可路由才会调用此函数，成功不调用
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("消息 {}, 被交换机 {} 退回, 应答代码 {}, 原因 {}, 路由 {}",
                new String(message.getBody()), exchange, replyCode, replyText, routingKey);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

3.3、运行测试