个人博客

生产者推送消息确认机制有2种，分别是：

  • 确认消息已发送到交换机(Exchange)
  • 确认消息已发送到队列(Queue)

1、配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
server:
  port: 30200

spring:
  rabbitmq:
    host: 148.70.153.63
    port: 5672
    username: libai
    password: password
    # 确认消息已发送到交换机(Exchange); Springboot2.2.x改为 publisher-confirm-type: correlated
    publisher-confirms: true
    # 确认消息已发送到队列(Queue)
    publisher-returns: true

2、RabbitTemplate配置

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
@Configuration
@Slf4j
public class RabbitMQConfig {
    /**
     * 1. 消息推送到server，但是在server里找不到交换机
     * 2. 消息推送到server，找到交换机了，但是没找到队列
     * 3. 消息推送成功
     */
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        // 设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);

        // 确认消息已发送到交换机(Exchange)
        // 1、2、3都会触发此回调函数
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            log.info("ConfirmCallback:     " + "相关数据：" + correlationData);
            log.info("ConfirmCallback:     " + "确认情况：" + ack);
            log.info("ConfirmCallback:     " + "原因：" + cause);
        });

        // 确认消息已发送到队列(Queue)
        // 只有2才会触发此回调函数
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("ReturnCallback:     " + "消息：" + message);
            log.info("ReturnCallback:     " + "回应码：" + replyCode);
            log.info("ReturnCallback:     " + "回应信息：" + replyText);
            log.info("ReturnCallback:     " + "交换机：" + exchange);
            log.info("ReturnCallback:     " + "路由键：" + routingKey);
        });
        return rabbitTemplate;
    }
}

在配置中重新定义了RabbitTemplate，分3种情况测试：

  1. 消息推送到server，但是在server里找不到交换机。
  2. 消息推送到server，找到交换机了，但是没找到队列。
  3. 消息推送成功。

结果：

  1. 只回调ConfirmCallback函数。
1
2
3
ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：false
ConfirmCallback:     原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'directExchange1' in vhost '/', class-id=60, method-id=40)
  1. 回调ConfirmCallback和ReturnCallback函数。
1
2
3
4
5
6
7
8
9
ReturnCallback:     消息：(Body:'{createTime=2020-09-14 22:48:22, messageId=02007bf3-3eb6-4e53-a539-45c28f012c46, messageData=RabbitMQ}' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回应码：312
ReturnCallback:     回应信息：NO_ROUTE
ReturnCallback:     交换机：directExchange
ReturnCallback:     路由键：directRouting2
--
ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null
  1. 只回调ConfirmCallback函数。
1
2
3
ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

每种情况都会回调ConfirmCallback函数；只有找不到队列时才会回调ReturnCallback函数。

参考链接

代码地址