1. 为什么需要生产者确认机制1.1 一条消息从发送到存储要经过哪几道关卡先聊一个最常见的问题消息到底什么时候算真正发出去了很多刚接触RabbitMQ的同学以为只要basicPublish方法调用不报错消息就稳妥妥地进了队列。这个认知在绝大多数场景下是错的。basicPublish方法的返回值是void它只是把消息交给了客户端内部的Socket缓冲区至于消息最终有没有通过网络到达Broker、有没有被Broker正确路由到队列、有没有在写入磁盘之前就崩溃丢数据调用方一概不知。我用一个实际案例说明一下这个坑有多深。某公司的订单系统用户下单后向RabbitMQ发一条消息通知下游的积分服务给用户加分。某天晚上Broker所在的机器突然断电重启之后运营发现积分少了一大堆。排查后确认RabbitMQ在重启过程中丢了一部分还在内存缓冲区的消息而这些消息在发送端看来basicPublish全部没有报错。这个场景其实涉及消息从生产者到Broker的三个关键环节生产者客户端把消息写入Socket缓冲区此时消息还在进程内消息通过网络传输到BrokerBroker先写入内存此时还没落盘Broker完成持久化如果队列和消息都设置了持久化才真正安全。传统的事务机制txSelect/txCommit可以解决这个问题但代价非常大——事务提交是同步阻塞操作会把信道变成事务模式吞吐量锐减实测在很多机器上比非事务模式慢几十倍以上。所以生产环境几乎没人用事务大家用的都是生产者确认机制Publisher Confirm这是RabbitMQ从3.0版本开始提供的一套轻量级可靠性方案。1.2 生产者确认到底确认了什么生产者确认机制的官方名称是 Publisher Confirms它要解决的是消息从生产者到Broker的可靠投递问题。一旦开启确认模式Broker每收到一条消息都会给生产者回一个确认Ack告诉生产者这条消息我已经收到了并且已经完成必要处理。这里有个非常重要的认知确认只代表Broker接收并处理了这条消息不代表消费者已经消费了它。消费者那边的可靠性由消费者手动Ack机制去保障这是两套完全独立的东西很多人一开始会混淆。具体到确认的内容又分两个层面如果消息被正确路由到了某个队列Broker会回Basic.Ack携带这条消息对应的deliveryTag投递标识如果消息无法被任何队列接收比如RoutingKey没匹配上且没有备份交换机兜底Broker会回Basic.Nack或者Basic.Return配合看具体情况通知生产者这条消息投递失败了。这套机制配合mandatory参数和ReturnListener基本可以覆盖消息是否到了Broker和消息是否被正确路由两个维度的可靠性验证。2. 生产者确认机制的核心运行逻辑2.1 确认模式如何开启生产者确认不是默认开启的需要在生产者端主动声明。在原生Java客户端中代码很简单ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); // 开启生产者确认模式 channel.confirmSelect();核心就是这一行confirmSelect()。调用了它之后当前信道就进入了确认模式。注意这个设置是对信道级别的也就是说同一个连接下你可以一条信道开确认另一条信道不开互不影响。开启之后每条通过这个信道发送的消息都会被Broker分配一个单调递增的deliveryTag从1开始计数。这个Tag是后续处理确认回调的关键依据也是排查问题时的第一线索。Spring Boot环境下更简单在配置里加上一行即可spring: rabbitmq: publisher-confirm-type: correlated这里有个值得注意的配置项类型。Spring Boot 2.x之后publisher-confirms这个老配置被废弃了换成了publisher-confirm-type可选值有none、simple和correlated。none表示不开启simple对应的是同步等待单条确认correlated对应的是异步回调方式。生产环境基本都用correlated后面会解释为什么。2.2 三种确认方式同步、批量、异步确认模式开启之后怎么拿到Broker的确认结果主流有三种方式。第一种同步确认调用channel.waitForConfirms()这个方法会阻塞当前线程直到这条消息收到了Ack或者Nack。channel.basicPublish(, test.queue, null, message.getBytes()); // 阻塞等待确认返回true表示Ackfalse表示Nack boolean confirmed channel.waitForConfirms();如果收到的确认是NackwaitForConfirms返回false如果等待过程中连接断开了会直接抛异常。这里还有一个waitForConfirmsOrDie()方法它更极端——收到Nack或者连接异常时直接抛异常不走返回值判断。同步确认的优点是逻辑极其简单代码可读性高缺点同样是性能差。每发一条消息就要等一个RTT网络往返时间吞吐量直接被拉低。实测对比在同样的网络环境下同步确认方式的吞吐量大约只有异步方式的三十分之一到五十分之一。第二种批量确认channel.basicPublish(, test.queue, null, message.getBytes()); channel.basicPublish(, test.queue, null, message.getBytes()); // 等待当前所有未确认消息的确认结果返回true表示之前所有消息都成功 boolean confirmed channel.waitForConfirms(5000);批量确认是一次性确认此前所有未确认的消息可以传入超时时间。它的优点是每个RTT能确认多条消息性能比同步确认好不少缺点是如果某一条被Nack了你只知道这一批里有失败的消息但不知道具体是哪一条。另外一旦返回false或者超时这一整批消息都得重新发送可能造成大量重复投递。第三种异步确认生产环境首选这才是真正推荐的方式。客户端维护一个回调监听器Broker的确认结果到达后自动触发回调发送方完全不用阻塞。channel.confirmSelect(); channel.addConfirmListener(new ConfirmListener() { Override public void handleAck(long deliveryTag, boolean multiple) throws IOException { // 处理确认成功的消息 } Override public void handleNack(long deliveryTag, boolean multiple) throws IOException { // 处理确认失败的消息 } });这个回调里有几个参数特别关键。deliveryTag是这条被确认消息的编号multiple是批量标志——如果multiple为true意味着所有小于等于这个deliveryTag的消息都已经被确认了。这个设计是为了减少网络开销Broker可以一次性确认一批消息不必每条都单独回一个Ack。在实际开发中multiple的处理逻辑非常影响正确性。如果简单地把deliveryTag当单条消息处理而Broker给了批量确认就会漏掉中间一堆消息的状态记录。正确的做法是维护一个未确认消息的集合比如用SortedSet或ConcurrentSkipListSet收到Ack后根据multiple的值批量移除或者把这一区间内的消息全部标记为已确认。2.3 mandatory参数与ReturnListener的关系生产者确认解决的是消息有没有到Broker但还有一个独立的问题消息到了Broker如果路由不到任何队列怎么办默认情况下路由不到队列的消息会被Broker悄悄丢弃而且生产者感知不到。这个问题和确认机制是两回事——即便消息被丢弃Broker照样会回一条Basic.Ack因为从Broker的角度来看它已经完成了接收和处理流程。要捕获这种路由失败的情况必须设置mandatory参数channel.basicPublish(exchange.direct, unroutable.key, true, null, message.getBytes());第三个参数true就是mandatory。开启后如果消息无法被路由Broker会先把消息退回给生产者通过ReturnListener通知channel.addReturnListener(returnMessage - { System.out.println(消息被退回原因: returnMessage.getReplyText()); System.out.println(交换机: returnMessage.getExchange()); System.out.println(路由键: returnMessage.getRoutingKey()); });有人会问既然有了确认机制为什么还需要区分这两个场景原因在于两者的语义完全不同Ack代表Broker接收成功但接收成功不代表路由成功Return代表路由失败消息会被退回给生产者需要生产者重新处理。正确的可靠性检查应该是两层配合开启confirmSelect管有没有收到开启mandatory管有没有路由出去。缺了任何一层都可能出现消息默默丢失但你完全不知道的情况。3. 代码实操异步确认完整落地3.1 设计一个可复用的异步确认封装直接放一段我在实际项目里用过的封装代码核心思路是用ConcurrentSkipListSet保存未确认消息的deliveryTag收到Broker的确认回调后按multiple标志批量移除已确认的Tag同时对超时未确认的消息做兜底补偿。public class ConfirmProducer implements AutoCloseable { private final Connection connection; private final Channel channel; // 保存未确认的deliveryTag private final ConcurrentSkipListSetLong unconfirmedTags new ConcurrentSkipListSet(); // 保存deliveryTag对应的消息内容用于超时重发 private final ConcurrentHashMapLong, byte[] messageStore new ConcurrentHashMap(); public ConfirmProducer(String host, int port) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(host); factory.setPort(port); this.connection factory.newConnection(); this.channel connection.createChannel(); this.channel.confirmSelect(); this.channel.addConfirmListener(new ConfirmListener() { Override public void handleAck(long deliveryTag, boolean multiple) { if (multiple) { // 批量确认所有小于等于deliveryTag的都已确认 ConcurrentNavigableMapLong, byte[] headMap messageStore.headMap(deliveryTag, true); headMap.keySet().forEach(unconfirmedTags::remove); headMap.clear(); } else { unconfirmedTags.remove(deliveryTag); messageStore.remove(deliveryTag); } } Override public void handleNack(long deliveryTag, boolean multiple) { // 确认失败记录日志并触发重发 if (multiple) { ConcurrentNavigableMapLong, byte[] headMap messageStore.headMap(deliveryTag, true); headMap.forEach((tag, msg) - { System.err.println(消息确认失败tag: tag); // 这里可以走重发逻辑 }); headMap.keySet().forEach(unconfirmedTags::remove); headMap.clear(); } else { System.err.println(消息确认失败tag: deliveryTag); unconfirmedTags.remove(deliveryTag); messageStore.remove(deliveryTag); } } }); } public void send(String exchange, String routingKey, byte[] body) throws IOException { long nextTag channel.getNextPublishSeqNo(); messageStore.put(nextTag, body); unconfirmedTags.add(nextTag); channel.basicPublish(exchange, routingKey, true, null, body); } public void checkTimeoutConfirms(long timeoutMs) { unconfirmedTags.forEach(tag - { // 如果消息发送超过timeoutMs还没有确认需要处理 System.err.println(消息超时未确认tag: tag); // 这里可以重新发送 }); } Override public void close() throws Exception { connection.close(); } }这个封装的几个设计要点channel.getNextPublishSeqNo()可以在发送之前拿到这条消息将要使用的deliveryTag这样就能把Tag和消息内容关联起来Broker回Ack时才知道对应的是哪条消息。用ConcurrentSkipListSet而不是普通Set是因为它支持有序遍历方便multipletrue时批量处理所有小于等于某个Tag的消息。checkTimeoutConfirms是兜底方案因为网络抖动时可能确认消息迟迟不来不能无限等下去必须有个超时机制触发重发。3.2 Spring Boot中的配置与回调实现如果你用的是Spring Boot思路一样只是API变了一下。先配置确认类型spring: rabbitmq: host: localhost port: 5672 publisher-confirm-type: correlated publisher-returns: true template: mandatory: true然后定义回调Configuration public class RabbitConfirmConfig { Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template new RabbitTemplate(connectionFactory); // 消息路由失败时的回调 template.setMandatory(true); template.setReturnsCallback(returned - { System.err.println(路由失败: returned.getReplyText()); System.err.println(交换机: returned.getExchange()); System.err.println(路由键: returned.getRoutingKey()); // 处理退回的消息 }); // 生产者确认回调 template.setConfirmCallback((correlationData, ack, cause) - { if (ack) { System.out.println(消息确认成功: correlationData.getId()); } else { System.err.println(消息确认失败: cause); // 处理失败消息比如重发 } }); return template; } }注意setReturnsCallback和setConfirmCallback这两个回调的触发条件完全不同前面已经分析过这里再次强调确认回调管Broker是否收到返回回调管消息是否被路由到队列。两个回调都要设置否则总有一条消息丢失的场景你感知不到。Spring Boot的correlationData参数非常有用它是你在发送时可以自定义的关联数据。可以在CorrelationData里放业务ID这样确认回调回来的时候你能直接知道是哪个业务操作的消息出问题了CorrelationData correlationData new CorrelationData(orderId); rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);3.3 消息ID与去重设计确认机制用上之后会暴露一个新的问题重发导致的消息重复。假设生产者发送消息后网络闪断Broker其实已经收到了消息并处理了但确认回包在半路丢了。生产者等不到确认超时后选择重发——这时候Broker里就有了两条一模一样的消息。下游消费者如果没做幂等处理就会重复积分、重复发短信、重复扣库存。解决方案是给每条消息设计全局唯一的消息ID消费者端用这个ID做去重。常见做法是先查Redis缓存或数据库唯一索引看这个ID是否已经处理过。// 发送时带上消息ID MessageProperties props new MessageProperties(); props.setMessageId(UUID.randomUUID().toString()); Message message new Message(body.getBytes(), props);这也是为什么前面封装里要维护messageStore的原因——确认失败后重发必须带着原始的消息ID重发而不是新生成一个ID否则消费者端的去重逻辑会失效最终造成重复数据。提示生产者确认机制本身不保证消息不重复它只保证消息被Broker接收到。消息幂等性需要生产者和消费者两端一起设计消费者去重是最后的兜底防线。4. 常见问题与排查技巧实录4.1 消息莫名其妙消失了问题出在哪这是使用生产者确认机制过程中遇到最多的问题。如果你发现消息发出去之后消费者端一直没收到而且确认回调也没报失败从以下几个方向排查第一检查是否只开启了确认没开启mandatory。这种情况下消息如果路由不到队列Broker会静默丢弃但确认回调照样返回成功。你表面上看到消息确认成功实际上消息已经没了。这是最隐蔽的坑。第二检查交换机类型和绑定关系。比如用的是direct类型的交换机路由键写错了一个字符或者队列根本没绑定到这个交换机消息就会被丢弃。用Web管理后台看交换机和队列的绑定关系一条条对。第三检查消息是否设置了过期时间TTL。如果队列设置了x-message-ttl消息在队列里待了超过TTL时间就会变成死信或者被丢弃。确认回调在消息入队时就已经返回Ack了但消息可能在队列里静静过期消费者永远收不到。第四检查消费者是否正常连接。有时候不是生产端的问题是消费者端自己挂了导致消息一直在队列里堆积。这时候看起来消息丢了其实消息还在队列里躺着。4.2 确认回调一直不触发怎么办开了确认模式发了消息结果handleAck和handleNack一个都没执行。这种情况多半是下面几个原因信道没有执行confirmSelect()。有人把参数配在了配置文件里但代码里没有真正调用这个方法确认模式压根没生效使用了错误的确认类型。Spring Boot里如果配置的是publisher-confirm-type: none回调永远不会触发信道已经关闭了。发送消息之后如果信道或连接被意外关闭未收到确认的回调就全部丢失了回调线程里抛了异常导致后续回调被中断。ConfirmListener的回调执行在独立的IO线程里如果你的业务代码抛了未捕获的RuntimeException可能影响后续回调处理。排查时先在回调里加日志确认回调线程有没有被调用。如果有日志但没输出说明模式没开启如果回调线程压根没执行优先查配置和信道状态。4.3 性能断崖式下跌怎么调优开了确认机制之后性能变差是很正常的但差距不能太离谱。如果吞吐量从每秒几千掉到每秒几十那肯定是有问题。最常见的原因是每条消息都用同步确认。这种模式性能天然低下因为每条消息都在等网络往返。优化方向很简单换成批量确认或者异步确认。我实测过同样的消息体、同样的网络环境异步确认的吞吐量能到同步确认的三十倍以上。第二个原因是每条消息都重新创建连接。连接创建的开销比信道创建高一个量级生产环境必须复用连接和信道否则性能会非常难看。第三个原因是批量确认的批次太小。如果用的是批量确认方式每条消息发完就立刻waitForConfirms那批次的优势完全发挥不出来。正确做法是积累一定量比如超过100条未确认或者达到一定时间间隔后再去确认。第四个原因是业务回调里做了重活。有些人收到确认回调后直接在回调里写数据库、调远程服务把IO线程堵死了。确认回调应该是轻量级操作顶多更新一下内存里的状态把真正的后续处理丢给线程池去做。// 错误示范在回调里做耗时操作 Override public void handleAck(long deliveryTag, boolean multiple) { // 直接在这里写数据库阻塞了IO线程 orderService.updateStatus(deliveryTag); } // 正确示范回调里只更新内存状态 Override public void handleAck(long deliveryTag, boolean multiple) { pendingConfirm.remove(deliveryTag); // 其他逻辑交给独立线程池 asyncExecutor.submit(() - orderService.handleConfirm(deliveryTag)); }4.4 生产者确认机制的常见陷阱速查表整理一个实际的排查参考表每一行都是我在项目中见过或者踩过的坑现象可能原因解决方案消息发送无异常消费者收不到未开启mandatory路由失败被丢弃开启mandatory并设置ReturnListener确认回调收到Ack但消息未入队消息过期、队列满了被丢弃检查TTL、队列长度限制、死信配置确认回调收到Nack路由失败、内部异常检查交换机绑定和路由键异步确认性能上不去回调里做了耗时操作回调只做轻量操作重活交给线程池批量确认偶尔丢消息批量中的某条失败整批重发对每条消息保存原始ID重发时保持ID不变连接断开后消息丢失未确认消息随连接一起丢失启动时检测未确认集合触发补偿重发Spring Boot确认回调不触发publisher-confirm-type配置错误改为correlated消息重复投递确认超时重发但Broker已收到消息ID 消费者幂等去重从这个表能看出来生产者确认机制不是孤立的一项配置它和mandatory、消息ID、幂等设计、死信机制这些都要配合起来才能构成一套完整的可靠性方案。我个人在实际操作中的体会是务必把确认机制想成两条腿走路一条腿是confirmSelect管Broker接收另一条腿是mandatory管路由结果。两条腿都站稳了消息才算真正发出去。如果只依赖其中任何一条都会有你不知道的消息处在薛定谔的状态里。5. 生产者确认与事务机制的对比选型5.1 为什么事务机制被冷落RabbitMQ官方其实提供了另一条可靠性路径——事务机制。通过txSelect开启事务模式发送消息之后调txCommit提交事务Broker确认提交成功后才算发送完成。如果中途出现问题可以txRollback回滚。这个机制看起来也很可靠为什么生产环境几乎没人用直接说结论性能代价太大了。事务提交是一个同步阻塞操作每一次txCommit都要阻塞等待Broker的响应相当于把每条消息的发送都变成了一次同步RPC。而且事务模式下消息是一条一条提交的无法批量确认吞吐量下降得非常明显。我用一个不严谨但直观的对比来说明差距同样的机器、同样的消息体事务机制每秒钟大约能发送几百条消息而异步确认机制每秒钟可以发送上万条。在需要高吞吐的消息场景下这个差距是决定性的。RabbitMQ官方文档也明确建议用生产者确认机制替代事务机制确认机制不会阻塞发送性能代价小得多同时可靠性可以做到同样水平。5.2 选型建议根据我的项目经验不同场景下的选择建议低吞吐、对可靠性要求极高的场景比如每分钟几十条的核心配置变更通知用同步确认就够了代码简单逻辑直白常规业务消息场景比如订单通知、状态变更事件用异步确认性能好又能保证不丢消息超高吞吐场景比如日志收集、数据管道用异步确认加批量维护同时配合mandatory做路由校验尽量减少不必要的确认开销极端可靠性要求比如支付回调、金融交易消息确认机制只是起点还要叠加消息ID、持久化、镜像队列、消费者幂等等全套方案。顺便说一下事务机制并非一无是处。如果你需要的是多条消息要么全部成功要么全部失败的原子性语义事务机制还是有它的用武之地的但现实中这种需求非常罕见因为消息中间件本来就不适合做分布式事务的原子提交。5.3 高可用部署下确认机制的表现确认机制和RabbitMQ的高可用方案配合使用时有几个额外的细节要注意。如果是普通集群模式非镜像队列消息只会存在于它被路由到的那个节点上。如果这个节点宕机了消息就丢了但生产者确认在宕机之前可能已经返回了Ack——所以确认机制在高可用集群里并不自动保证消息安全。要真正不丢消息必须使用镜像队列或者仲裁队列Quorum Queue让消息在多个节点上有副本。仲裁队列是RabbitMQ 3.8之后主推的高可用方案底层基于Raft协议在节点故障时能自动选出新主节点并保持消息不丢失。它在确认机制上的表现和普通队列有些差异生产者确认要等消息被多数节点写入之后才返回Ack所以你会觉得确认速度比普通队列慢一些这是正常的也是可靠性换来的代价。如果你的集群是镜像队列模式需要注意版本问题。老版本的镜像队列在故障切换时消息可能因为主从同步延迟而丢失生产者在故障发生瞬间收到Ack但消息随着主节点一起消失了。这个问题在仲裁队列上做了改进所以新项目尽量优先考虑仲裁队列而不是继续用镜像队列。6. 一点实操心得收尾第一次接触生产者确认机制的时候我以为搞懂了confirmSelect就万事大吉结果栽在了multiple参数上。当时维护一个List来记录未确认TagBroker批量确认时只删了回调里那一个Tag导致内存里积压了一堆未确认但实际上已经成功的数据最后补偿重发机制把全部消息又发了一遍下游消费端去重表瞬间被撑爆。从那以后我才真正理解批量确认不是简化逻辑而是对数据结构设计提出了更高的要求。另外一个小建议上生产之前一定要做一次故障演练。把Broker直接停掉观察生产者的确认回调会出现什么现象把消费者的连接断开观察消息在队列里的堆积情况把某个队列删除观察mandatory回调是否触发。只有亲眼见过这些故障场景的表现你才能对自己系统的可靠性格局有一个真实的判断而不是停留在配置看起来没问题的乐观里。确认机制本身不复杂复杂的是把它放进真实的分布式系统里和其他环节正确配合。把各个环节都想清楚消息丢失的噩梦就能远离你。