消息可能丢在三个地方
“发了消息但下游没收到”是 RabbitMQ 最常见的故障描述，但消息丢失可能发生在生产端、Broker 端、消费端三个环节，每一环的排查手段都不一样。生产端：网络中断、Broker 没确认，消息其实没进去；Broker 端：队列不持久、消息不持久、节点故障切换，内存里的消息没了；消费端：自动 ack 后业务抛异常，或者先 ack 再处理，进程一挂消息就丢了。
可靠性不是靠一个配置项解决的，而是三端各做各的保障：生产端确认 + 本地消息表，Broker 端持久化 + 复制，消费端手动 ack + 幂等。代价是吞吐下降、实现变复杂，所以先想清楚哪些消息“丢不起”。
发送侧与 Broker 侧的保障
Spring Boot 3.2 里开启发布确认：
spring:
rabbitmq:
publisher-confirm-type: correlated
publisher-returns: true
template:
mandatory: true
publisher-confirm 只表示 Broker 接收了消息，不代表消息进了队列。routing key 写错、队列没绑定，消息会被直接丢弃。mandatory=true 配合 ReturnsCallback 才能捕获“无法路由”的消息：
@Configuration
public class RabbitConfirmConfig {
@Bean
public RabbitTemplateCustomizer confirmCustomizer(MessageLogService messageLogService) {
return rabbitTemplate -> {
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (!ack && correlationData != null) {
messageLogService.markFailed(correlationData.getId(), cause);
}
});
rabbitTemplate.setReturnsCallback(returned ->
messageLogService.markUnroutable(
returned.getMessage().getMessageProperties().getMessageId(),
returned.getReplyText()));
};
}
}
只记日志还不够。确认失败的消息要落库重发，否则日志一滚就找不回来。更稳的做法是本地消息表：业务写库和消息记录在同一个本地事务里，定时任务扫描待发送记录，发送成功再更新状态。这样即使 Broker 短暂不可用，消息也不会丢，代价是消费端必须幂等。
Broker 端：持久化和复制
交换机、队列都要声明 durable，消息要设置持久化：
Message message = MessageBuilder.withBody(bytes)
.setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setDeliveryMode(MessageDeliveryMode.PERSISTENT)
.setMessageId(orderId.toString())
.build();
rabbitTemplate.send(RabbitConfig.ORDER_EXCHANGE, RabbitConfig.ORDER_ROUTING_KEY, message);
只设队列 durable 不够，消息本身不持久化，重启后队列还在、消息没了。普通经典队列在主节点故障时，未同步到从节点的消息仍可能丢；对可靠性要求高的业务应使用 quorum 队列，它基于 Raft 复制，已确认的消息不会因为单节点故障消失。quorum 队列的吞吐低于经典队列，适合订单、支付这类不能丢的消息，不适合日志、埋点这类可丢数据。
消费端：ack 的时机决定一切
自动 ack 模式下，消息一投递就被标记为已消费，业务还没执行完，进程重启消息就没了。可靠消费要用手动 ack，并且 业务成功后再 ack：
@RabbitListener(queues = RabbitConfig.ORDER_QUEUE, ackMode = "MANUAL")
public void onMessage(OrderCreatedMessage message, Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
try {
orderHandler.handle(message);
channel.basicAck(deliveryTag, false);
} catch (BusinessException e) {
log.error("订单消息处理失败, orderId={}", message.orderId(), e);
channel.basicNack(deliveryTag, false, false);
}
}
basicNack 的 requeue 传 true 要谨慎：可重试的异常（数据库连接超时）可以重新入队，不可重试的异常（参数非法）重新入队只会无限循环。更常见的方案是配置重试次数和死信队列，超过次数后转人工处理。
重复消费无法避免，RabbitMQ 保证的是 at-least-once。消费逻辑必须幂等，最简单的是唯一索引 + 插入判重：
CREATE TABLE mq_consume_log (
message_id VARCHAR(64) NOT NULL,
consumer_group VARCHAR(64) NOT NULL,
consumed_at DATETIME NOT NULL,
PRIMARY KEY (message_id, consumer_group)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4;
@Transactional(rollbackFor = Exception.class)
public void handle(OrderCreatedMessage message) {
int inserted = consumeLogMapper.insertIgnore(message.messageId(), "order-consumer");
if (inserted == 0) {
log.info("消息已消费, messageId={}", message.messageId());
return;
}
// 真正的业务处理
}
常见坑
1. 先 ack 再处理业务
这是最典型的丢消息写法。ack 之后数据库事务回滚、进程崩溃，Broker 都认为消息已经消费成功。ack 必须放在业务成功之后，并且要等事务提交完成。
2. 事务方法和 ack 的顺序搞反
在 @Transactional 方法里 ack，方法返回后事务才提交；如果提交失败，消息已经被确认。要么把 ack 放到事务外层，要么用 TransactionSynchronization#afterCommit 再确认。
3. requeue=true 造成死循环
不可重试的异常反复入队，CPU 和日志被打爆，队列永远消费不完。区分可重试与不可重试异常，配置最大重试次数和死信队列。
4. 只配了 confirm，没配 returns
消息没路由到任何队列时，Broker 仍然会 confirm，生产端以为发送成功。mandatory 和 ReturnsCallback 要一起配。
5. 消费端反序列化失败后无限重试
消息体格式变更、类字段删除都可能导致反序列化异常。给消息加版本号，消费端兼容旧版本；无法处理的消息转入死信队列，不要让它卡住整个队列。
排查步骤
先确认消息卡在哪一段：
rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers
rabbitmqctl list_connections name state channels
rabbitmqctl list_exchanges name type durable
messages_ready 持续增长说明没有消费者或消费能力不足；messages_unacknowledged 高说明消费者拿了消息不确认，通常是业务阻塞或 ack 逻辑有问题。再看应用日志里的 confirm 回调和 returns 回调，确认失败的消息有没有落库重发。需要看消息级链路时，可以临时开启 firehose tracer，但不要在生产长期开启，开销很大。
小结
消息可靠性是一个端到端的工程：生产端确认并落库，Broker 端持久化并复制，消费端手动 ack 并幂等。任何一端偷懒，另外两端做得再好也补不回来。先给消息分级，再决定投入多少可靠性成本。
评论