文章目录RabbitMQ 发布确认模式Publisher Confirms一、为什么需要发布确认二、开启确认模式三、三种确认策略3.1 单条同步确认3.2 批量同步确认推荐入门3.3 异步回调确认生产推荐四、关键细节与常见坑4.1 deliveryTag 在 Confirm 模式下是发布序号4.2 序号必须在发送前获取4.3 multipletrue 时的清理逻辑4.4 nack 的常见原因4.5 同步 vs 异步的选择五、三种模式对比总结六、延伸阅读RabbitMQ 发布确认模式Publisher Confirms一、为什么需要发布确认在简单工作队列中生产者把消息basicPublish出去就认为发送成功。但这条消息可能在中途丢失网络抖动Broker 没收到交换机不存在消息被丢弃队列满了消息被拒绝发布确认Publisher Confirms是 AMQP 0-9-1 的扩展机制让 Broker 主动告诉生产者“这条消息我已经收到了”。思路类似 TCP 的 ACK但默认不开启需要手动调用channel.confirmSelect()。注意区分两个确认机制确认方含义Publisher ConfirmBroker → 生产者消息已被 Broker接收入队或路由成功Consumer Ack消费者 → Broker消息已被消费者处理完成发布确认不能保证消费者一定收到或处理成功那是消费者 ACK 的职责。在资金流水、库存同步等对可靠性要求高的场景发布确认是生产端的标配。二、开启确认模式无论采用哪种确认策略第一步都一样channel.confirmSelect();// 将 Channel 设为 confirm 模式此后每次basicPublishBroker 都会返回ack成功或nack失败。三、三种确认策略策略方式适用场景吞吐量单条同步发一条 →waitForConfirms()消息极少、要求逐条可控最低批量同步发 N 条 →waitForConfirmsOrDie()大批量、可容忍批量失败中等异步回调ConfirmListener边发边收高吞吐、需细粒度重发最高下面分别给出完整示例。连接参数统一使用常量类管理推荐做法// constants/RabbitConstant.javapublicfinalclassRabbitConstant{publicstaticfinalStringHOST你的服务器IP;publicstaticfinalintPORT5672;publicstaticfinalStringUSERNAMEstudy;publicstaticfinalStringPASSWORDstudy;publicstaticfinalStringVIRTUAL_HOSTtest-virtual;publicstaticfinalStringQUEUE_HELLOhello;}Maven 依赖dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.25.0/version/dependency3.1 单条同步确认每发一条就阻塞等待 Broker 回复最简单也最慢channel.confirmSelect();for(inti0;i10;i){Stringbody消息_i;channel.basicPublish(,RabbitConstant.QUEUE_HELLO,null,body.getBytes());// 等待当前这条消息的确认超时 5 秒抛 IOExceptionchannel.waitForConfirmsOrDie(5_000);System.out.println(消息 i 确认成功);}waitForConfirmsOrDie()的行为全部 ack → 正常返回收到 nack → 抛IOException: n messages not confirmed超时未收到回复 → 抛IOException: timed out waiting for confirms3.2 批量同步确认推荐入门攒够一批再统一等待大幅减少网络往返ConnectionFactoryfactorynewConnectionFactory();factory.setHost(RabbitConstant.HOST);factory.setPort(RabbitConstant.PORT);factory.setUsername(RabbitConstant.USERNAME);factory.setPassword(RabbitConstant.PASSWORD);factory.setVirtualHost(RabbitConstant.VIRTUAL_HOST);Connectionconnectionfactory.newConnection();Channelchannelconnection.createChannel();channel.queueDeclare(RabbitConstant.QUEUE_HELLO,true,false,false,null);channel.confirmSelect();// ← 原文遗漏了这一步不调用则不会收到确认intbatchSize100;// 每批条数intoutstandingCount0;// 当前批次已发未确认条数for(inti0;i560;i){Stringbody批量消息_i;channel.basicPublish(,RabbitConstant.QUEUE_HELLO,null,body.getBytes());outstandingCount;if(outstandingCountbatchSize){channel.waitForConfirmsOrDie(5_000);// 等待本批全部确认System.out.println(【批量确认】batchSize 条消息确认成功);outstandingCount0;}}// 处理最后不足一批的剩余消息if(outstandingCount0){channel.waitForConfirmsOrDie(5_000);System.out.println(【批量确认】剩余 outstandingCount 条消息确认成功);}channel.close();connection.close();multiple参数的含义同步模式下体现在 Broker 侧Broker 可以对一批消息回复一条 ackdeliveryTag表示此序号及之前的消息全部确认。同步的waitForConfirms()会在内部处理这些批量 ack。3.3 异步回调确认生产推荐发送与确认解耦吞吐量最高。核心思路发送前用getNextPublishSeqNo()取序号用SortedSet跟踪未确认序号用MapLong, byte[]保存消息体供 nack 时重发注册ConfirmListener处理 ack / nackchannel.confirmSelect();// 未确认序号集合线程安全SortedSetLongunconfirmedSetCollections.synchronizedSortedSet(newTreeSet());// 序号 → 消息体用于 nack 重发MapLong,byte[]messageStorenewConcurrentHashMap();channel.addConfirmListener(newConfirmListener(){OverridepublicvoidhandleAck(longdeliveryTag,booleanmultiple){System.out.println(【ACK】序号deliveryTag, 批量multiple);if(multiple){// 批量确认deliveryTag 及之前的所有序号均已确认unconfirmedSet.headSet(deliveryTag1).clear();messageStore.keySet().removeIf(tag-tagdeliveryTag);}else{unconfirmedSet.remove(deliveryTag);messageStore.remove(deliveryTag);}}OverridepublicvoidhandleNack(longdeliveryTag,booleanmultiple){System.out.println(【NACK】序号deliveryTag, 批量multiple);// 收集需要重发的序号SortedSetLongfailedTagsmultiple?newTreeSet(unconfirmedSet.headSet(deliveryTag1)):newTreeSet(Collections.singleton(deliveryTag));// 从跟踪集合中移除if(multiple){unconfirmedSet.headSet(deliveryTag1).clear();}else{unconfirmedSet.remove(deliveryTag);}// 重发失败消息for(Longtag:failedTags){byte[]bodymessageStore.remove(tag);if(body!null){longnewSeqNochannel.getNextPublishSeqNo();// 先取序号channel.basicPublish(,RabbitConstant.QUEUE_HELLO,null,body);unconfirmedSet.add(newSeqNo);messageStore.put(newSeqNo,body);System.out.println(【重发】原序号tag → 新序号newSeqNo);}}}});// 发送消息for(inti0;i10;i){byte[]bodyBytes(消息_i).getBytes();longseqNochannel.getNextPublishSeqNo();// ← 必须在 basicPublish 之前调用channel.basicPublish(,RabbitConstant.QUEUE_HELLO,null,bodyBytes);unconfirmedSet.add(seqNo);messageStore.put(seqNo,bodyBytes);}// 等待所有消息确认完毕while(!unconfirmedSet.isEmpty()){Thread.sleep(10);}System.out.println(所有消息确认完毕);四、关键细节与常见坑4.1deliveryTag在 Confirm 模式下是发布序号在ConfirmListener回调里deliveryTag不是消费者那边的投递标签而是publish sequence number从 1 开始递增。不要和消费者Envelope.getDeliveryTag()混用。4.2 序号必须在发送前获取// ✅ 正确longseqNochannel.getNextPublishSeqNo();channel.basicPublish(...);// ❌ 错误先发送再取序号序号会错位channel.basicPublish(...);longseqNochannel.getNextPublishSeqNo();4.3multipletrue时的清理逻辑Broker 回复: ack(deliveryTag5, multipletrue) 含义: 序号 1、2、3、4、5 全部确认成功 清理方式: unconfirmedSet.headSet(5 1).clear() 即删除所有 6 的序号4.4 nack 的常见原因原因说明交换机不存在消息无法路由队列达到x-max-length队列满消息被拒绝集群脑裂后恢复极少数情况下历史消息被 nacknack不会抛异常必须在handleNack里自行实现重发或告警。4.5 同步 vs 异步的选择消息量推荐模式 100/秒单条同步代码最简单100~10000/秒批量同步batchSize 建议 100~500 10000/秒异步回调配合消息存储 重发五、三种模式对比总结模式核心 API失败处理吞吐量单条同步waitForConfirmsOrDie抛 IOException★☆☆☆☆批量同步waitForConfirmsOrDie抛 IOException★★★☆☆异步回调ConfirmListenerhandleNack 重发★★★★★六、延伸阅读Mandatory 标志 ReturnListener消息无法路由到任何队列时Broker 会将消息退回生产者与 Confirm 互补事务模式tx也能保证可靠性但性能约为 Confirm 的 1/100生产环境不推荐Spring AMQPRabbitTemplate封装了 Confirm 回调配置publisher-confirm-type: correlated即可使用