当前位置：首页 > article >正文

Spring Boot 集成 RabbitMQ 并实现消息确认机制

article 2025/2/21 18:01:36

1. 引言

在 Spring Boot 项目中，RabbitMQ 是一种常见的消息中间件，用于实现异步通信。为了确保消息可靠传输，我们需要配置 RabbitTemplate，并实现 ConfirmCallback 和 ReturnsCallback 两种回调机制。

本文将介绍如何在 Spring Boot 中

2. RabbitMQ 配置类

首先，我们创建 RabbitMqConfig 配置类：

集成 RabbitMQ，并详细讲解 RabbitTemplate 的使用。

@EnableRabbit
@Configuration
public class RabbitMqConfig {

    private static Logger log = LoggerFactory.getLogger(RabbitMqConfig.class);

    @Bean
    RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        
        // 必须设置为 true，否则消息投递失败不会触发 ReturnCallback
        rabbitTemplate.setMandatory(true);

        // 消息发送确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            String messageId = correlationData != null ? correlationData.getId() : "null";
            if (ack) {
                log.info("消息成功发送到交换机，消息ID：{}", messageId);
            } else {
                log.error("消息发送到交换机失败，消息ID：{}，原因：{}", messageId, cause);
            }
        });

        // 消息路由失败回调
        rabbitTemplate.setReturnsCallback(returned -> {
            String correlationId = returned.getMessage().getMessageProperties()
                    .getHeader("spring_returned_message_correlation");
            log.error("消息发送失败，ID：{}，应答码：{}，原因：{}，交换机：{}，路由键：{}",
                    correlationId, returned.getReplyCode(), returned.getReplyText(),
                    returned.getExchange(), returned.getRoutingKey());
        });
        
        return rabbitTemplate;
    }
}

3. 消息投递的四种情况

在使用 RabbitMQ 时，消息投递可能遇到以下四种情况：

消息推送到 RabbitMQ 服务器，但找不到交换机
- ConfirmCallback 被触发，ack = false
消息推送到服务器，找到交换机，但找不到队列
- ConfirmCallback 被触发，ack = true
- ReturnsCallback 也会被触发
消息推送到服务器，既找不到交换机，也找不到队列
- ConfirmCallback 被触发，ack = false
消息推送成功
- ConfirmCallback 被触发，ack = true

4. 发送消息示例

@Autowired
private RabbitTemplate rabbitTemplate;

public void sendMessage(String exchange, String routingKey, String message) {
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
    log.info("发送消息：{}，交换机：{}，路由键：{}，消息ID：{}", message, exchange, routingKey, correlationData.getId());
}