一、消息丢失的可能场景

消息从生产者到消费者，经过以下流程：

每一步都可能出问题：

因此，可靠性保障需要三管齐下：

1. ✅ 确保生产者一定把消息发送到 MQ

2. ✅ 确保 MQ 不会丢失消息

3. ✅ 确保消费者一定成功处理消息

二、生产者可靠性保障

2.1 生产者重试机制（应对网络抖动）

当RabbitTemplate与 MQ 连接超时时，SpringAMQP 提供阻塞式重试机制。

配置application.yml：

yaml

spring: rabbitmq: connection-timeout: 1s # 连接超时时间 template: retry: enabled: true # 开启重试 initial-interval: 1000ms # 初始等待时间 multiplier: 1 # 等待时长倍数 max-attempts: 3 # 最大重试次数

⚠️注意：重试是阻塞的，会占用当前线程。对性能敏感的业务，建议禁用重试或用异步线程发送。

2.2 生产者确认机制（应对路由失败 / MQ 内部异常）

RabbitMQ 提供两种确认机制：

开启确认机制：

spring: rabbitmq: publisher-confirm-type: correlated # 异步回调 publisher-returns: true # 开启 Return 机制

2.2.1 配置 ReturnCallback（统一处理路由失败）

每个RabbitTemplate只能配置一个ReturnCallback，建议在配置类中统一设置：

@Slf4j @Configuration @AllArgsConstructor public class MqConfig { private final RabbitTemplate rabbitTemplate; @PostConstruct public void init() { rabbitTemplate.setReturnsCallback(returned -> { log.error("触发 return callback"); log.debug("exchange: {}", returned.getExchange()); log.debug("routingKey: {}", returned.getRoutingKey()); log.debug("message: {}", returned.getMessage()); log.debug("replyCode: {}", returned.getReplyCode()); log.debug("replyText: {}", returned.getReplyText()); }); } }

当路由失败时，日志会输出类似：

replyCode: 312 replyText: NO_ROUTE

2.2.2 配置 ConfirmCallback（按消息处理回执）

由于每条消息的处理逻辑可能不同，ConfirmCallback在每次发送时动态定义。

发送消息并添加回调：

@Test void testPublisherConfirm() { // 1. 创建 CorrelationData（包含唯一 id） CorrelationData cd = new CorrelationData(); // 2. 添加 ConfirmCallback cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() { @Override public void onFailure(Throwable ex) { log.error("发送消息异常", ex); } @Override public void onSuccess(CorrelationData.Confirm result) { if (result.isAck()) { log.debug("收到 ACK，消息发送成功"); } else { log.error("收到 NACK，发送失败，原因：{}", result.getReason()); } } }); // 3. 发送消息（携带 CorrelationData） rabbitTemplate.convertAndSend("hmall.direct", "q", "hello", cd); }

2.2.3 回执结果分析

💡建议：大多数业务无需开启生产者确认，因为路由失败和 Exchange 错误通常是编程问题，可以在开发阶段规避。仅在极高可靠性要求的业务中开启，且只处理NACK即可。

六、总结：最佳实践清单