1. 从一次线上事故说起为什么发送确认这件事不能省去年帮一个朋友排查过一个挺典型的问题。他们的订单系统用 Spring Boot 整合了 RabbitMQ 做异步解耦下单成功后往队列里丢一条消息让下游的库存、积分、通知服务各自消费。上线头两个月一切正常直到某天大促运营反馈有一批订单的积分没到账但订单本身是成功的。查日志发现那批订单的消息压根就没进到 Broker 里——生产端调用convertAndSend之后代码继续往下走没有任何异常抛出但消息就是丢了。原因其实不复杂他们的交换机名字在配置文件里写错了一个字母消息发到了一个不存在的交换机上。RabbitMQ 的默认行为是静默丢弃——交换机找不到就扔掉不报错、不返回、不阻塞。生产端以为发成功了实际上消息连 Broker 的门都没进。这就是为什么消息发送、失败回调、发送确认这三件事必须一起做。很多人整合 RabbitMQ 的时候只关注能不能发出去、能不能收到跑通 Demo 就收工了。但生产环境和 Demo 最大的区别在于Demo 里消息丢了你看得见生产环境里消息丢了你看不见。等到业务方找上门往往已经过去几个小时数据对不上补数据又是一堆麻烦。这篇内容我打算把 Spring Boot 整合 RabbitMQ 的发送链路彻底讲透重点放在三个层面消息怎么发出去发送、发失败了怎么知道失败回调、发成功了怎么确认发送确认。适合已经跑通过基础 Demo、想把生产端做扎实的开发者。如果你还在纠结Spring Boot 怎么连 RabbitMQ建议先把基础配置跑通再来看这篇因为这里讲的都是基础之上那些真正会坑人的细节。2. 发送链路的三道关卡Confirm、Return、Callback 到底谁管谁在动手写代码之前必须先把 RabbitMQ 生产端的可靠性机制理清楚。很多人把发送确认和失败回调混为一谈结果配置了一堆监听器出问题的时候发现该响的没响。我画不了图但可以用一个快递的类比把这三道关卡讲明白。2.1 Confirm 确认机制Broker 告诉你我收到了Confirm发布确认解决的是生产端到交换机这一段的问题。当你把publisher-confirm-type打开之后每条消息都会带一个唯一的投递标签delivery tagBroker 收到消息后会异步回调告诉你结果ack消息成功到达交换机Broker 认账了。nackBroker 因为内部原因比如队列满了、磁盘告警没能接收。注意Confirm 只保证消息到了交换机它不保证消息到了队列。这是最容易误解的一点。交换机收到消息之后如果路由不到任何队列Confirm 依然会返回 ack——因为在 Broker 看来它的活儿已经干完了。2.2 Return 退回机制路由不到队列时把消息退回来Return解决的是交换机到队列这一段的问题。当消息到达交换机但根据 routing key 找不到匹配的队列时如果开启了mandatory参数Broker 会把消息退回给生产端触发ReturnsCallback。这里有个关键前提mandatory必须设为 true。默认是 false也就是路由失败直接静默丢弃连退回都不退。我见过太多人只配了 Confirm 没配 mandatory结果交换机写对了、routing key 写错了消息照样丢Confirm 还一直返回 ack查起来一脸懵。2.3 三者的职责边界对照把这三者的关系用表格理一遍比看十遍文档都清楚机制覆盖链路触发条件关键配置典型用途Confirm生产端 → 交换机消息到达或未到达交换机publisher-confirm-type: correlated判断 Broker 是否收到Return交换机 → 队列消息路由不到任何队列mandatory: trueReturnsCallback发现 routing key 配错本地回调生产端内部发送过程抛异常try-catch 或setConfirmCallback处理网络中断等异常注意Confirm 和 Return 是两条独立的回调链路不是二选一。生产端要真正可靠两个都得配。只配 Confirm路由失败你发现不了只配 ReturnBroker 没收到你也发现不了。2.4 为什么异步确认比同步确认更值得用RabbitMQ 支持同步确认waitForConfirms和异步确认回调。同步确认的写法是发一条等一条Broker 返回了才发下一条。这种方式逻辑简单但吞吐量会断崖式下跌——每条消息都要等一个网络往返QPS 直接砍到个位数。异步确认则是发出去就继续Broker 通过回调异步通知结果。生产端可以用一个ConcurrentHashMap把未确认的消息暂存起来收到 ack 就移除收到 nack 就重发或落库。这样吞吐量能保持在正常水平代价是代码复杂度上升。生产环境我基本都用异步除非是那种一天发不了几条消息的后台任务。3. 环境搭建依赖、配置项与那些容易写错的参数这一节把整合的骨架搭起来。我用的版本组合是 Spring Boot 3.2.x Spring AMQP 3.2.x RabbitMQ 3.12这套组合目前比较稳。如果你还在 Spring Boot 2.x配置项的写法略有差异我会在关键处标注。3.1 依赖引入与版本对齐pom.xml里核心就一个依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency看起来简单但有个坑spring-boot-starter-amqp会传递引入spring-rabbit和amqp-client。如果你项目里其他地方也引了amqp-client版本可能冲突。我遇到过一次项目里手动引了 5.9 的amqp-client而 starter 带的是 5.20结果 Confirm 回调死活不触发。排查了半天才发现是版本打架。所以引依赖之后用mvn dependency:tree | grep amqp确认一下版本别嫌麻烦。3.2 application.yml 里的关键配置配置项是重灾区我直接把一份生产可用的配置贴出来然后逐条解释spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest virtual-host: / publisher-confirm-type: correlated publisher-returns: true template: mandatory: true listener: simple: acknowledge-mode: manual prefetch: 1逐条说publisher-confirm-type有三个值。none关闭simple开启但用同步方式correlated开启异步回调。生产环境必须用 correlated否则setConfirmCallback不生效。publisher-returns开启退回机制配合mandatory使用。template.mandatory这个必须显式设为 true。很多人只配了publisher-returns: true就以为完事了其实mandatory默认是 false退回机制根本不触发。acknowledge-mode消费端的确认模式这里设 manual 是为了后面讲消费确认时用生产端发送不依赖它。prefetch每次预取的消息数设 1 是为了公平分发避免一个消费者堆积。提示Spring Boot 2.x 里publisher-confirm-type的写法是publisher-confirms: true布尔值3.x 改成了枚举。升级的时候这里最容易漏改配置不生效还不报错。3.3 RabbitTemplate 的定制化配置光靠 yml 还不够RabbitTemplate需要手动注入回调。我一般写一个配置类Configuration public class RabbitMQConfig { Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template new RabbitTemplate(connectionFactory); template.setMandatory(true); template.setConfirmCallback((correlationData, ack, cause) - { if (ack) { log.info(消息到达交换机, id{}, correlationData ! null ? correlationData.getId() : null); } else { log.error(消息未到达交换机, id{}, cause{}, correlationData ! null ? correlationData.getId() : null, cause); } }); template.setReturnsCallback(returned - { log.error(消息路由失败, exchange{}, routingKey{}, replyText{}, message{}, returned.getExchange(), returned.getRoutingKey(), returned.getReplyText(), new String(returned.getMessage().getBody())); }); return template; } }这里有个细节setConfirmCallback里的correlationData可能为 null。如果你发送的时候没传CorrelationData回调里就拿不到消息标识只能知道有一条消息 ack 了但不知道是哪条。所以发送时必须传 CorrelationData这是后面做重发的关键。4. 消息发送的三种姿势与 CorrelationData 的正确用法发送消息看起来就是一行convertAndSend但里面门道不少。这一节把发送方式、CorrelationData 的构造、以及发送时的异常处理讲清楚。4.1 三种发送方式的适用场景Spring AMQP 提供了几种发送方法我按使用频率排一下convertAndSend(exchange, routingKey, object)最常用自动把对象序列化成消息体。适合大多数场景。convertAndSend(exchange, routingKey, object, correlationData)带确认标识的版本做可靠投递必须用这个。convertAndSend(exchange, routingKey, object, messagePostProcessor)需要在发送前修改消息属性比如设置过期时间、自定义 header时用。我个人的习惯是只要这个业务消息不能丢一律用带CorrelationData的版本。多写一个参数换来的是出问题时能精确定位到是哪条消息。4.2 CorrelationData 里到底该放什么CorrelationData是 Spring AMQP 提供的一个包装类核心就一个id字段和一个Future。很多人随手new CorrelationData(UUID.randomUUID().toString())就完事了但这样在重发场景下会出问题——你不知道这条消息对应的是哪条业务数据。我的做法是把业务主键塞进去CorrelationData correlationData new CorrelationData(orderId); rabbitTemplate.convertAndSend(order.exchange, order.create, orderDTO, correlationData);这样在 Confirm 回调里correlationData.getId()拿到的就是订单号。收到 nack 的时候直接拿订单号去查库、重发或者标记异常链路是通的。CorrelationData还有个getFuture()方法返回一个CompletableFutureConfirm。如果你需要同步等待确认结果比如某些强一致场景可以future.get(3, TimeUnit.SECONDS)。但我不建议在主流程里这么干会阻塞业务线程一般只在补偿任务里用。4.3 发送时的异常处理不能只靠回调Confirm 和 Return 都是异步回调它们处理的是消息发出去了但结果不理想的情况。但还有一种情况是消息根本没发出去——比如连接断了、序列化失败、Broker 拒绝连接。这时候回调不会触发异常会直接抛在convertAndSend这一行。所以发送代码必须包一层 try-catchtry { rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData); } catch (AmqpException e) { log.error(消息发送异常, orderId{}, orderId, e); // 落库、重试或告警 }我见过有人只配了 Confirm 回调发送代码裸奔结果网络抖动的时候异常直接冒到 Controller用户看到 500但消息其实没发出去也没记录事后完全没法追溯。4.4 一个完整的发送封装示例把上面的东西串起来写一个可复用的发送工具方法public void sendReliable(String exchange, String routingKey, Object payload, String bizId) { CorrelationData correlationData new CorrelationData(bizId); try { rabbitTemplate.convertAndSend(exchange, routingKey, payload, correlationData); } catch (AmqpException e) { log.error(发送异常, bizId{}, bizId, e); saveToRetryTable(bizId, payload, SEND_EXCEPTION); } }这个方法把发送异常这一层兜住了剩下的到达交换机但路由失败和未到达交换机交给回调处理。三层防护叠起来消息丢失的概率能压到极低。5. 失败回调的实战从 nack 到路由失败怎么定位和补救回调配好了只是第一步真正难的是收到失败回调之后怎么办。这一节讲两类失败的处理思路以及我踩过的几个坑。5.1 Confirm 回调收到 nack 时的处理策略nack 意味着 Broker 明确告诉你我没收到。常见原因有队列满了触发流控、磁盘空间不足、内存告警。这时候消息是确定丢了的必须补救。我的处理策略分三档立即重发一次如果是偶发的流控隔几百毫秒重发大概率能成功。但要注意别在回调线程里直接重发会阻塞回调。我一般丢到一个延迟队列或者用ScheduledExecutorService异步重发。落库待补偿重发一次还失败就把消息内容、业务 ID、失败原因写进一张mq_send_fail表由定时任务扫描重试。告警连续 nack 超过阈值直接触发告警这通常意味着 Broker 出了大问题不是重发能解决的。这里有个坑回调线程里不要做耗时操作。Confirm 回调是在amqp-client的 IO 线程里执行的你在里面查库、发 HTTP 请求会拖慢整个连接的消息处理。我见过有人在回调里同步调用下游接口结果把 IO 线程堵死所有消息的确认都延迟了。正确做法是丢到业务线程池里异步处理。5.2 Return 回调routing key 写错的典型排查Return 回调触发基本可以断定是路由配置问题。我遇到过的原因按频率排routing key 拼写错误比如order.create写成了order.creat。队列没绑定到交换机或者绑定的 routing key 不匹配。交换机类型用错比如 topic 交换机用了精确匹配的 key。排查的时候ReturnsCallback里的replyText会给出原因通常是NO_ROUTE。把exchange、routingKey、message打出来去 RabbitMQ 管理后台对一下绑定关系基本一眼就能看出来。注意Return 回调触发时消息已经被退回了不会进任何队列。所以 Return 的处理逻辑和 nack 类似——落库、重发、告警。但重发之前一定要先修 routing key否则重发多少次都是白搭。5.3 一个真实的排查链路复盘回到开头那个积分没到账的案例完整排查过程是这样的业务方反馈积分缺失先查订单库确认订单状态正常。查积分服务的消费日志发现那批订单的消息压根没被消费。查 RabbitMQ 管理后台队列里没有堆积说明消息没进队列。查生产端日志发现 Confirm 回调全是 ack没有 nack。这时候意识到问题在路由环节检查配置发现交换机名字写错了。但为什么 Return 没触发因为mandatory没设 true路由失败被静默丢弃了。整个链路卡在第 6 步。如果当初配了mandatory: true和 Return 回调第 3 步就能发现异常不用等到业务方找上门。这个案例我后来在团队里讲了好几次核心教训就是Confirm 的 ack 不代表消息安全它只代表 Broker 收到了不代表消息进了队列。5.4 回调里的日志要打哪些字段日志打得好排查省一半力气。我的 Confirm 和 Return 回调里固定打这几个字段回调类型必打字段作用ConfirmcorrelationData.id、ack、cause定位是哪条消息、成功还是失败、失败原因Returnexchange、routingKey、replyText、message body定位路由配置、失败原因、消息内容cause字段特别重要nack 的时候它会给出具体原因比如CHANNEL_ERROR、ACCESS_REFUSED。没有这个字段你只知道失败了不知道为什么失败。6. 消费端确认与生产端确认的配合别让消息在两头都丢了生产端做得再扎实消费端如果配错消息照样丢。这一节把两端的确认机制串起来讲因为它们是配套的。6.1 手动 ack 与自动 ack 的取舍Spring AMQP 的消费端确认模式有三种none自动确认消息一投递给消费者就认为消费成功。生产环境千万别用消费者处理到一半挂了消息就丢了。auto根据方法是否抛异常决定 ack 还是 nack。比 none 好但有个坑——如果方法执行成功但业务逻辑其实失败了比如返回了错误码但没抛异常消息照样被 ack 掉。manual手动调用basicAck或basicNack。最可靠但代码要自己写。我基本都用 manual配合 try-catch 精确控制。消费成功的业务逻辑走完再 ack出异常就 nack 并决定是否重新入队。6.2 消费端 nack 时 requeue 的陷阱basicNack有个requeue参数。设 true 消息会重新入队设 false 消息会被丢弃或进死信队列。这里有个经典陷阱如果消息本身有问题比如格式错误导致反序列化失败requeue 设 true 会导致消息无限循环——消费失败、重新入队、再消费失败、再入队CPU 直接跑满。我踩过一次一个格式错误的消息把整个消费者卡死了。正确做法是业务异常可以 requeue但要有重试次数上限格式异常、反序列化异常一律 requeuefalse直接进死信队列。判断依据是异常类型MessageConversionException这类直接判死BusinessException这类可以重试。6.3 生产端和消费端的确认如何形成闭环把两端串起来看一条消息的完整生命周期是这样的生产端发送Confirm 确认到达交换机。Return 确认路由到队列没触发 Return 就是路由成功。消息进队列等待消费。消费者拉取消息处理业务。处理成功手动 ack消息从队列删除。处理失败nack根据 requeue 决定重新入队还是进死信。任何一环出问题消息都可能丢。生产端的 Confirm 和 Return 保证消息进队列消费端的 manual ack 保证消息被正确处理。两头都做扎实整个链路才可靠。提示死信队列不是可选项是必选项。所有可能失败的消息最终都要有个去处否则要么丢要么无限循环。配一个死信交换机把重试耗尽的消息丢进去人工介入处理。7. 几个我踩过的坑和对应的规避手法这一节把散落在各处的经验集中一下都是实际项目里踩出来的文档上不一定写。7.1 序列化方式不一致导致的消息乱码生产端用Jackson2JsonMessageConverter消费端用默认的SimpleMessageConverter结果消费端拿到的是字节数组反序列化直接报错。这个坑的根源是两端的 MessageConverter 必须一致。我的做法是在配置类里统一定义一个Jackson2JsonMessageConverterBean生产端和消费端都注入它。如果跨语言比如生产端 Java、消费端 Python那就约定好 JSON 格式别用 Java 特有的序列化。7.2 回调不触发的几种情况Confirm 回调不触发排查下来通常是这几个原因publisher-confirm-type没设成correlated设成了simple或none。RabbitTemplate是手动 new 的没走 Spring 容器回调没注入。依赖版本冲突amqp-client版本过低不支持异步确认。Return 回调不触发基本就一个原因mandatory没设 true。这个我强调过好几次了因为它真的太容易漏。7.3 大批量发送时的内存问题做数据同步的时候一次性往队列里灌几十万条消息如果每条都new CorrelationData并存在一个 Map 里等确认内存直接爆掉。CorrelationData本身不大但几十万个累积起来也可观。我的处理方式是分批发送 批量确认。每批 1000 条发完等这批的确认结果用CompletableFuture.allOf聚合确认完再发下一批。这样内存占用可控吞吐量也不差。7.4 关于重试次数的经验值重试不是越多越好。我的经验是同步重试最多 1 次异步补偿最多 3 次。同步重试多了会阻塞业务异步补偿超过 3 次基本说明不是偶发问题再重试也是浪费资源不如直接告警让人来看。重试间隔用指数退避第一次 1 秒第二次 5 秒第三次 30 秒。别用固定间隔否则 Broker 压力大的时候重试会雪上加霜。8. 写在最后可靠投递是个系统工程把 Spring Boot 整合 RabbitMQ 的发送链路做可靠不是配几个参数就完事的。Confirm、Return、异常捕获、CorrelationData、消费端 manual ack、死信队列这些是一整套配合起来的东西缺一环整个链路就有漏洞。我个人的体会是先想清楚消息丢了会怎样再决定投入多少精力做可靠性。像日志采集这种丢几条无所谓的场景配个基础发送就行但订单、支付、积分这种涉及钱的Confirm Return 落库补偿 死信队列一个都不能少。技术方案永远是为业务服务的别为了看起来专业过度设计也别为了省事在关键链路上偷懒。最后分享一个我常用的自检清单上线前对着过一遍能挡掉大部分低级问题publisher-confirm-type是不是correlatedpublisher-returns和template.mandatory是不是都开了发送时有没有传CorrelationData里面是不是业务主键Confirm 和 Return 回调里有没有打全日志字段发送代码有没有 try-catch 兜底消费端是不是 manual acknack 的 requeue 逻辑有没有防死循环死信队列配了没有这七条过完生产端的可靠性基本就有底了。剩下的就是根据业务量调优参数那是另一个话题了。 
长河君 · 2026/10/13 11:13:56 · 阅读 1.2k Spring Boot整合RabbitMQ消息发送确认与失败回调实战
1. 从一次线上事故说起为什么发送确认这件事不能省去年帮一个朋友排查过一个挺典型的问题。他们的订单系统用 Spring Boot 整合了 RabbitMQ 做异步解耦下单成功后往队列里丢一条消息让下游的库存、积分、通知服务各自消费。上线头两个月一切正常直到某天大促运营反馈有一批订单的积分没到账但订单本身是成功的。查日志发现那批订单的消息压根就没进到 Broker 里——生产端调用convertAndSend之后代码继续往下走没有任何异常抛出但消息就是丢了。原因其实不复杂他们的交换机名字在配置文件里写错了一个字母消息发到了一个不存在的交换机上。RabbitMQ 的默认行为是静默丢弃——交换机找不到就扔掉不报错、不返回、不阻塞。生产端以为发成功了实际上消息连 Broker 的门都没进。这就是为什么消息发送、失败回调、发送确认这三件事必须一起做。很多人整合 RabbitMQ 的时候只关注能不能发出去、能不能收到跑通 Demo 就收工了。但生产环境和 Demo 最大的区别在于Demo 里消息丢了你看得见生产环境里消息丢了你看不见。等到业务方找上门往往已经过去几个小时数据对不上补数据又是一堆麻烦。这篇内容我打算把 Spring Boot 整合 RabbitMQ 的发送链路彻底讲透重点放在三个层面消息怎么发出去发送、发失败了怎么知道失败回调、发成功了怎么确认发送确认。适合已经跑通过基础 Demo、想把生产端做扎实的开发者。如果你还在纠结Spring Boot 怎么连 RabbitMQ建议先把基础配置跑通再来看这篇因为这里讲的都是基础之上那些真正会坑人的细节。2. 发送链路的三道关卡Confirm、Return、Callback 到底谁管谁在动手写代码之前必须先把 RabbitMQ 生产端的可靠性机制理清楚。很多人把发送确认和失败回调混为一谈结果配置了一堆监听器出问题的时候发现该响的没响。我画不了图但可以用一个快递的类比把这三道关卡讲明白。2.1 Confirm 确认机制Broker 告诉你我收到了Confirm发布确认解决的是生产端到交换机这一段的问题。当你把publisher-confirm-type打开之后每条消息都会带一个唯一的投递标签delivery tagBroker 收到消息后会异步回调告诉你结果ack消息成功到达交换机Broker 认账了。nackBroker 因为内部原因比如队列满了、磁盘告警没能接收。注意Confirm 只保证消息到了交换机它不保证消息到了队列。这是最容易误解的一点。交换机收到消息之后如果路由不到任何队列Confirm 依然会返回 ack——因为在 Broker 看来它的活儿已经干完了。2.2 Return 退回机制路由不到队列时把消息退回来Return解决的是交换机到队列这一段的问题。当消息到达交换机但根据 routing key 找不到匹配的队列时如果开启了mandatory参数Broker 会把消息退回给生产端触发ReturnsCallback。这里有个关键前提mandatory必须设为 true。默认是 false也就是路由失败直接静默丢弃连退回都不退。我见过太多人只配了 Confirm 没配 mandatory结果交换机写对了、routing key 写错了消息照样丢Confirm 还一直返回 ack查起来一脸懵。2.3 三者的职责边界对照把这三者的关系用表格理一遍比看十遍文档都清楚机制覆盖链路触发条件关键配置典型用途Confirm生产端 → 交换机消息到达或未到达交换机publisher-confirm-type: correlated判断 Broker 是否收到Return交换机 → 队列消息路由不到任何队列mandatory: trueReturnsCallback发现 routing key 配错本地回调生产端内部发送过程抛异常try-catch 或setConfirmCallback处理网络中断等异常注意Confirm 和 Return 是两条独立的回调链路不是二选一。生产端要真正可靠两个都得配。只配 Confirm路由失败你发现不了只配 ReturnBroker 没收到你也发现不了。2.4 为什么异步确认比同步确认更值得用RabbitMQ 支持同步确认waitForConfirms和异步确认回调。同步确认的写法是发一条等一条Broker 返回了才发下一条。这种方式逻辑简单但吞吐量会断崖式下跌——每条消息都要等一个网络往返QPS 直接砍到个位数。异步确认则是发出去就继续Broker 通过回调异步通知结果。生产端可以用一个ConcurrentHashMap把未确认的消息暂存起来收到 ack 就移除收到 nack 就重发或落库。这样吞吐量能保持在正常水平代价是代码复杂度上升。生产环境我基本都用异步除非是那种一天发不了几条消息的后台任务。3. 环境搭建依赖、配置项与那些容易写错的参数这一节把整合的骨架搭起来。我用的版本组合是 Spring Boot 3.2.x Spring AMQP 3.2.x RabbitMQ 3.12这套组合目前比较稳。如果你还在 Spring Boot 2.x配置项的写法略有差异我会在关键处标注。3.1 依赖引入与版本对齐pom.xml里核心就一个依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency看起来简单但有个坑spring-boot-starter-amqp会传递引入spring-rabbit和amqp-client。如果你项目里其他地方也引了amqp-client版本可能冲突。我遇到过一次项目里手动引了 5.9 的amqp-client而 starter 带的是 5.20结果 Confirm 回调死活不触发。排查了半天才发现是版本打架。所以引依赖之后用mvn dependency:tree | grep amqp确认一下版本别嫌麻烦。3.2 application.yml 里的关键配置配置项是重灾区我直接把一份生产可用的配置贴出来然后逐条解释spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest virtual-host: / publisher-confirm-type: correlated publisher-returns: true template: mandatory: true listener: simple: acknowledge-mode: manual prefetch: 1逐条说publisher-confirm-type有三个值。none关闭simple开启但用同步方式correlated开启异步回调。生产环境必须用 correlated否则setConfirmCallback不生效。publisher-returns开启退回机制配合mandatory使用。template.mandatory这个必须显式设为 true。很多人只配了publisher-returns: true就以为完事了其实mandatory默认是 false退回机制根本不触发。acknowledge-mode消费端的确认模式这里设 manual 是为了后面讲消费确认时用生产端发送不依赖它。prefetch每次预取的消息数设 1 是为了公平分发避免一个消费者堆积。提示Spring Boot 2.x 里publisher-confirm-type的写法是publisher-confirms: true布尔值3.x 改成了枚举。升级的时候这里最容易漏改配置不生效还不报错。3.3 RabbitTemplate 的定制化配置光靠 yml 还不够RabbitTemplate需要手动注入回调。我一般写一个配置类Configuration public class RabbitMQConfig { Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template new RabbitTemplate(connectionFactory); template.setMandatory(true); template.setConfirmCallback((correlationData, ack, cause) - { if (ack) { log.info(消息到达交换机, id{}, correlationData ! null ? correlationData.getId() : null); } else { log.error(消息未到达交换机, id{}, cause{}, correlationData ! null ? correlationData.getId() : null, cause); } }); template.setReturnsCallback(returned - { log.error(消息路由失败, exchange{}, routingKey{}, replyText{}, message{}, returned.getExchange(), returned.getRoutingKey(), returned.getReplyText(), new String(returned.getMessage().getBody())); }); return template; } }这里有个细节setConfirmCallback里的correlationData可能为 null。如果你发送的时候没传CorrelationData回调里就拿不到消息标识只能知道有一条消息 ack 了但不知道是哪条。所以发送时必须传 CorrelationData这是后面做重发的关键。4. 消息发送的三种姿势与 CorrelationData 的正确用法发送消息看起来就是一行convertAndSend但里面门道不少。这一节把发送方式、CorrelationData 的构造、以及发送时的异常处理讲清楚。4.1 三种发送方式的适用场景Spring AMQP 提供了几种发送方法我按使用频率排一下convertAndSend(exchange, routingKey, object)最常用自动把对象序列化成消息体。适合大多数场景。convertAndSend(exchange, routingKey, object, correlationData)带确认标识的版本做可靠投递必须用这个。convertAndSend(exchange, routingKey, object, messagePostProcessor)需要在发送前修改消息属性比如设置过期时间、自定义 header时用。我个人的习惯是只要这个业务消息不能丢一律用带CorrelationData的版本。多写一个参数换来的是出问题时能精确定位到是哪条消息。4.2 CorrelationData 里到底该放什么CorrelationData是 Spring AMQP 提供的一个包装类核心就一个id字段和一个Future。很多人随手new CorrelationData(UUID.randomUUID().toString())就完事了但这样在重发场景下会出问题——你不知道这条消息对应的是哪条业务数据。我的做法是把业务主键塞进去CorrelationData correlationData new CorrelationData(orderId); rabbitTemplate.convertAndSend(order.exchange, order.create, orderDTO, correlationData);这样在 Confirm 回调里correlationData.getId()拿到的就是订单号。收到 nack 的时候直接拿订单号去查库、重发或者标记异常链路是通的。CorrelationData还有个getFuture()方法返回一个CompletableFutureConfirm。如果你需要同步等待确认结果比如某些强一致场景可以future.get(3, TimeUnit.SECONDS)。但我不建议在主流程里这么干会阻塞业务线程一般只在补偿任务里用。4.3 发送时的异常处理不能只靠回调Confirm 和 Return 都是异步回调它们处理的是消息发出去了但结果不理想的情况。但还有一种情况是消息根本没发出去——比如连接断了、序列化失败、Broker 拒绝连接。这时候回调不会触发异常会直接抛在convertAndSend这一行。所以发送代码必须包一层 try-catchtry { rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData); } catch (AmqpException e) { log.error(消息发送异常, orderId{}, orderId, e); // 落库、重试或告警 }我见过有人只配了 Confirm 回调发送代码裸奔结果网络抖动的时候异常直接冒到 Controller用户看到 500但消息其实没发出去也没记录事后完全没法追溯。4.4 一个完整的发送封装示例把上面的东西串起来写一个可复用的发送工具方法public void sendReliable(String exchange, String routingKey, Object payload, String bizId) { CorrelationData correlationData new CorrelationData(bizId); try { rabbitTemplate.convertAndSend(exchange, routingKey, payload, correlationData); } catch (AmqpException e) { log.error(发送异常, bizId{}, bizId, e); saveToRetryTable(bizId, payload, SEND_EXCEPTION); } }这个方法把发送异常这一层兜住了剩下的到达交换机但路由失败和未到达交换机交给回调处理。三层防护叠起来消息丢失的概率能压到极低。5. 失败回调的实战从 nack 到路由失败怎么定位和补救回调配好了只是第一步真正难的是收到失败回调之后怎么办。这一节讲两类失败的处理思路以及我踩过的几个坑。5.1 Confirm 回调收到 nack 时的处理策略nack 意味着 Broker 明确告诉你我没收到。常见原因有队列满了触发流控、磁盘空间不足、内存告警。这时候消息是确定丢了的必须补救。我的处理策略分三档立即重发一次如果是偶发的流控隔几百毫秒重发大概率能成功。但要注意别在回调线程里直接重发会阻塞回调。我一般丢到一个延迟队列或者用ScheduledExecutorService异步重发。落库待补偿重发一次还失败就把消息内容、业务 ID、失败原因写进一张mq_send_fail表由定时任务扫描重试。告警连续 nack 超过阈值直接触发告警这通常意味着 Broker 出了大问题不是重发能解决的。这里有个坑回调线程里不要做耗时操作。Confirm 回调是在amqp-client的 IO 线程里执行的你在里面查库、发 HTTP 请求会拖慢整个连接的消息处理。我见过有人在回调里同步调用下游接口结果把 IO 线程堵死所有消息的确认都延迟了。正确做法是丢到业务线程池里异步处理。5.2 Return 回调routing key 写错的典型排查Return 回调触发基本可以断定是路由配置问题。我遇到过的原因按频率排routing key 拼写错误比如order.create写成了order.creat。队列没绑定到交换机或者绑定的 routing key 不匹配。交换机类型用错比如 topic 交换机用了精确匹配的 key。排查的时候ReturnsCallback里的replyText会给出原因通常是NO_ROUTE。把exchange、routingKey、message打出来去 RabbitMQ 管理后台对一下绑定关系基本一眼就能看出来。注意Return 回调触发时消息已经被退回了不会进任何队列。所以 Return 的处理逻辑和 nack 类似——落库、重发、告警。但重发之前一定要先修 routing key否则重发多少次都是白搭。5.3 一个真实的排查链路复盘回到开头那个积分没到账的案例完整排查过程是这样的业务方反馈积分缺失先查订单库确认订单状态正常。查积分服务的消费日志发现那批订单的消息压根没被消费。查 RabbitMQ 管理后台队列里没有堆积说明消息没进队列。查生产端日志发现 Confirm 回调全是 ack没有 nack。这时候意识到问题在路由环节检查配置发现交换机名字写错了。但为什么 Return 没触发因为mandatory没设 true路由失败被静默丢弃了。整个链路卡在第 6 步。如果当初配了mandatory: true和 Return 回调第 3 步就能发现异常不用等到业务方找上门。这个案例我后来在团队里讲了好几次核心教训就是Confirm 的 ack 不代表消息安全它只代表 Broker 收到了不代表消息进了队列。5.4 回调里的日志要打哪些字段日志打得好排查省一半力气。我的 Confirm 和 Return 回调里固定打这几个字段回调类型必打字段作用ConfirmcorrelationData.id、ack、cause定位是哪条消息、成功还是失败、失败原因Returnexchange、routingKey、replyText、message body定位路由配置、失败原因、消息内容cause字段特别重要nack 的时候它会给出具体原因比如CHANNEL_ERROR、ACCESS_REFUSED。没有这个字段你只知道失败了不知道为什么失败。6. 消费端确认与生产端确认的配合别让消息在两头都丢了生产端做得再扎实消费端如果配错消息照样丢。这一节把两端的确认机制串起来讲因为它们是配套的。6.1 手动 ack 与自动 ack 的取舍Spring AMQP 的消费端确认模式有三种none自动确认消息一投递给消费者就认为消费成功。生产环境千万别用消费者处理到一半挂了消息就丢了。auto根据方法是否抛异常决定 ack 还是 nack。比 none 好但有个坑——如果方法执行成功但业务逻辑其实失败了比如返回了错误码但没抛异常消息照样被 ack 掉。manual手动调用basicAck或basicNack。最可靠但代码要自己写。我基本都用 manual配合 try-catch 精确控制。消费成功的业务逻辑走完再 ack出异常就 nack 并决定是否重新入队。6.2 消费端 nack 时 requeue 的陷阱basicNack有个requeue参数。设 true 消息会重新入队设 false 消息会被丢弃或进死信队列。这里有个经典陷阱如果消息本身有问题比如格式错误导致反序列化失败requeue 设 true 会导致消息无限循环——消费失败、重新入队、再消费失败、再入队CPU 直接跑满。我踩过一次一个格式错误的消息把整个消费者卡死了。正确做法是业务异常可以 requeue但要有重试次数上限格式异常、反序列化异常一律 requeuefalse直接进死信队列。判断依据是异常类型MessageConversionException这类直接判死BusinessException这类可以重试。6.3 生产端和消费端的确认如何形成闭环把两端串起来看一条消息的完整生命周期是这样的生产端发送Confirm 确认到达交换机。Return 确认路由到队列没触发 Return 就是路由成功。消息进队列等待消费。消费者拉取消息处理业务。处理成功手动 ack消息从队列删除。处理失败nack根据 requeue 决定重新入队还是进死信。任何一环出问题消息都可能丢。生产端的 Confirm 和 Return 保证消息进队列消费端的 manual ack 保证消息被正确处理。两头都做扎实整个链路才可靠。提示死信队列不是可选项是必选项。所有可能失败的消息最终都要有个去处否则要么丢要么无限循环。配一个死信交换机把重试耗尽的消息丢进去人工介入处理。7. 几个我踩过的坑和对应的规避手法这一节把散落在各处的经验集中一下都是实际项目里踩出来的文档上不一定写。7.1 序列化方式不一致导致的消息乱码生产端用Jackson2JsonMessageConverter消费端用默认的SimpleMessageConverter结果消费端拿到的是字节数组反序列化直接报错。这个坑的根源是两端的 MessageConverter 必须一致。我的做法是在配置类里统一定义一个Jackson2JsonMessageConverterBean生产端和消费端都注入它。如果跨语言比如生产端 Java、消费端 Python那就约定好 JSON 格式别用 Java 特有的序列化。7.2 回调不触发的几种情况Confirm 回调不触发排查下来通常是这几个原因publisher-confirm-type没设成correlated设成了simple或none。RabbitTemplate是手动 new 的没走 Spring 容器回调没注入。依赖版本冲突amqp-client版本过低不支持异步确认。Return 回调不触发基本就一个原因mandatory没设 true。这个我强调过好几次了因为它真的太容易漏。7.3 大批量发送时的内存问题做数据同步的时候一次性往队列里灌几十万条消息如果每条都new CorrelationData并存在一个 Map 里等确认内存直接爆掉。CorrelationData本身不大但几十万个累积起来也可观。我的处理方式是分批发送 批量确认。每批 1000 条发完等这批的确认结果用CompletableFuture.allOf聚合确认完再发下一批。这样内存占用可控吞吐量也不差。7.4 关于重试次数的经验值重试不是越多越好。我的经验是同步重试最多 1 次异步补偿最多 3 次。同步重试多了会阻塞业务异步补偿超过 3 次基本说明不是偶发问题再重试也是浪费资源不如直接告警让人来看。重试间隔用指数退避第一次 1 秒第二次 5 秒第三次 30 秒。别用固定间隔否则 Broker 压力大的时候重试会雪上加霜。8. 写在最后可靠投递是个系统工程把 Spring Boot 整合 RabbitMQ 的发送链路做可靠不是配几个参数就完事的。Confirm、Return、异常捕获、CorrelationData、消费端 manual ack、死信队列这些是一整套配合起来的东西缺一环整个链路就有漏洞。我个人的体会是先想清楚消息丢了会怎样再决定投入多少精力做可靠性。像日志采集这种丢几条无所谓的场景配个基础发送就行但订单、支付、积分这种涉及钱的Confirm Return 落库补偿 死信队列一个都不能少。技术方案永远是为业务服务的别为了看起来专业过度设计也别为了省事在关键链路上偷懒。最后分享一个我常用的自检清单上线前对着过一遍能挡掉大部分低级问题publisher-confirm-type是不是correlatedpublisher-returns和template.mandatory是不是都开了发送时有没有传CorrelationData里面是不是业务主键Confirm 和 Return 回调里有没有打全日志字段发送代码有没有 try-catch 兜底消费端是不是 manual acknack 的 requeue 逻辑有没有防死循环死信队列配了没有这七条过完生产端的可靠性基本就有底了。剩下的就是根据业务量调优参数那是另一个话题了。