做消息中间件，最怕听到的就是“消息丢了”四个字。我做了几年消息系统，线上问题里十个有八个最后都指向同一个源头——生产者这边根本不知道自己发出去的消息到底有没有进到 Broker。RabbitMQ 的“生产者确认机制”（Publisher Confirm）就是专门补这条信息链路的。这篇文章不打算从 AMQP 协议抄定义，而是把我在真实项目里用 RabbitMQ 排查消息丢失时验证过的那套思路完整写下来：三种确认模式怎么选、代码怎么写、Spring Boot 里怎么配、以及哪些文档里不会写的坑，一次性讲透。
1. 生产者确认机制的设计逻辑与核心概念
1.1 一条消息从发出到落地，中间到底断在哪里
先还原一下完整链路：生产者把消息交给 Channel，Channel 发到 Broker 的 Exchange，Exchange 根据 RoutingKey 把消息路由到一个或多个 Queue，Queue 再根据持久化配置决定是否写磁盘，最后才是消费者拉取。请记住，这里每一步都有可能出问题：
- 网络抖动，报文根本没到 Broker；
- 交换机内部处理失败，消息被丢弃；
- 路由不到任何队列，消息被静默清除；
- 队列没有持久化，Broker 重启后内存数据全部蒸发。
在没有确认机制时，生产者调用完 basicPublish 就认为结束了，这其实和“把信投进邮筒就不再管”是一样的。RabbitMQ 官方推荐用 Publisher Confirm 来补救：它让 Broker 在成功接收并处理消息后，通过 Channel 回一个明确的 ack。这个 ack 不是消费端的 ack，而是专门针对生产者的写入确认，本质是一个异步回调，告诉生产者“你刚才那份数据我收下了，而且已经按当前队列策略做完了该做的持久化动作”。
1.2 Confirm 机制到底确认了什么
要注意，RabbitMQ 的确认并不保证“消息已经刷到物理磁盘”。对普通经典队列来说，确认是在消息被写入队列（包括内存和异步落盘逻辑）后发出；对 Quorum Queue 来说，要等 Raft 组完成多数副本提交，可靠性等级完全不同。也就是说，确认的是“Broker 已接收并完成本阶段持久化要求”，不是“灾难一定不会丢”。这一点在架构评审时一定要向团队讲清楚，否则容易让同事产生“开了 confirm 就绝对不丢”的错误安全感。
在 Java 客户端里，开启确认只需一行channel.confirmSelect()，之后每个发布的序号（deliveryTag）会单调递增。Broker 返回的回执有几类：
basic.ack：正常确认；basic.nack：Broker 内部处理失败；basic.return：带 mandatory 发布但没有任何队列接收时，消息被退回。
很多新手以为配置了 confirm 就万事大吉，实际上 mandatory 和 return 的配合才是保证“不丢”的关键，后面第 4 节会有专门的排查记录。
1.3 和 Kafka、RocketMQ 对比一下确认语义
既然聊可靠性，就免不了横向比一比。Kafka 的 producer 配置里acks=1表示 leader 写入成功就返回，acks=-1（all）表示 ISR 全部同步完才返回，这是一个“多档位”的确认模型。RocketMQ 则是同步发送时通过 SendResult 判断 sendStatus，事务消息还有半消息补偿机制。
RabbitMQ 确认机制最有特点的地方，是“按消息粒度 + 每 Channel 序号”的回执模型。你有能力精确知道哪一条失败，这意味着重试、幂等、补偿可以设计得很细。代价是它不像 Kafka 那样一个参数就能切换所有节点确认，你的存储形态（经典队列或 Quorum Queue）会直接影响确认语义。
选型时我个人的经验是：如果业务需要复杂路由，RabbitMQ 的可靠投递链路配合 confirm + return 是够用的；如果追求极致的顺序和分区吞吐，Kafka 的 acks 模型更适合。至于 RocketMQ，在事务消息和延迟消息上更有优势。三者的对比热度一直很高，但真正决定选型的还是业务形态，而不是社区里谁嗓门大。
2. 三种确认模式：选型、代码与性能权衡
2.1 同步单条确认
最直接的模式：把channel.waitForConfirms()放在每次 publish 之后，每发一条就同步等回执。
channel.confirmSelect(); for (int i = 0; i < 1000; i++) { channel.basicPublish(EXCHANGE, ROUTING_KEY, null, ("msg-" + i).getBytes()); boolean ok = channel.waitForConfirms(5000); // 5s 超时 if (!ok) { // 重新投递或落异常表 } }优点是确认粒度最细，哪条失败就处理哪条。缺点也明显，RPC 式往返把吞吐压得很低。我压测的时候，单线程同步一条条确认大概只能跑出每秒几百到一两千的吞吐，具体看网络 RTT。网络越远越惨，跨机房的场景基本不能用。
适用场景：低频、对每条数据可靠性都极度敏感，比如充值回调、证件状态变更这类量级几百 QPS 以内的接口。除此之外不建议在消息流式场景里用它。
2.2 批量确认
把waitForConfirms()攒到一批消息之后调用，比如每发 100 条确认一次：
channel.confirmSelect(); for (int i = 0; i < 1000; i++) { channel.basicPublish(EXCHANGE, ROUTING_KEY, null, ("msg-" + i).getBytes()); if (i % 100 == 99) { boolean ok = channel.waitForConfirms(5000); if (!ok) { // 整批重发 } } }批量确认把每条消息的 RTT 摊薄，吞吐可以提升好几倍。但坏消息是：只要这一批里有一条 nack，你并不知道具体哪几条失败了，只能整批重发。对很多业务来说，重发会引入重复消息，下游必须做幂等。如果你能接受“按批次重放”，批量确认是性价比不错的选择；如果业务对重复零容忍，还是老实走异步确认。
2.3 异步监听确认：生产环境的首选
异步确认的核心是给 Channel 注册 ConfirmListener，让 ack/nack 在后台线程回调，主线程只负责发消息，不阻塞。代码如下：
Channel channel = connection.createChannel(); channel.confirmSelect(); final SortedMap<Long, String> pending = new ConcurrentSkipListMap<>(); channel.addConfirmListener( // handleAck (deliveryTag, multiple) -> { if (multiple) { ConcurrentNavigableMap<Long, String> confirmed = pending.headMap(deliveryTag, true); confirmed.clear(); } else { pending.remove(deliveryTag); } }, // handleNack (deliveryTag, multiple) -> { if (multiple) { ConcurrentNavigableMap<Long, String> failed = pending.headMap(deliveryTag, true); failed.forEach((tag, msg) -> handleRetry(msg)); failed.clear(); } else { String msg = pending.remove(deliveryTag); handleRetry(msg); } } ); for (int i = 0; i < 10000; i++) { long tag = channel.getNextPublishSeqNo(); pending.put(tag, "payload-" + i); channel.basicPublish(EXCHANGE, ROUTE_KEY, null, ("payload-" + i).getBytes()); }几个细节必须交代清楚。
第一，pending 集合要放在发布前插入，而不是发布后。因为确认回调是异步的，极端情况下消息刚发完，Broker 的 ack 就已经到达客户端，回调线程可能抢在主线程 put 之前执行 remove，导致 pending 数据错乱。先 put 再发可以保证回调执行时 key 一定存在。
第二，multiple=true是 RabbitMQ 批量确认的关键优化。Broker 会一次性告诉客户端“这个序号之前的所有消息都成功了”，所以 pending 要用headMap按序号批量清理，不要傻傻地按 deliveryTag 一条一条 remove。
第三，nack 出现时不要盲目死循环重发。建议落一张本地异常表，或者按约定退避重试，否则持续重发会把 Broker 打趴。
2.4 三种模式的选型对比
| 模式 | 吞吐量 | 定位准确性 | 复杂度 | 适合场景 |
|---|---|---|---|---|
| 单条同步确认 | 低（百~千级/秒） | 精确到条 | 最低 | 低频关键接口 |
| 批量确认 | 中（数千/秒） | 按批次 | 低 | 允许批量重放的日志聚合 |
| 异步监听确认 | 高（万级/秒） | 精确到条 | 中 | 中高频业务主链路 |
补充一句：不要只盯着数字天花板，还要看消息体大小和网络 RTT。我压测时 512 字节小报文，异步模式单 channel 能跑到 1-2 万/秒；换 100KB 大报文，直接掉到两三千。生产上先用小报文跑通链路，再加真实报文测。
3. 从原生 API 到 Spring Boot 的落地实操
3.1 原生 API 的完整骨架
先串一个不会出错的经典链路：创建连接、开启 confirm、发送、确认回执。
ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { channel.confirmSelect(); channel.basicPublish("exchange.direct", "order.created", null, "hello".getBytes()); if (channel.waitForConfirms(3000)) { System.out.println("publish confirmed"); } else { System.out.println("publish not confirmed"); } }注意 try-with-resources 里连接和 channel 一起关，释放顺序是先 channel 后 connection。线上高并发环境不要用这种最简写法的全局单 channel，要配合连接池或异步线程发送，否则 channel 单点压满会造成性能瓶颈。
3.2 Spring Boot 配置与回调
Spring Boot 工程里更常用 RabbitTemplate。这里有两个容易踩的版本坑：早期版本用spring.rabbitmq.publisher-confirms=true，Spring Boot 2.2 之后改成publisher-confirm-type。推荐相关配置如下。
spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest publisher-confirm-type: correlated # 开启发送端确认 publisher-returns: true # 开启消息路由失败回调 template: mandatory: true # 路由不到队列时把消息退还生产者Java 回调代码：
@Configuration public class RabbitMqConfirmConfig { @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template = new RabbitTemplate(connectionFactory); template.setMandatory(true); template.setConfirmCallback((correlationData, ack, cause) -> { if (!ack) { log.error("消息未确认，cause={}", cause); // 落库、告警或补偿 } }); template.setReturnsCallback(returned -> { log.error("路由未找到队列，消息退还，exchange={}, routingKey={}, message={}", returned.getExchange(), returned.getRoutingKey(), returned.getMessage()); }); return template; } }CorrelationData 是你自己传的业务标识。发送时：
CorrelationData cd = new CorrelationData(UUID.randomUUID().toString()); template.convertAndSend("exchange.direct", "order.created", orderMsg, cd);回调里可以通过cd.getId()映射到业务消息，做到精确追踪。Spring 这种封装的好处是回调线程安全细节都处理好了；坏处是刚开始排查问题的时候容易陷在框架回调里看不清原理。所以我建议先在原生客户端把链路读懂，再看 Spring 封装，这样遇到问题不至于瞎猜。
3.3 确认、持久化、消费 ack 三者配套
很多团队只开了 confirm，认为消息就不会丢了，这里有个大坑。丢消息其实是三道阀门：生产端确认、队列和消息持久化、消费端手动 ack。
- 生产端确认解决的是“发送阶段是否成功”。
- 队列
durable=true加消息持久化（deliveryMode=2）解决的是“Broker 重启后是否还在”。 - 消费端手动 ack 解决的是“处理完之后再从队列删除”。
任何一道没做对，都不能说自己消息不丢。比如队列是 durable 但消息发布没设置持久化，Broker 重启后消息直接蒸发；比如消费者开了 autoAck，消息刚消费但业务逻辑异常，队列视为已消费，消息就没了。我这几年见到的“丢消息”case，超过一半不是 confirm 没开，而是这三道阀门少关了一道。
Quorum Queue 是 RabbitMQ 3.8 之后主推的可靠性队列，对高可用要求更严的场景可以直接把队列类型配成 quorum。它的确认语义会和经典队列有差异，最直观的表现是发布确认的耗时稍微变长，因为要等多数副本完成同步。对吞吐要求高的二进制日志场景，用经典队列资源更省。
4. 真实环境常见问题与排查记录
4.1 收到 ack 却丢消息：忘配 mandatory
这是新手最容易踩的坑：confirm 回调正常返回 ack，但消息根本没进队列。原因在于，如果发布时没有把 mandatory 设为 true，交换机路由不到任何队列时，Broker 会直接丢弃这条消息，但仍然给你一个 ack——它认为“发布操作成功完成，只是没人要”。逻辑就是这么讽刺。
解法就一条：mandatory + returns 回调一起开。一旦路由不到队列，Broker 会把消息退回生产者，并触发 return 回调，在 return 回调里做补偿或告警。Spring 里只要配置mandatory: true，并实现 ReturnsCallback 即可。
还有个细节：开了 mandatory 之后，return 回调执行顺序可能在 ack 回调之后，不要依赖回调时序做判断，要在两个回调里分别打印关键日志。我见过有人只在 ack 回调里做业务处理，结果 mandatory 一直没生效，消息静默丢了半个月才发现。
4.2 confirm 超时与 pending 无限膨胀
异步模式下最常见的问题就是 pending 集合不断变大。常见原因有三个。
第一，Broker 磁盘 IO 被打满，确认消息排不上队。排查命令是rabbitmqctl list_queues name messages messages_unacknowledged，再看服务器iostat，经常能看到磁盘 util 100% 的情况。
第二，单 channel 发送速度超过了处理速度，导致 Broker 回执积压。客户端这边确认线程都在跑，如果积压严重，内存和 map 都会涨。解决办法是加限流（比如信号量），或者多开几个 channel 分摊。
第三，连接假死。TCP 连接还在，但双方已经收不到对端报文。这种问题要用心跳参数兜底，connectionFactory.setRequestedHeartbeat(30)，同时客户端主动检测 pending 大小，超过阈值就重置连接或告警。
4.3 性能损耗到底有多大
很多人在方案评审时担心 confirm 机制拖慢生产速度。我直接把压测典型数据列出来，供参考。小消息 512B、单客户端、经典队列、不开 confirm 能跑到 3 万/秒以上；开异步 confirm 后大约 1.5 到 2 万/秒；开单条同步 confirm 大概 800 到 1500/秒。批量 confirm 介于两者之间，具体数值取决于批量大小。
所以选择依据很清晰：如果你的链路本身就只有几千 QPS，开同步确认完全够；如果量很大，用异步确认，别为省那点性能关掉可靠性开关。真到了同步确认跑不满业务的情况，第一优化方向应该是批量发送和连接池，而不是关闭 confirm。
4.4 Docker 部署和管理界面那些坑
顺着大家高频搜的问题也提一嘴。很多人用 Docker 部署 RabbitMQ，管理界面能打开，但 admin 账号创建 virtual host 时报错，大概率不是 RabbitMQ 坏了，而是 default user 的权限正则没配好。默认 guest 用户只能在 localhost 访问，跨容器访问要新建用户并给足 permissions，正则至少写成^$或者具体 vhost 名。这些都是部署侧的小问题，不影响确认机制本身，但排查时容易被误导。
还有遇到 quorum queue 配置后，发布确认明显变慢的情况。这是 quorum 队列的同步语义导致的，不是故障。判断方法很简单：同一条链路，临时建一个 classic 队列测试吞吐，对比确认耗时分布。
5. 实战中反复踩坑后的一些体会
5.1 可靠投递的几个实操习惯
做 RabbitMQ 生产链路这几年，我认为“生产者确认机制”是性价比最高的一个高级特性。它不复杂，却能把消息丢失问题从“黑盒”变成“可追踪”。实际落地时我有几个不成文的习惯：
- 所有核心发送通道强制开启 confirm + mandatory + returns 三件套，并把回调日志纳入监控，nack 或 return 必须有值班告警；
- 异步确认的 pending 集合绝不裸奔，给它加阈值监控，超过 10000 条就报警；
- 不要试图用确认机制解决所有丢消息问题，它只是第一道门，消费端 ack、队列持久化、备份机制都要同时做。
踩过几次坑之后你会认同一个结论：消息可靠性从来不是某一个开关，而是一整套链路设计。
5.2 排查链路的黄金原则
最后分享一个小技巧：排查消息丢失时，先在发送端打一条带唯一 id 的日志，再到管理界面看 queue 里的 message count，最后看消费端消费日志。三段日志对得上，问题就到不了 Broker 这一层；对不上，顺藤摸瓜反而更快。
如果某条消息发送端显示 ack 了、队列里也看到了，但消费者就是没收到，那问题大概率出在消费端 ack 和重试逻辑上，而不是生产确认。如果发送端没有 ack，但管理界面里消息数在涨，那就要检查 ConfirmListener 有没有注册成功、版本是否把 confirm 开关配置对了。这套方法帮我省了很多半夜看日志的时间，强烈建议你也搭一套起来。
