RabbitMQ消息的发布确认机制详解

RabbitMQ发布确认机制确保消息从生产者成功传输到交换机和队列，提高系统可靠性。在Spring Boot项目中，通过配置publisher-confirm-type和publisher-returns，启用发布确认和消息返回机制。配置RabbitTemplate的确认回调和返回回调，可以捕捉消息传输状态，处理不同传输结果。测试场景包括消息无法到达交换机、消息到达交换机但无法到达队列以及消息成功到达队列。通过合理设置和优化，可以确保高并发环境下的消息可靠传输，适用于金融支付、电商系统等对消息传输可靠性要求高的场景。

1. RabbitMQ发布确认机制概述

发布确认（Publisher Confirms）是RabbitMQ提供的一种机制，用于确保消息从生产者发送到RabbitMQ服务器并被成功处理。与事务机制不同，发布确认的性能开销更小，非常适合高吞吐量的场景。发布确认机制提供了两种类型的确认：

• 消息到达交换机（Exchange）后的确认
• 消息从交换机路由到队列（Queue）后的确认

2. 配置文件中添加发布确认相关配置

在Spring Boot项目中，通过配置文件来启用发布确认机制非常方便。以下是需要添加到application.properties或application.yml中的配置：

# 消息到达交换机后会回调发送者
spring.rabbitmq.publisher-confirm-type=correlated
# 消息无法路由到队列时回调发送者
spring.rabbitmq.publisher-returns=true
1
2
3
4

配置解释：

• publisher-confirm-type：设置为correlated表示使用CorrelationData来关联确认与发送的消息。
• publisher-returns：设置为true表示启用消息返回机制，当消息无法路由到队列时会触发回调。

3. 发布确认类型

在Spring AMQP中，发布确认类型通过ConfirmType枚举类来定义：

public enum ConfirmType {
    SIMPLE,     // 使用 RabbitTemplate#waitForConfirms() 或 waitForConfirmsOrDie()
    CORRELATED, // 使用 CorrelationData 关联确认与发送的消息
    NONE        // 不启用发布确认
}
1
2
3
4
5

4. 配置RabbitTemplate

为了使用发布确认机制，需要配置RabbitTemplate，包括设置确认回调和返回回调：

@Slf4j
@Configuration
public class RabbitTemplateConfig {

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);

        // 设置mandatory为true，当找不到队列时，broker会调用basic.return方法将消息返还给生产者
        rabbitTemplate.setMandatory(true);

        // 设置确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息已经到达Exchange");
            } else {
                log.info("消息没有到达Exchange");
            }
            if (correlationData != null) {
                log.info("相关数据：" + correlationData);
            }
            if (cause != null) {
                log.info("原因：" + cause);
            }
        });

        // 设置返回回调
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息无法到达队列时触发");
            log.info("ReturnCallback:     " + "消息：" + message);
            log.info("ReturnCallback:     " + "回应码：" + replyCode);
            log.info("ReturnCallback:     " + "回应信息：" + replyText);
            log.info("ReturnCallback:     " + "交换机：" + exchange);
            log.info("ReturnCallback:     " + "路由键：" + routingKey);
        });

        return 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
36
37
38
39
40

5. 配置测试交换机和队列

为了测试发布确认机制，我们需要配置相应的交换机和队列：

@Slf4j
@Configuration
public class ConfirmConfig {

    @Bean
    public Queue confirmQueue() {
        return new Queue(Constant.CONFIRM_QUEUE, false);
    }

    @Bean
    DirectExchange confirmExchange() {
        DirectExchange directExchange = new DirectExchange(Constant.CONFIRM_EXCHANGE, false, false);
        directExchange.addArgument("alternate-exchange", Constant.CONFIRM_BACKUP_EXCHANGE);
        return directExchange;
    }

    @Bean
    Binding bindingConfirm() {
        return BindingBuilder.bind(confirmQueue()).to(confirmExchange()).with(Constant.CONFIRM_ROUTING_KEY);
    }

    @Bean
    FanoutExchange backupExchange() {
        return new FanoutExchange(Constant.CONFIRM_BACKUP_EXCHANGE, false, false);
    }

    @Bean
    public Queue backupQueue() {
        return new Queue(Constant.CONFIRM_BACKUP_QUEUE, false);
    }

    @Bean
    public Queue warningQueue() {
        return new Queue(Constant.CONFIRM_WARNING_QUEUE, false);
    }

