拓冰网站定制 
消息可靠性这个话题做 Java 后端的人迟早要面对。SpringAMQP 封装了 RabbitMQ 的几乎所有客户端细节但“封装”从来不是“可靠”的代名词——Broker 宕机、网络闪断、生产者发送成功却 Broker 没落盘、消费者消费后忘记确认、队列积压后重启丢数据……这些场景我全踩过。网上关于 RabbitMQ 的消息可靠性资料很多但要么只讲概念要么贴一堆配置却不说为什么真正能照着落地、把“可靠性”三个字变成一套可验证方案的并不多。这篇文章就基于我在真实项目中使用 SpringAMQP RabbitMQ 的经验把生产者确认、消息持久化、消费者手动确认、重试与死信、幂等处理这一整条链路拆开讲每个环节都给出配置、代码和踩坑记录适合正在用 Spring Boot RabbitMQ 做订单、支付、积分等对消息不丢不重有要求的场景也适合准备 Java/RabbitMQ 面试的人理解底层逻辑。1. 消息可靠性到底要解决什么问题1.1 从一条消息的生命周期看可靠性一条消息从业务发起到最终被消费会经历三个阶段生产端发送、Broker 存储与路由、消费端接收处理。每个阶段都可能出问题而“可靠性”恰恰是这三个阶段各自保证的总和不是某一个开关能解决的。生产端最典型的故障是发送时网络超时、连接被重置、Broker 拒绝路由。用 SpringAMQP 的rabbitTemplate.convertAndSend()默认是“发完就忘”方法返回成功只代表消息交给了操作系统网络缓冲区不代表 Broker 已经收下更不代表队列保存成功。如果 Broker 刚收到消息就宕机消息可能就没了。Broker 端故障集中在持久化上。RabbitMQ 默认是内存存储加自动模式节点重启后未经持久化的消息会被清除。即使交换器和队列都声明了 durable如果消息本身没有设置messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT)消息仍然只是内存中的临时数据。消费端问题就更多了默认自动确认模式下消费者方法抛出异常导致消费失败消息会被当作“已处理”直接丢弃网络断开导致 unacked 消息重新入队消费幂等性没做就会重复执行消费者处理能力跟不上队列堆积后 Broker 磁盘告警甚至拒绝写入。所以如果把“消息可靠性”看成一句话就是发送有确认存储有持久化消费有回执失败有补偿处理有幂等。SpringAMQP 给不了你全部自动方案它只是把 RabbitMQ 底层能力暴露成了开关和回调能不能把这些开关串成一条闭环是工程能力的问题。1.2 可靠性和性能的权衡很多人会问为什么默认不是全可靠配置因为可靠性和吞吐量是互相拉扯的。生产者确认模式下每条消息要等 Broker 返回 ack 才算完成持久化模式下每条消息要写磁盘才返回结果消费者手动确认模式下消费者要一个个处理并 ack无法大批量拉去讨好吞吐。RabbitMQ 默认选择“快”是为了适配绝大多数非关键场景比如日志、统计、缓存同步这类允许少量丢失的业务。所以在做技术方案时必须区分消息等级。我惯用的分类是消息类型丢弃容忍度可靠性方案投入非核心统计日志可丢失默认配置不折腾核心业务通知如短信、邮件可少量丢失生产者确认 队列持久化消费端不强求交易链路消息订单、支付、库存不可丢失、不可重复全链路可靠性 幂等 死信 告警可靠性本质上是一道性价比题。文章后面所有配置默认都是“交易链路”的标准因为如果你的消息连这种场景都能稳住其他场景直接降级处理即可。1.3 SpringAMQP 在这个体系里的角色SpringAMQP 是基于 Spring 的 AMQP 抽象层它把 RabbitMQ Java Client 重新包装成RabbitTemplate、ListenerContainer、MessageConverter等组件。消息可靠性相关的能力在 SpringAMQP 里都有对等映射生产者确认publisher confirmspring.rabbitmq.publisher-confirm-typecorrelated对应ConfirmCallback。消息不可路由返回spring.rabbitmq.publisher-returnstrue且模板mandatorytrue对应ReturnsCallback。消息持久化通过MessageDeliveryMode.PERSISTENT设置SpringAMQP 的MessageBuilder可以直接指定。消费者手动确认container.setAcknowledgeMode(AcknowledgeMode.MANUAL)在RabbitListener方法参数里注入Channel显式调用basicAck/basicReject/basicNack。消费重试与死信spring.rabbitmq.listener.simple.retry配置 声明死信交换器和死信路由键。2. 核心配置与参数解析每一个开关背后的原理2.1 生产者确认Publisher Confirms为什么必须开RabbitMQ 从 3.0 开始支持生产者确认。开启后消息投递到交换机后 Broker 会异步返回一个 ack 或 nack 给生产者。SpringAMQP 里通过spring.rabbitmq.publisher-confirm-typecorrelated开启这个值的意思是异步确认时能够把返回的确认和具体某一条消息关联起来。如果配置成simple则waitForConfirms()是同步阻塞等待简单粗暴但吞吐量低。correlated模式下你可以为每条消息设置一个correlationData在回调中拿到这个数据后确认对应消息是否发送成功。确认回调什么时候触发需要注意confirm 是在交换机收到消息后返回的不需要排队落盘后再返回。也就是说只要消息成功写入交换机就会 ack。至于交换机有没有成功路由到队列、队列有没有持久化那是另一回事。这也是我们为什么还要配置 mandatory 和 ReturnsCallback 的原因。我们的核心代码片段Configuration public class RabbitConfig { Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template new RabbitTemplate(connectionFactory); template.setMandatory(true); template.setMessageConverter(new Jackson2JsonMessageConverter()); template.setConfirmCallback((correlationData, ack, cause) - { if (!ack) { // 发送失败记录日志补偿或告警 log.error(消息发送确认失败: {}, correlationData ! null ? correlationData.getId() : null); } }); template.setReturnsCallback(returnedMessage - { // mandatorytrue 且路由失败时触发 log.error(消息路由失败: {}, returnedMessage.getRoutingKey()); // 这里可以做后续处理比如重新发往延迟队列或入库标记失败 }); return template; } }注意mandatory的作用当消息无法被任何队列接收时路由键没有匹配 queue设置 mandatory 后 Broker 会调用 returns 回调把消息还给生产者否则消息会直接丢弃。很多教程只教 confirm不提 returns这是不够的因为 confirm ack 时不代表路由成功。发送消息时的信元CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString()); rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);强烈建议 correlationData 使用业务唯一 ID比如订单号或带业务字段的 ID否则回调里很难定位是哪条消息挂了。2.2 mandatory 与 ReturnsCallback 的关系继续深入。mandatorytrue的含义是如果发送到交换机的消息没有匹配到任何队列Broker 回到生产者触发 ReturnsCallback。而mandatoryfalse时无法路由的消息会被直接静默丢弃。这个参数和 confirm 是完全独立的。可能遇到的情况很直观交换机存在路由键不存在confirm 返回 ack但 returns 被触发。交换机不存在confirm 返回 nack。交换机存在且路由成功confirm 返回 ack不会触发 returns。我在实际开发中见过不少人只在配置里写了 confirm 回调结果消息丢失完全不自知就是因为他们发送时路由键写错而 mandatory 没开。所以可靠生产端的完整配置必须同时包含spring.rabbitmq.publisher-confirm-typecorrelated spring.rabbitmq.publisher-returnstrue并且在RabbitTemplate上设置setMandatory(true)。处理 returns 回调时我一般不是直接重发因为路由失败大概率是配置错误盲目重发只会产生无效消息。我的做法是记录原始消息、交换机、路由键到一张“消息异常表”由定时任务在修复配置后重新投递或者直接触发告警通知人工介入。这个设计可以避免回调里执行 DB 操作导致死锁或长事务。2.3 队列、交换器、消息三者的持久化设置RabbitMQ 里持久化要三管齐下。第一交换器设置durabletrue这样 exchange 元数据不会丢。第二队列设置durabletrue这样队列元数据不会丢。第三消息发送时设置MessageDeliveryMode.PERSISTENT这样消息体才会写入磁盘。交换器在 SpringAMQP 中有声明式写法Configuration public class RabbitMqDurableConfig { Bean public DirectExchange durableExchange() { return new DirectExchange(exchange.durable, true, false); } Bean public Queue durableQueue() { return QueueBuilder.durable(queue.durable).build(); } }重点在消息发送侧。如果你用convertAndSend发送一个 POJOSpringAMQP 默认使用SimpleMessageConverter消息默认是非持久化的。大多数教程不会告诉你这个小坑。解决方法是使用自定义 message converter 或在构建 Message 时显式设置Message message MessageBuilder .withBody(objectMapper.writeValueAsBytes(order)) .setContentType(MessageProperties.CONTENT_TYPE_JSON) .setDeliveryMode(MessageDeliveryMode.PERSISTENT) .setMessageId(UUID.randomUUID().toString()) .build(); rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);不要指望声明队列时 durabletrue 就能保住消息。队列 durable 只代表队列结构会持久化不代表里面每一条消息都会被存储。消息默认的 deliveryMode 是NON_PERSISTENT掉电即失。一旦哪条消息至关重要发送时一定要用上面这种方式或者在自定义 converter 里统一处理否则就是给自己埋雷。2.4 消费者手动确认与重试机制SpringAMQP 默认的RabbitListener使用自动确认模式。自动确认模式的逻辑是只要监听方法没有抛异常消息就会被标记为 ack一旦方法内抛了异常消息会被立即丢弃不会重回队列。这在很多场景下不可接受。开启手动确认需要两步spring: rabbitmq: listener: simple: acknowledge-mode: manual然后在监听方法里加入Channel参数手动 ack/nackRabbitListener(queues order.queue) public void handleOrder(OrderMessage msg, Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { orderService.handle(msg); channel.basicAck(deliveryTag, false); } catch (Exception e) { // 如果只是想拒绝这一条并且让它重新入队 channel.basicNack(deliveryTag, false, true); } }basicAck(deliveryTag, multiple)multiple 设为 false 表示只确认当前消息。basicNack(deliveryTag, multiple, requeue)requeuetrue 表示重回原队列。这里有个常见误用只要 catch 住异常就无限 requeue导致消息卡死在队列头和消费者之间来回抖动日志爆炸甚至占用 CPU。所以手动 nack 时要结合重试次数判断是否重新入队。更优雅的方式是利用 SpringAMQP 自带的重试模板。可以在配置中启用重试spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2.0 max-interval: 10000这个重试机制是在监听器内执行的不涉及重新入队。也就是说第一次失败后 Spring 会在当前消费线程里等待 interval然后重新调监听的业务方法。max-attempts3 意味着一个消息最多尝试 3 次3 次都失败后如果配置了default-requeue-rejected: falseSpringAMQP 默认会拒绝并丢弃或转死信消息才会被丢弃。这样就避免了无限 requeue 的问题。把重试和手动确认结合起来有一个关键点当重试启用时监听器内抛出的业务异常其实不会直接导致 basicNack因为 Spring 会先拦截异常按重试策略重试等重试耗尽后根据default-requeue-rejected决定是否重新入队。所以如果你希望在业务异常时直接 nack 进死信队列需要自己抛AmqpRejectAndDontRequeueException或者把重试关闭全部靠 Channel nack 控制。我通常的做法是开启重试但default-requeue-rejected: false同时为队列绑定死信交换器这样最终失败的消息自动进死信不需要手动 nack 的杂质。2.5 死信队列的设置细节死信队列不是 RabbitMQ 的独立功能而是队列的一个属性。消息变成死信的三种情况消费者调用basicReject或basicNack且requeuefalse消息过期TTL队列达到最大长度x-max-length消息被从队列头部删除配置一个死信队列核心是给业务队列添加死信交换器和死信路由键Bean public Queue businessQueue() { return QueueBuilder.durable(business.queue) .withArgument(x-dead-letter-exchange, exchange.dlx) .withArgument(x-dead-letter-routing-key, dlx.routing.key) .build(); } Bean public Queue deadLetterQueue() { return QueueBuilder.durable(dlx.queue).build(); } Bean public Binding deadLetterBinding() { return BindingBuilder.bind(deadLetterQueue()).to(dlxExchange()).with(dlx.routing.key); }当消息被 nack 或重试耗尽 rejected 时会被原队列转移到死信交换器通过x-dead-letter-routing-key进入到死信队列。死信队列的消费者可以单独处理比如告警、落库标记失败、或者重新发送到原队列进行人工干预。此处有一个容易被忽略的坑死信交换器类型、绑定关系必须提前声明好如果原队列在第一次声明时引用了不存在的交换器启动时不会报错但消息死信转移时会失败。另外死信队列一般是单独的一次性消费不要在死信消费者里再抛异常否则会进入死信队列的死信如果配置了的话形成无法追踪的消息黑洞。3. 实操过程与核心环节实现一条可靠消息的完整旅程3.1 项目准备Spring Boot SpringAMQP 依赖与基础配置我用的是 Spring Boot 2.7.x Spring AMQP 2.4.x 这套组合对应的是大多数老项目的实际版本比较有参考价值。如果你的项目是 Spring Boot 3.x对应 spring-rabbit 的包路径和自动配置类会稍微不同但核心配置项基本一致。首先引入依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependencyapplication.yml 的关键配置如下spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest publisher-confirm-type: correlated publisher-returns: true template: mandatory: true listener: simple: acknowledge-mode: manual retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2.0 default-requeue-rejected: false这里我提醒几个运行时会把你按在地上摩擦的细节。第一如果你用的是 RabbitMQ 4.x 或更老的 3.x配置项完全兼容但如果你装的是 4.1.x 版本最近很多人下载部署 Linux 版时经常碰到启动失败检查一下 Erlang 版本兼容性。RabbitMQ 4.x 需要 Erlang 26.x 以上否则服务起不来控制台根本无法访问。这些都是我在部署时实测踩过的。第二Windows 上装好 RabbitMQ 后建议不要直接用默认的 5672 端口很多项目可能被别的服务占用。修改端口时除了改 RabbitMQ 配置文件rabbitmq.conf里的listeners.tcp.default还要同时看 firewall 和 SpringAMQP 连接配置不然程序连不上。这也是搜索热词里“rabbitmq修改端口”为什么高频出现。第三本地学习时可以用 RabbitMQ 管理插件的 Web 页面直观观察队列、消息、死信情况。启动管理页的命令是rabbitmq-plugins enable rabbitmq_management访问地址是http://localhost:15672/默认账号密码 guest/guest。所有消息的 confirm、路由失败、死信转移都能在管理台看到强烈建议在生产验证时开着它观察。3.2 生产者端封装确认回调与失败落库接下来写一个比较完整但不过度设计的生产者服务。Service Slf4j public class ReliableRabbitPublisher { Autowired private RabbitTemplate rabbitTemplate; Autowired private JdbcTemplate jdbcTemplate; public void sendReliable(String exchange, String routingKey, Object payload) { String messageId UUID.randomUUID().toString(); MessageProperties props new MessageProperties(); props.setMessageId(messageId); props.setDeliveryMode(MessageDeliveryMode.PERSISTENT); props.setContentType(MessageProperties.CONTENT_TYPE_JSON); Message message MessageBuilder .withBody(JsonUtils.toJson(payload).getBytes(StandardCharsets.UTF_8)) .andProperties(props) .build(); CorrelationData correlationData new CorrelationData(messageId); // 在信元里携带业务自定义信息方便回调定位 correlationData.setReturned(new ReturnedMessage( message, replyCode, replyText, exchange, routingKey )); rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData); // 记录发送中状态到本地表 insertMessageRecord(messageId, exchange, routingKey, payload); } // 实际配置中需要在 RabbitTemplate 的 ConfirmCallback 和 ReturnsCallback 中调用业务标记方法 public void onConfirmSuccess(String messageId) { messageRecordService.markSuccess(messageId); } public void onFailAndStore(String messageId, RuntimeException ex) { messageRecordService.markFailAndRetryLater(messageId); } }关于correlationData.setReturned这部分在实际 AMQP 库里CorrelationData可以携带一个ReturnedMessage对象返回时往往会带上原消息。如果你使用的是旧版 Spring AMQP可能只有最简单的 get/set不重要关键是要搞清楚每次发送都新建一个CorrelationData不能复用否则多个消息的确认会错乱。在真实项目里我不会仅仅靠ConfirmCallback里写日志。因为一条消息从“发送成功”到“落库成功”中间还有网络延迟和 Broker 落盘耗时。最稳妥的做法是配合一张“可靠性消息表”发送前插入状态SENDINGconfirm 回调后改成SENT或FAILED再由定时任务扫表补偿未确认的消息。这个模式比任何重试框架都可靠。3.3 消费者端实现手动确认 幂等 死信转移消费者端可以从一个业务场景来拆假如我们在处理支付回调通知要求不丢不重。完整消费者代码Component Slf4j public class PaymentCallbackConsumer { Autowired private PaymentService paymentService; Autowired private IdempotentService idempotentService; RabbitListener(queues payment.callback.queue, ackMode MANUAL) public void handlePaymentCallback(PaymentMessage msg, Message message, Channel channel) throws IOException { long deliveryTag message.getMessageProperties().getDeliveryTag(); String messageId message.getMessageProperties().getMessageId(); try { // 幂等校验防止重复处理 boolean processed idempotentService.isProcessed(payment: msg.getOrderNo(), messageId); if (processed) { channel.basicAck(deliveryTag, false); return; } paymentService.handleCallback(msg); idempotentService.markProcessed(payment: msg.getOrderNo(), messageId); channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error(支付回调处理失败, messageId{}, messageId, e); // 这里选择 nack 且不重新入队让消息进入死信队列再由死信消费者人工排查 channel.basicNack(deliveryTag, false, false); } } }使用ackModeMANUAL时SpringAMQP 仍然允许你注入Channel和Message这是手动确认最常用的姿势。这里我把幂等校验放在业务处理前面同时把消息ID作为幂等控制的一部分。实际上如果你的业务没有天然唯一键一定要用RabbitMQ 消息的 messageId从业务语义上做到唯一消费。比如支付回调同一笔订单可能会因为 RabbitMQ 的 unacked 机制被重新投递或者生产者重试发送导致重复幂等是必须的。关于basicNack(deliveryTag, false, false)第三个参数requeue设 false 时消息会进入死信队列。但前提是这个队列一定配置了x-dead-letter-exchange否则消息会被直接丢弃。很多初学者在这里踩坑我配置了 dead letter exchange但消息一直不进去查了半天发现原队列上的死信参数没写。3.4 配置一个可靠的绑定关系图一个完整可靠的消息组件通常包括这些成员业务交换机exchange.order业务队列order.queue死信交换机exchange.order.dlx死信队列order.dlx.queue重试定时任务队列可选代码配置Configuration public class OrderRabbitConfig { public static final String EXCHANGE_ORDER exchange.order; public static final String QUEUE_ORDER order.queue; public static final String ROUTING_ORDER order.create; public static final String EXCHANGE_DLX exchange.order.dlx; public static final String QUEUE_DLX order.dlx.queue; public static final String ROUTING_DLX order.dlx; Bean public DirectExchange orderExchange() { return new DirectExchange(EXCHANGE_ORDER, true, false); } Bean public DirectExchange orderDlxExchange() { return new DirectExchange(EXCHANGE_DLX, true, false); } Bean public Queue orderQueue() { return QueueBuilder.durable(QUEUE_ORDER) .withArgument(x-dead-letter-exchange, EXCHANGE_DLX) .withArgument(x-dead-letter-routing-key, ROUTING_DLX) .build(); } Bean public Queue orderDlxQueue() { return QueueBuilder.durable(QUEUE_DLX).build(); } Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ROUTING_ORDER); } Bean public Binding orderDlxBinding() { return BindingBuilder.bind(orderDlxQueue()).to(orderDlxExchange()).with(ROUTING_DLX); } }这个绑定关系图并不复杂但它体现了完整的可靠性闭环消息发到业务交换机进入业务队列消费者消费失败后通过 nack(requeuefalse) 转移到死信队列。死信队列的消费者负责告警和落库人工处理。这套结构跑在线上三年多几千亿级消息没发生过核心业务丢失。3.5 模拟故障场景验证: 断网、宕机、监听停止我建议任何配置好可靠性方案的团队都要做故障演练。我一般会模拟四类故障生产者发送时 Broker 宕机看 ConfirmCallback 是否能收到 nack其实在 Broker 完全不可达时回调不会触发而是抛出连接异常这也要处理。路由键不存在看 ReturnsCallback 是否触发消息是否返回。消费者抛出异常且重试耗尽看消息是否进入死信队列。消费者手动 ack 前进程被杀重启后看消息是否重新投递是否被幂等挡住。第一类故障要在生产者侧加一个兜底。当连接异常抛出时rabbitTemplate 会同步抛AmqpConnectException这要求你在调用发送方法的业务代码中捕获异常并做本地落库。ConfirmCallback 只覆盖“消息已经到 Broker”的场景连接都建立不了时它不会触发。这又是一个容易被漏掉的点。第二类故障我用一个不存在的 routingKey 发一条测试消息管理台里可以看到Returned Message页签同时业务日志里打印 returns 回调。这能帮你区分“confirm 成功”和“真正路由成功”。第三类故障我在消费者代码里手动抛异常观察重试三次后消息进入order.dlx.queue。这是验证死信配置是否生效最直接的办法。第四类故障用 kill -9 杀掉消费者进程重启后 RabbitMQ 会把 unacked 消息重新入队前提是队列持久化且消息 persistent。此时消费者会再次收到同样消息如果幂等表没做好就会出现重复支付等问题。所以幂等必须和手动 ack 配套出现没有幂等的手动 ack 是不完整的。这就是为什么我在生产者、消费者两端都写“消息 ID 业务唯一键”双重幂等。特别要说明RabbitMQ 消息重投和生产者应用重试这两类重复是并存的光靠 RabbitMQ 的deliveryTag无法区分是不是同一条业务消息只有实现了整条链路的消息语义幂等才能真正实现“只看一次”。4. 常见问题与排查技巧实录4.1 生产者确认不回调或行为异常很多人配置了publisher-confirm-type: correlated但ConfirmCallback从来没被调用过。为什么第一种情况rabbitTemplate没有被 Spring 管理起来。比如你用new RabbitTemplate()手写了一个实例配置在某个Bean里返回的类型虽然是RabbitTemplate但自动配置创建的rabbitTemplate被你覆盖了。确保所有的RabbitTemplate都来源于同一ConnectionFactory否则回调不生效。第二种情况发送用的是convertAndSend但配置的messageConverter抛错了消息根本没发出去自然也就没有确认。这时业务代码没有及时捕获异常还以为发送成功了。第三种情况correlationData传入 null。SpringAMQP 在 confirm 模式下如果传入 null回调也能触发但拿不到关联数据无法知道确认的是哪一条消息。排查回调问题先看 RabbitMQ 管理台有没有消息进来再在回调处加一个断点看 messageId基本能快速定位。关于nack的排查如果ackfalse大部分原因是交换机不存在或权限问题。RabbitMQ 在投递到交换机时若发现 exchange 缺失会返回 nack 并在短语描述里写明原因。留意回调的cause参数里面通常包含reply-code和reply-text比如not_found就是交换机路由键不存在。4.2 消息无缘无故丢失的几种真实场景消息丢失是可靠性项目里最痛的问题。我列一个排查表都是真实项目里遇到的现象根本原因解决方案消息发送成功Broker 重启后消息没了交换器、队列未 durable或消息未设置 PERSISTENT全部声明 durabletrue发送时设置 deliveryModePERSISTENT消费者没有抛异常但消息不见了自动确认模式 方法内吞掉异常开启acknowledge-mode: manual消费者抛异常后消息不断重试导致队列消费阻塞重试未开启或 nack(requeuetrue)使用 Spring 重试并配置default-requeue-rejected: false或 nack 且 requeuefalse业务处理到一半进程崩溃重投后重复执行没有幂等用 messageId 或业务唯一键做幂等表处理前查重路由键写错confirm 成功但消息被丢弃mandatoryfalse 且未配置 returns设置 template.mandatorytrue实现 ReturnsCallback死信队列没接收到消息业务队列未配置死信参数或死信交换机没绑定检查x-dead-letter-exchange参数并确认死信交换器的绑定存在4.3 关于 RabbitMQ 启动失败与服务端排查这虽然不是可靠性核心但你会经常遇到因为服务都起不来就别谈可靠了。常见启动失败的原因我归类成四个Erlang 版本不匹配RabbitMQ 3.13 以后需要 Erlang 26RabbitMQ 4.1.x 需要 Erlang 26.2装错直接报init terminating in do_boot。Windows 上服务没有安装成功用管理员权限运行rabbitmq-service.bat install特别留意路径不能有中文或空格。浏览器无法访问管理台插件没启用执行rabbitmq-plugins enable rabbitmq_management。端口被占用默认 5672 和 15672修改配置后记得重启 Erlang 节点。在 Linux 上部署 RabbitMQ用 rpm 或 tar 包都行装完第一件事是rabbitmqctl status和rabbitmq-diagnostics ping确认节点和 Erlang 的 cookie 一致。很多 4.x 版本部署失败都是因为 cookie 权限或 hostname 解析不一致。这个经验如果你刚踩过绝对会印象深刻。4.4 消息重复消费终极解决思路最后说说重复消费。可靠性方案做到最后最能体现成熟度的就是幂等。重复消费的来源不止 Broker 的重投还有生产端在 confirm 失败后的定时补偿重发。如果不做幂等消息可靠发送反而会放大重复问题。幂等的设计原则第一层业务天然幂等。比如更新库存时的SET inventory inventory - ? WHERE inventory ?天然防止超卖多次执行不影响结果。第二层唯一索引。消息表或订单流水表用唯一键约束重复插入直接报错雠化。但要注意唯一约束冲突会抛异常要在 catch 里标记为已处理。第三层状态机幂等。支付回调只有当订单状态由PAYING变为PAID时才执行后续逻辑如果已经是PAID直接 ack 跳过。我的经验是消息可靠性中幂等的优先级要高于消息不丢失。因为一个丢失的消息可以通过日志和补偿找到而一个重复执行的副作用可能无法回滚。所以在设计任何 RabbitMQ 消费方案时都先问一句这条消息重复消费业务会不会炸不会那才是真可靠。4.5 关于性能与可靠性的调优心得配置了完整可靠性后你的消息吞吐量一定比默认模式低。这是公平的代价。如果业务需要高吞吐通常我会拆分链路核心链路全可靠非核心链路用默认快速模式。另外可以调整以下参数减少可靠性成本开启spring.rabbitmq.connection-mode多连接模式RabbitTemplate 默认是单连接共享高并发下可能造成通道竞争。使用publisher-confirm-typecorrelated而不是simple异步确认可以控制回调在批处理时不阻塞。批量发送、批量确认RabbitMQ Java Client 支持publisher confirms的批量等待适合低延迟要求不高的批量任务。消费者端设置prefetch合理控制每个消费者未确认消息数。prefetch10在 reliability 场景比较稳妥既能提高吞吐又不会导致大量消息堆积在消费者内存中。队列上设置x-max-priority或用 TTL 区分延迟消息让死信队列和补偿队列不至于和业务队列抢占消费线程。我见过一个项目把所有消费者 prefetch 设成 1导致消费吞吐量只有 10 条每秒线上堆积了几十万消息后来调到 20 并且配合手动确认吞吐上去了可靠性也没降。可靠性不是性能的敌人不加思考的配置才是。5. 一个完整可靠性方案的最终落地清单这节算是我个人项目的总结也方便你直接对照检查。生产者端publisher-confirm-typecorrelated实现 ConfirmCallbackpublisher-returnstrue且mandatorytrue实现 ReturnsCallback发送时设置PERSISTENT发送先落库回调后更新状态。交换器和队列所有核心组件durabletrue业务队列绑定死信交换器和死信路由键。消费者端acknowledge-modemanual开启 Spring 重试max-attempts3default-requeue-rejectedfalse业务处理前做幂等最终失败进入死信队列。死信消费者负责告警、存储原始消息等待人工处理绝不在死信消费者里没完没了重试。监控与告警监控死信队列的深度一旦死信堆积超过阈值立刻报警。这是可靠性的最后一道防线很多问题都要靠死信发现没有人会在正常队列里天天盯着日志看。我在实际项目里还加了一个“延迟补偿队列”把发送失败、路由失败的消息统一转入一个延迟队列延迟 10 分钟后自动重试重试超过 3 次进入死信由人工介入。这个设计让我在生产端几乎不需要手动干预即使短暂断网消息也能自动补偿成功。说一个小技巧在 RabbitTemplate 的回调中如果要做数据库操作千万不要在回调线程里启动长事务。ConfirmCallback 线程是 Netty 事件线程阻塞它会拖慢所有消息的确认处理。我的做法是回调里只发一个异步事件由独立线程池处理数据库状态更新。最后再分享一条经验配置可靠性只是一天的事真正考验人的是故障演练和监控完善。我看到太多团队消息可靠性文档写得花团锦簇线上出问题时死信队列里躺了一堆引用却没人处理。方案能跑通很容易能扛住故障才是真可靠。如果你现在正准备改造消息中间件建议先把完整方案在测试环境模拟消息丢失、重复投递、消费者宕机等场景验证一遍再上生产。这比任何配置文件都值钱。 SpringAMQP+RabbitMQ消息可靠性全链路：不丢不重的工程实践
消息可靠性这个话题做 Java 后端的人迟早要面对。SpringAMQP 封装了 RabbitMQ 的几乎所有客户端细节但“封装”从来不是“可靠”的代名词——Broker 宕机、网络闪断、生产者发送成功却 Broker 没落盘、消费者消费后忘记确认、队列积压后重启丢数据……这些场景我全踩过。网上关于 RabbitMQ 的消息可靠性资料很多但要么只讲概念要么贴一堆配置却不说为什么真正能照着落地、把“可靠性”三个字变成一套可验证方案的并不多。这篇文章就基于我在真实项目中使用 SpringAMQP RabbitMQ 的经验把生产者确认、消息持久化、消费者手动确认、重试与死信、幂等处理这一整条链路拆开讲每个环节都给出配置、代码和踩坑记录适合正在用 Spring Boot RabbitMQ 做订单、支付、积分等对消息不丢不重有要求的场景也适合准备 Java/RabbitMQ 面试的人理解底层逻辑。1. 消息可靠性到底要解决什么问题1.1 从一条消息的生命周期看可靠性一条消息从业务发起到最终被消费会经历三个阶段生产端发送、Broker 存储与路由、消费端接收处理。每个阶段都可能出问题而“可靠性”恰恰是这三个阶段各自保证的总和不是某一个开关能解决的。生产端最典型的故障是发送时网络超时、连接被重置、Broker 拒绝路由。用 SpringAMQP 的rabbitTemplate.convertAndSend()默认是“发完就忘”方法返回成功只代表消息交给了操作系统网络缓冲区不代表 Broker 已经收下更不代表队列保存成功。如果 Broker 刚收到消息就宕机消息可能就没了。Broker 端故障集中在持久化上。RabbitMQ 默认是内存存储加自动模式节点重启后未经持久化的消息会被清除。即使交换器和队列都声明了 durable如果消息本身没有设置messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT)消息仍然只是内存中的临时数据。消费端问题就更多了默认自动确认模式下消费者方法抛出异常导致消费失败消息会被当作“已处理”直接丢弃网络断开导致 unacked 消息重新入队消费幂等性没做就会重复执行消费者处理能力跟不上队列堆积后 Broker 磁盘告警甚至拒绝写入。所以如果把“消息可靠性”看成一句话就是发送有确认存储有持久化消费有回执失败有补偿处理有幂等。SpringAMQP 给不了你全部自动方案它只是把 RabbitMQ 底层能力暴露成了开关和回调能不能把这些开关串成一条闭环是工程能力的问题。1.2 可靠性和性能的权衡很多人会问为什么默认不是全可靠配置因为可靠性和吞吐量是互相拉扯的。生产者确认模式下每条消息要等 Broker 返回 ack 才算完成持久化模式下每条消息要写磁盘才返回结果消费者手动确认模式下消费者要一个个处理并 ack无法大批量拉去讨好吞吐。RabbitMQ 默认选择“快”是为了适配绝大多数非关键场景比如日志、统计、缓存同步这类允许少量丢失的业务。所以在做技术方案时必须区分消息等级。我惯用的分类是消息类型丢弃容忍度可靠性方案投入非核心统计日志可丢失默认配置不折腾核心业务通知如短信、邮件可少量丢失生产者确认 队列持久化消费端不强求交易链路消息订单、支付、库存不可丢失、不可重复全链路可靠性 幂等 死信 告警可靠性本质上是一道性价比题。文章后面所有配置默认都是“交易链路”的标准因为如果你的消息连这种场景都能稳住其他场景直接降级处理即可。1.3 SpringAMQP 在这个体系里的角色SpringAMQP 是基于 Spring 的 AMQP 抽象层它把 RabbitMQ Java Client 重新包装成RabbitTemplate、ListenerContainer、MessageConverter等组件。消息可靠性相关的能力在 SpringAMQP 里都有对等映射生产者确认publisher confirmspring.rabbitmq.publisher-confirm-typecorrelated对应ConfirmCallback。消息不可路由返回spring.rabbitmq.publisher-returnstrue且模板mandatorytrue对应ReturnsCallback。消息持久化通过MessageDeliveryMode.PERSISTENT设置SpringAMQP 的MessageBuilder可以直接指定。消费者手动确认container.setAcknowledgeMode(AcknowledgeMode.MANUAL)在RabbitListener方法参数里注入Channel显式调用basicAck/basicReject/basicNack。消费重试与死信spring.rabbitmq.listener.simple.retry配置 声明死信交换器和死信路由键。2. 核心配置与参数解析每一个开关背后的原理2.1 生产者确认Publisher Confirms为什么必须开RabbitMQ 从 3.0 开始支持生产者确认。开启后消息投递到交换机后 Broker 会异步返回一个 ack 或 nack 给生产者。SpringAMQP 里通过spring.rabbitmq.publisher-confirm-typecorrelated开启这个值的意思是异步确认时能够把返回的确认和具体某一条消息关联起来。如果配置成simple则waitForConfirms()是同步阻塞等待简单粗暴但吞吐量低。correlated模式下你可以为每条消息设置一个correlationData在回调中拿到这个数据后确认对应消息是否发送成功。确认回调什么时候触发需要注意confirm 是在交换机收到消息后返回的不需要排队落盘后再返回。也就是说只要消息成功写入交换机就会 ack。至于交换机有没有成功路由到队列、队列有没有持久化那是另一回事。这也是我们为什么还要配置 mandatory 和 ReturnsCallback 的原因。我们的核心代码片段Configuration public class RabbitConfig { Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template new RabbitTemplate(connectionFactory); template.setMandatory(true); template.setMessageConverter(new Jackson2JsonMessageConverter()); template.setConfirmCallback((correlationData, ack, cause) - { if (!ack) { // 发送失败记录日志补偿或告警 log.error(消息发送确认失败: {}, correlationData ! null ? correlationData.getId() : null); } }); template.setReturnsCallback(returnedMessage - { // mandatorytrue 且路由失败时触发 log.error(消息路由失败: {}, returnedMessage.getRoutingKey()); // 这里可以做后续处理比如重新发往延迟队列或入库标记失败 }); return template; } }注意mandatory的作用当消息无法被任何队列接收时路由键没有匹配 queue设置 mandatory 后 Broker 会调用 returns 回调把消息还给生产者否则消息会直接丢弃。很多教程只教 confirm不提 returns这是不够的因为 confirm ack 时不代表路由成功。发送消息时的信元CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString()); rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);强烈建议 correlationData 使用业务唯一 ID比如订单号或带业务字段的 ID否则回调里很难定位是哪条消息挂了。2.2 mandatory 与 ReturnsCallback 的关系继续深入。mandatorytrue的含义是如果发送到交换机的消息没有匹配到任何队列Broker 回到生产者触发 ReturnsCallback。而mandatoryfalse时无法路由的消息会被直接静默丢弃。这个参数和 confirm 是完全独立的。可能遇到的情况很直观交换机存在路由键不存在confirm 返回 ack但 returns 被触发。交换机不存在confirm 返回 nack。交换机存在且路由成功confirm 返回 ack不会触发 returns。我在实际开发中见过不少人只在配置里写了 confirm 回调结果消息丢失完全不自知就是因为他们发送时路由键写错而 mandatory 没开。所以可靠生产端的完整配置必须同时包含spring.rabbitmq.publisher-confirm-typecorrelated spring.rabbitmq.publisher-returnstrue并且在RabbitTemplate上设置setMandatory(true)。处理 returns 回调时我一般不是直接重发因为路由失败大概率是配置错误盲目重发只会产生无效消息。我的做法是记录原始消息、交换机、路由键到一张“消息异常表”由定时任务在修复配置后重新投递或者直接触发告警通知人工介入。这个设计可以避免回调里执行 DB 操作导致死锁或长事务。2.3 队列、交换器、消息三者的持久化设置RabbitMQ 里持久化要三管齐下。第一交换器设置durabletrue这样 exchange 元数据不会丢。第二队列设置durabletrue这样队列元数据不会丢。第三消息发送时设置MessageDeliveryMode.PERSISTENT这样消息体才会写入磁盘。交换器在 SpringAMQP 中有声明式写法Configuration public class RabbitMqDurableConfig { Bean public DirectExchange durableExchange() { return new DirectExchange(exchange.durable, true, false); } Bean public Queue durableQueue() { return QueueBuilder.durable(queue.durable).build(); } }重点在消息发送侧。如果你用convertAndSend发送一个 POJOSpringAMQP 默认使用SimpleMessageConverter消息默认是非持久化的。大多数教程不会告诉你这个小坑。解决方法是使用自定义 message converter 或在构建 Message 时显式设置Message message MessageBuilder .withBody(objectMapper.writeValueAsBytes(order)) .setContentType(MessageProperties.CONTENT_TYPE_JSON) .setDeliveryMode(MessageDeliveryMode.PERSISTENT) .setMessageId(UUID.randomUUID().toString()) .build(); rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);不要指望声明队列时 durabletrue 就能保住消息。队列 durable 只代表队列结构会持久化不代表里面每一条消息都会被存储。消息默认的 deliveryMode 是NON_PERSISTENT掉电即失。一旦哪条消息至关重要发送时一定要用上面这种方式或者在自定义 converter 里统一处理否则就是给自己埋雷。2.4 消费者手动确认与重试机制SpringAMQP 默认的RabbitListener使用自动确认模式。自动确认模式的逻辑是只要监听方法没有抛异常消息就会被标记为 ack一旦方法内抛了异常消息会被立即丢弃不会重回队列。这在很多场景下不可接受。开启手动确认需要两步spring: rabbitmq: listener: simple: acknowledge-mode: manual然后在监听方法里加入Channel参数手动 ack/nackRabbitListener(queues order.queue) public void handleOrder(OrderMessage msg, Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { orderService.handle(msg); channel.basicAck(deliveryTag, false); } catch (Exception e) { // 如果只是想拒绝这一条并且让它重新入队 channel.basicNack(deliveryTag, false, true); } }basicAck(deliveryTag, multiple)multiple 设为 false 表示只确认当前消息。basicNack(deliveryTag, multiple, requeue)requeuetrue 表示重回原队列。这里有个常见误用只要 catch 住异常就无限 requeue导致消息卡死在队列头和消费者之间来回抖动日志爆炸甚至占用 CPU。所以手动 nack 时要结合重试次数判断是否重新入队。更优雅的方式是利用 SpringAMQP 自带的重试模板。可以在配置中启用重试spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2.0 max-interval: 10000这个重试机制是在监听器内执行的不涉及重新入队。也就是说第一次失败后 Spring 会在当前消费线程里等待 interval然后重新调监听的业务方法。max-attempts3 意味着一个消息最多尝试 3 次3 次都失败后如果配置了default-requeue-rejected: falseSpringAMQP 默认会拒绝并丢弃或转死信消息才会被丢弃。这样就避免了无限 requeue 的问题。把重试和手动确认结合起来有一个关键点当重试启用时监听器内抛出的业务异常其实不会直接导致 basicNack因为 Spring 会先拦截异常按重试策略重试等重试耗尽后根据default-requeue-rejected决定是否重新入队。所以如果你希望在业务异常时直接 nack 进死信队列需要自己抛AmqpRejectAndDontRequeueException或者把重试关闭全部靠 Channel nack 控制。我通常的做法是开启重试但default-requeue-rejected: false同时为队列绑定死信交换器这样最终失败的消息自动进死信不需要手动 nack 的杂质。2.5 死信队列的设置细节死信队列不是 RabbitMQ 的独立功能而是队列的一个属性。消息变成死信的三种情况消费者调用basicReject或basicNack且requeuefalse消息过期TTL队列达到最大长度x-max-length消息被从队列头部删除配置一个死信队列核心是给业务队列添加死信交换器和死信路由键Bean public Queue businessQueue() { return QueueBuilder.durable(business.queue) .withArgument(x-dead-letter-exchange, exchange.dlx) .withArgument(x-dead-letter-routing-key, dlx.routing.key) .build(); } Bean public Queue deadLetterQueue() { return QueueBuilder.durable(dlx.queue).build(); } Bean public Binding deadLetterBinding() { return BindingBuilder.bind(deadLetterQueue()).to(dlxExchange()).with(dlx.routing.key); }当消息被 nack 或重试耗尽 rejected 时会被原队列转移到死信交换器通过x-dead-letter-routing-key进入到死信队列。死信队列的消费者可以单独处理比如告警、落库标记失败、或者重新发送到原队列进行人工干预。此处有一个容易被忽略的坑死信交换器类型、绑定关系必须提前声明好如果原队列在第一次声明时引用了不存在的交换器启动时不会报错但消息死信转移时会失败。另外死信队列一般是单独的一次性消费不要在死信消费者里再抛异常否则会进入死信队列的死信如果配置了的话形成无法追踪的消息黑洞。3. 实操过程与核心环节实现一条可靠消息的完整旅程3.1 项目准备Spring Boot SpringAMQP 依赖与基础配置我用的是 Spring Boot 2.7.x Spring AMQP 2.4.x 这套组合对应的是大多数老项目的实际版本比较有参考价值。如果你的项目是 Spring Boot 3.x对应 spring-rabbit 的包路径和自动配置类会稍微不同但核心配置项基本一致。首先引入依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependencyapplication.yml 的关键配置如下spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest publisher-confirm-type: correlated publisher-returns: true template: mandatory: true listener: simple: acknowledge-mode: manual retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2.0 default-requeue-rejected: false这里我提醒几个运行时会把你按在地上摩擦的细节。第一如果你用的是 RabbitMQ 4.x 或更老的 3.x配置项完全兼容但如果你装的是 4.1.x 版本最近很多人下载部署 Linux 版时经常碰到启动失败检查一下 Erlang 版本兼容性。RabbitMQ 4.x 需要 Erlang 26.x 以上否则服务起不来控制台根本无法访问。这些都是我在部署时实测踩过的。第二Windows 上装好 RabbitMQ 后建议不要直接用默认的 5672 端口很多项目可能被别的服务占用。修改端口时除了改 RabbitMQ 配置文件rabbitmq.conf里的listeners.tcp.default还要同时看 firewall 和 SpringAMQP 连接配置不然程序连不上。这也是搜索热词里“rabbitmq修改端口”为什么高频出现。第三本地学习时可以用 RabbitMQ 管理插件的 Web 页面直观观察队列、消息、死信情况。启动管理页的命令是rabbitmq-plugins enable rabbitmq_management访问地址是http://localhost:15672/默认账号密码 guest/guest。所有消息的 confirm、路由失败、死信转移都能在管理台看到强烈建议在生产验证时开着它观察。3.2 生产者端封装确认回调与失败落库接下来写一个比较完整但不过度设计的生产者服务。Service Slf4j public class ReliableRabbitPublisher { Autowired private RabbitTemplate rabbitTemplate; Autowired private JdbcTemplate jdbcTemplate; public void sendReliable(String exchange, String routingKey, Object payload) { String messageId UUID.randomUUID().toString(); MessageProperties props new MessageProperties(); props.setMessageId(messageId); props.setDeliveryMode(MessageDeliveryMode.PERSISTENT); props.setContentType(MessageProperties.CONTENT_TYPE_JSON); Message message MessageBuilder .withBody(JsonUtils.toJson(payload).getBytes(StandardCharsets.UTF_8)) .andProperties(props) .build(); CorrelationData correlationData new CorrelationData(messageId); // 在信元里携带业务自定义信息方便回调定位 correlationData.setReturned(new ReturnedMessage( message, replyCode, replyText, exchange, routingKey )); rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData); // 记录发送中状态到本地表 insertMessageRecord(messageId, exchange, routingKey, payload); } // 实际配置中需要在 RabbitTemplate 的 ConfirmCallback 和 ReturnsCallback 中调用业务标记方法 public void onConfirmSuccess(String messageId) { messageRecordService.markSuccess(messageId); } public void onFailAndStore(String messageId, RuntimeException ex) { messageRecordService.markFailAndRetryLater(messageId); } }关于correlationData.setReturned这部分在实际 AMQP 库里CorrelationData可以携带一个ReturnedMessage对象返回时往往会带上原消息。如果你使用的是旧版 Spring AMQP可能只有最简单的 get/set不重要关键是要搞清楚每次发送都新建一个CorrelationData不能复用否则多个消息的确认会错乱。在真实项目里我不会仅仅靠ConfirmCallback里写日志。因为一条消息从“发送成功”到“落库成功”中间还有网络延迟和 Broker 落盘耗时。最稳妥的做法是配合一张“可靠性消息表”发送前插入状态SENDINGconfirm 回调后改成SENT或FAILED再由定时任务扫表补偿未确认的消息。这个模式比任何重试框架都可靠。3.3 消费者端实现手动确认 幂等 死信转移消费者端可以从一个业务场景来拆假如我们在处理支付回调通知要求不丢不重。完整消费者代码Component Slf4j public class PaymentCallbackConsumer { Autowired private PaymentService paymentService; Autowired private IdempotentService idempotentService; RabbitListener(queues payment.callback.queue, ackMode MANUAL) public void handlePaymentCallback(PaymentMessage msg, Message message, Channel channel) throws IOException { long deliveryTag message.getMessageProperties().getDeliveryTag(); String messageId message.getMessageProperties().getMessageId(); try { // 幂等校验防止重复处理 boolean processed idempotentService.isProcessed(payment: msg.getOrderNo(), messageId); if (processed) { channel.basicAck(deliveryTag, false); return; } paymentService.handleCallback(msg); idempotentService.markProcessed(payment: msg.getOrderNo(), messageId); channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error(支付回调处理失败, messageId{}, messageId, e); // 这里选择 nack 且不重新入队让消息进入死信队列再由死信消费者人工排查 channel.basicNack(deliveryTag, false, false); } } }使用ackModeMANUAL时SpringAMQP 仍然允许你注入Channel和Message这是手动确认最常用的姿势。这里我把幂等校验放在业务处理前面同时把消息ID作为幂等控制的一部分。实际上如果你的业务没有天然唯一键一定要用RabbitMQ 消息的 messageId从业务语义上做到唯一消费。比如支付回调同一笔订单可能会因为 RabbitMQ 的 unacked 机制被重新投递或者生产者重试发送导致重复幂等是必须的。关于basicNack(deliveryTag, false, false)第三个参数requeue设 false 时消息会进入死信队列。但前提是这个队列一定配置了x-dead-letter-exchange否则消息会被直接丢弃。很多初学者在这里踩坑我配置了 dead letter exchange但消息一直不进去查了半天发现原队列上的死信参数没写。3.4 配置一个可靠的绑定关系图一个完整可靠的消息组件通常包括这些成员业务交换机exchange.order业务队列order.queue死信交换机exchange.order.dlx死信队列order.dlx.queue重试定时任务队列可选代码配置Configuration public class OrderRabbitConfig { public static final String EXCHANGE_ORDER exchange.order; public static final String QUEUE_ORDER order.queue; public static final String ROUTING_ORDER order.create; public static final String EXCHANGE_DLX exchange.order.dlx; public static final String QUEUE_DLX order.dlx.queue; public static final String ROUTING_DLX order.dlx; Bean public DirectExchange orderExchange() { return new DirectExchange(EXCHANGE_ORDER, true, false); } Bean public DirectExchange orderDlxExchange() { return new DirectExchange(EXCHANGE_DLX, true, false); } Bean public Queue orderQueue() { return QueueBuilder.durable(QUEUE_ORDER) .withArgument(x-dead-letter-exchange, EXCHANGE_DLX) .withArgument(x-dead-letter-routing-key, ROUTING_DLX) .build(); } Bean public Queue orderDlxQueue() { return QueueBuilder.durable(QUEUE_DLX).build(); } Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ROUTING_ORDER); } Bean public Binding orderDlxBinding() { return BindingBuilder.bind(orderDlxQueue()).to(orderDlxExchange()).with(ROUTING_DLX); } }这个绑定关系图并不复杂但它体现了完整的可靠性闭环消息发到业务交换机进入业务队列消费者消费失败后通过 nack(requeuefalse) 转移到死信队列。死信队列的消费者负责告警和落库人工处理。这套结构跑在线上三年多几千亿级消息没发生过核心业务丢失。3.5 模拟故障场景验证: 断网、宕机、监听停止我建议任何配置好可靠性方案的团队都要做故障演练。我一般会模拟四类故障生产者发送时 Broker 宕机看 ConfirmCallback 是否能收到 nack其实在 Broker 完全不可达时回调不会触发而是抛出连接异常这也要处理。路由键不存在看 ReturnsCallback 是否触发消息是否返回。消费者抛出异常且重试耗尽看消息是否进入死信队列。消费者手动 ack 前进程被杀重启后看消息是否重新投递是否被幂等挡住。第一类故障要在生产者侧加一个兜底。当连接异常抛出时rabbitTemplate 会同步抛AmqpConnectException这要求你在调用发送方法的业务代码中捕获异常并做本地落库。ConfirmCallback 只覆盖“消息已经到 Broker”的场景连接都建立不了时它不会触发。这又是一个容易被漏掉的点。第二类故障我用一个不存在的 routingKey 发一条测试消息管理台里可以看到Returned Message页签同时业务日志里打印 returns 回调。这能帮你区分“confirm 成功”和“真正路由成功”。第三类故障我在消费者代码里手动抛异常观察重试三次后消息进入order.dlx.queue。这是验证死信配置是否生效最直接的办法。第四类故障用 kill -9 杀掉消费者进程重启后 RabbitMQ 会把 unacked 消息重新入队前提是队列持久化且消息 persistent。此时消费者会再次收到同样消息如果幂等表没做好就会出现重复支付等问题。所以幂等必须和手动 ack 配套出现没有幂等的手动 ack 是不完整的。这就是为什么我在生产者、消费者两端都写“消息 ID 业务唯一键”双重幂等。特别要说明RabbitMQ 消息重投和生产者应用重试这两类重复是并存的光靠 RabbitMQ 的deliveryTag无法区分是不是同一条业务消息只有实现了整条链路的消息语义幂等才能真正实现“只看一次”。4. 常见问题与排查技巧实录4.1 生产者确认不回调或行为异常很多人配置了publisher-confirm-type: correlated但ConfirmCallback从来没被调用过。为什么第一种情况rabbitTemplate没有被 Spring 管理起来。比如你用new RabbitTemplate()手写了一个实例配置在某个Bean里返回的类型虽然是RabbitTemplate但自动配置创建的rabbitTemplate被你覆盖了。确保所有的RabbitTemplate都来源于同一ConnectionFactory否则回调不生效。第二种情况发送用的是convertAndSend但配置的messageConverter抛错了消息根本没发出去自然也就没有确认。这时业务代码没有及时捕获异常还以为发送成功了。第三种情况correlationData传入 null。SpringAMQP 在 confirm 模式下如果传入 null回调也能触发但拿不到关联数据无法知道确认的是哪一条消息。排查回调问题先看 RabbitMQ 管理台有没有消息进来再在回调处加一个断点看 messageId基本能快速定位。关于nack的排查如果ackfalse大部分原因是交换机不存在或权限问题。RabbitMQ 在投递到交换机时若发现 exchange 缺失会返回 nack 并在短语描述里写明原因。留意回调的cause参数里面通常包含reply-code和reply-text比如not_found就是交换机路由键不存在。4.2 消息无缘无故丢失的几种真实场景消息丢失是可靠性项目里最痛的问题。我列一个排查表都是真实项目里遇到的现象根本原因解决方案消息发送成功Broker 重启后消息没了交换器、队列未 durable或消息未设置 PERSISTENT全部声明 durabletrue发送时设置 deliveryModePERSISTENT消费者没有抛异常但消息不见了自动确认模式 方法内吞掉异常开启acknowledge-mode: manual消费者抛异常后消息不断重试导致队列消费阻塞重试未开启或 nack(requeuetrue)使用 Spring 重试并配置default-requeue-rejected: false或 nack 且 requeuefalse业务处理到一半进程崩溃重投后重复执行没有幂等用 messageId 或业务唯一键做幂等表处理前查重路由键写错confirm 成功但消息被丢弃mandatoryfalse 且未配置 returns设置 template.mandatorytrue实现 ReturnsCallback死信队列没接收到消息业务队列未配置死信参数或死信交换机没绑定检查x-dead-letter-exchange参数并确认死信交换器的绑定存在4.3 关于 RabbitMQ 启动失败与服务端排查这虽然不是可靠性核心但你会经常遇到因为服务都起不来就别谈可靠了。常见启动失败的原因我归类成四个Erlang 版本不匹配RabbitMQ 3.13 以后需要 Erlang 26RabbitMQ 4.1.x 需要 Erlang 26.2装错直接报init terminating in do_boot。Windows 上服务没有安装成功用管理员权限运行rabbitmq-service.bat install特别留意路径不能有中文或空格。浏览器无法访问管理台插件没启用执行rabbitmq-plugins enable rabbitmq_management。端口被占用默认 5672 和 15672修改配置后记得重启 Erlang 节点。在 Linux 上部署 RabbitMQ用 rpm 或 tar 包都行装完第一件事是rabbitmqctl status和rabbitmq-diagnostics ping确认节点和 Erlang 的 cookie 一致。很多 4.x 版本部署失败都是因为 cookie 权限或 hostname 解析不一致。这个经验如果你刚踩过绝对会印象深刻。4.4 消息重复消费终极解决思路最后说说重复消费。可靠性方案做到最后最能体现成熟度的就是幂等。重复消费的来源不止 Broker 的重投还有生产端在 confirm 失败后的定时补偿重发。如果不做幂等消息可靠发送反而会放大重复问题。幂等的设计原则第一层业务天然幂等。比如更新库存时的SET inventory inventory - ? WHERE inventory ?天然防止超卖多次执行不影响结果。第二层唯一索引。消息表或订单流水表用唯一键约束重复插入直接报错雠化。但要注意唯一约束冲突会抛异常要在 catch 里标记为已处理。第三层状态机幂等。支付回调只有当订单状态由PAYING变为PAID时才执行后续逻辑如果已经是PAID直接 ack 跳过。我的经验是消息可靠性中幂等的优先级要高于消息不丢失。因为一个丢失的消息可以通过日志和补偿找到而一个重复执行的副作用可能无法回滚。所以在设计任何 RabbitMQ 消费方案时都先问一句这条消息重复消费业务会不会炸不会那才是真可靠。4.5 关于性能与可靠性的调优心得配置了完整可靠性后你的消息吞吐量一定比默认模式低。这是公平的代价。如果业务需要高吞吐通常我会拆分链路核心链路全可靠非核心链路用默认快速模式。另外可以调整以下参数减少可靠性成本开启spring.rabbitmq.connection-mode多连接模式RabbitTemplate 默认是单连接共享高并发下可能造成通道竞争。使用publisher-confirm-typecorrelated而不是simple异步确认可以控制回调在批处理时不阻塞。批量发送、批量确认RabbitMQ Java Client 支持publisher confirms的批量等待适合低延迟要求不高的批量任务。消费者端设置prefetch合理控制每个消费者未确认消息数。prefetch10在 reliability 场景比较稳妥既能提高吞吐又不会导致大量消息堆积在消费者内存中。队列上设置x-max-priority或用 TTL 区分延迟消息让死信队列和补偿队列不至于和业务队列抢占消费线程。我见过一个项目把所有消费者 prefetch 设成 1导致消费吞吐量只有 10 条每秒线上堆积了几十万消息后来调到 20 并且配合手动确认吞吐上去了可靠性也没降。可靠性不是性能的敌人不加思考的配置才是。5. 一个完整可靠性方案的最终落地清单这节算是我个人项目的总结也方便你直接对照检查。生产者端publisher-confirm-typecorrelated实现 ConfirmCallbackpublisher-returnstrue且mandatorytrue实现 ReturnsCallback发送时设置PERSISTENT发送先落库回调后更新状态。交换器和队列所有核心组件durabletrue业务队列绑定死信交换器和死信路由键。消费者端acknowledge-modemanual开启 Spring 重试max-attempts3default-requeue-rejectedfalse业务处理前做幂等最终失败进入死信队列。死信消费者负责告警、存储原始消息等待人工处理绝不在死信消费者里没完没了重试。监控与告警监控死信队列的深度一旦死信堆积超过阈值立刻报警。这是可靠性的最后一道防线很多问题都要靠死信发现没有人会在正常队列里天天盯着日志看。我在实际项目里还加了一个“延迟补偿队列”把发送失败、路由失败的消息统一转入一个延迟队列延迟 10 分钟后自动重试重试超过 3 次进入死信由人工介入。这个设计让我在生产端几乎不需要手动干预即使短暂断网消息也能自动补偿成功。说一个小技巧在 RabbitTemplate 的回调中如果要做数据库操作千万不要在回调线程里启动长事务。ConfirmCallback 线程是 Netty 事件线程阻塞它会拖慢所有消息的确认处理。我的做法是回调里只发一个异步事件由独立线程池处理数据库状态更新。最后再分享一条经验配置可靠性只是一天的事真正考验人的是故障演练和监控完善。我看到太多团队消息可靠性文档写得花团锦簇线上出问题时死信队列里躺了一堆引用却没人处理。方案能跑通很容易能扛住故障才是真可靠。如果你现在正准备改造消息中间件建议先把完整方案在测试环境模拟消息丢失、重复投递、消费者宕机等场景验证一遍再上生产。这比任何配置文件都值钱。