    @Bean
    Binding bindingConfirmBackup() {
        return BindingBuilder.bind(backupQueue()).to(backupExchange());
    }

    @Bean
    Binding bindingConfirmWarning() {
        return BindingBuilder.bind(warningQueue()).to(backupExchange());
    }
}
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
36
37
38
39
40
41
42
43
44
45
46

6. 测试场景及分析

6.1 消息无法到达交换机

测试代码：

@Autowired
RabbitTemplate rabbitTemplate;
String msg = "一条用于发布确认的消息";

@GetMapping("/noExchange")
public void noExchange() {
    rabbitTemplate.convertAndSend("noExchange", "noExchange", msg);
}
1
2
3
4
5
6
7
8

配置了rabbitTemplate.setMandatory(true)，当消息无法到达交换机时会回调：

ConfirmCallback 消息没有到达Exchange
ConfirmCallback 原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'noExchange' in vhost '/', class-id=60, method-id=40)
1
2

6.2 消息到达交换机但无法到达队列

测试代码：

@GetMapping("/toExchange")
public void toExchange() {
    rabbitTemplate.convertAndSend(Constant.CONFIRM_EXCHANGE, "xxx.xxx.xxx", msg);
}
1
2
3
4

输出：

ConfirmCallback 消息已经到达Exchange
1

没有收到无法到达队列的消息，是因为配置了备份队列，消息被路由到了备份队列。

6.3 注掉备份队列再试

修改配置：

@Bean
DirectExchange confirmExchange() {
    DirectExchange directExchange = new DirectExchange(Constant.CONFIRM_EXCHANGE, true, false);
    return directExchange;
}
1
2
3
4
5

测试结果：

消息无法到达队列时触发
ReturnCallback:     消息：(Body:'一条用于发布确认的消息' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回应码：312
ReturnCallback:     回应信息：NO_ROUTE
ReturnCallback:     交换机：myConfirmExchange
ReturnCallback:     路由键：xxx.xxx.xxx
ConfirmCallback 消息已经到达Exchange
1
2
3
4
5
6
7

此时，ConfirmCallback和ReturnCallback都被调用了。

6.4 成功到达队列

测试代码：

@GetMapping("/toQueue")
public void toQueue() {
    rabbitTemplate.convertAndSend(Constant.CONFIRM_EXCHANGE, Constant.CONFIRM_ROUTING_KEY, msg);
}
1
2
3
4

输出：

ConfirmCallback 消息已经到达Exchange
1

7. 发布确认流程

下图展示了RabbitMQ发布确认流程：

8. 深入解析RabbitMQ发布确认机制

8.1 事务机制与发布确认机制的对比

事务机制和发布确认机制都是确保消息可靠投递的手段，但它们在实现和性能方面有明显区别：

• 事务机制：通过txSelect、txCommit和txRollback实现，性能开销较大，不适合高并发场景。
• 发布确认机制：通过异步确认消息是否成功到达交换机和队列，性能开销小，适合高并发场景。

8.2 发布确认机制的优缺点

优点

1. 性能高：相比事务机制，发布确认机制对性能的影响较小。
2. 异步处理：使用回调函数处理确认结果，不阻塞消息发送。
3. 可靠性高：确保消息成功到达交换机和队列，提高系统可靠性。

缺点

1. 实现复杂：需要配置和处理回调函数，增加了代码复杂度。
2. 延迟高：确认机制引入了额外的网络延迟。

8.3 发布确认机制的应用场景

1. 金融支付系统：确保支付消息的可靠传输，避免重复支付或支付丢失。
2. 电商系统：确保订单消息的可靠传输，避免订单丢失或重复处理。
3. 日志系统：确保日志消息的可靠传输，避免日志丢

失。

8.4 发布确认机制的最佳实践

1. 合理设置超时时间：在高并发场景下，设置合理的超时时间，避免消息发送阻塞。
2. 优化回调函数：回调函数中避免复杂逻辑，确保回调处理快速完成。
3. 监控和报警：建立监控机制，及时发现和处理消息投递失败问题。

9. 总结

本文详细介绍了RabbitMQ消息的发布确认机制，包括配置、实现及其在不同场景下的表现。通过合理配置和使用发布确认机制，可以有效提高消息传输的可靠性，确保消息在高并发环境下的可靠投递。希望本文能够帮助读者深入理解并应用RabbitMQ的发布确认机制，提高系统的可靠性和性能。