1. 一条消息从生产到消费会死在四个位置1.1 先给消息的完整旅程画一条线先讲一个我踩过的真实坑。某次线上订单数据对不上排查了半天才发现是一条支付回调消息在 RabbitMQ 里凭空消失了。当时用的配置在很多人看来已经挺安全队列是持久化的消息发送也设置了持久化属性消费者接口也正常在消费。但最后还是丢了原因就出在持久化这三个字上——很多人以为声明了 durable 队列就等于消息不会丢实际上这是最常见的误解之一。要弄清楚 RabbitMQ 怎么防止数据丢失首先得回答的不是怎么防而是数据会在哪里丢。一条消息从业务系统里发出来到消费者真正处理完中间要走四段路第一段是从生产者进程到 Broker 的交换器第二段是交换器按照路由键把消息送进某个队列第三段是消息在队列里排队等消费者来取这段时间消息是存在 Broker 的内存和磁盘上的第四段是从队列投递到消费者进程并完成业务处理。每一段都可能出现静默失败而且默认配置下 RabbitMQ 的很多行为是宁可丢弃也不报错的。你得先接受一个残酷的现实RabbitMQ 本身不是一个默认不丢消息的中间件。它追求的是灵活和高性能可靠性需要使用者自己通过一系列配置去补齐。好消息是只要把每一段路的可靠性机制都打开消息丢失是可以被控制在极小概率的。这篇文章要讲的就是这个补齐的过程。1.2 一张表看清四个风险点把四个位置展开每个位置对应的丢失场景和解决方案可以整理成下面这张表环节消息丢在哪默认行为对应方案生产者 → 交换器网络抖动、连接断开、Broker 拒绝发送失败后生产者无感知开启 publisher confirm交换器 → 队列路由键写错、队列未绑定、交换器不存在消息被静默丢弃mandatory ReturnListener / 备用交换器Broker 存储Broker 宕机、内存换页、磁盘写满非持久化消息直接丢失durable 队列 持久化消息 水位保护队列 → 消费者消费者处理中崩溃、业务异常未捕获autoAck 模式下消息已确认手动 ack prefetch 重试/死信这四类风险里最容易被人忽略的是第二类和第四类。第二类风险在于发到了交换器但没有进队列你甚至可以在日志里看到发送成功实际消息却无影无踪第四类风险在于消费者把消息从队列里取出来了但业务还没做完进程就挂了这时候消息同样救不回来。后面所有配置本质上都是在跟这两个静默位置做对抗。2. 生产者这关开启 publisher confirm 是底线mandatory 是进阶2.1 confirm 机制的原理与代码生产者这一侧第一件要做的事就是开启发布确认。RabbitMQ 默认情况下只是把消息当作 UDP 一样发出去发完就完事了至于消息有没有到 Broker生产者完全不知道。这在可靠性要求高的场景里是不可接受的。开启 confirm 的方式非常简单在发送之前调用一次channel.confirmSelect()之后每条消息发给 BrokerBroker 处理完都会回调一个 ack失败则回调 nack。这里有个关键点对于设置了持久化的消息Broker 是在完成磁盘写入之后才会返回确认也就是说拿到 ack 基本可以认为消息已经安全落盘了。下面是原生 Java 客户端的写法ConnectionFactory factory new ConnectionFactory(); factory.setHost(192.168.1.10); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 开启发布确认 channel.confirmSelect(); String exchange order.exchange; String routingKey order.created; String message {\orderId\: 12345}; // PERSISTENT_TEXT_PLAIN 表示消息持久化deliveryMode 2 channel.basicPublish(exchange, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8)); // 同步等待确认超时时间 5 秒 if (channel.waitForConfirms(5000)) { System.out.println(Broker 已确认消息安全); } else { System.out.println(等待确认超时需要走重发逻辑); } }注意waitForConfirms是同步阻塞的写法适合消息量不太大的场景。如果发送量很大应该用异步监听channel.addConfirmListener((deliveryTag, multiple) - { // 收到 ack可以在这里更新数据库里的发送状态 System.out.println(消息确认成功, tag deliveryTag); }, (deliveryTag, multiple) - { // 收到 nack说明 Broker 写入失败需要重发 System.out.println(消息确认失败, tag deliveryTag); });2.2 mandatory 和 return listener路由不出去就退货第二个非常容易踩的坑是消息到达交换器、但路由不到任何队列。举个例子你声明了一个 direct 类型的交换器消费者绑定队列时用的路由键是order.paid但生产者发送时写成了order.pay那么这条消息就被 RabbitMQ 直接丢弃了而且不会报任何错。要避免这种静默丢弃可以在发送时把mandatory参数设为true。当消息路由不到任何队列时Broker 不会丢弃它而是通过 ReturnListener 把消息退还给你让你有机会做补偿处理channel.addReturnListener((replyCode, replyText, exchangeName, routingKey, properties, body) - { System.out.println(消息被退回, 路由键: routingKey , 内容: new String(body)); // 此处可以把消息标记为失败或者换一个路由键重新发送 }); // 第三个参数 true 就是 mandatory channel.basicPublish(exchange, routingKey, true, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));这段代码加上之后路由失败的消息至少不会死无对证。我在实际项目中见过太多排查很久才发现原来路由键就拼错了一个字母的案例mandatory是这类问题最直接的照妖镜。2.3 备用交换器连 return 都不想写时的兜底如果你觉得每条消息都写 return 监听比较啰嗦RabbitMQ 还提供了一层更优雅的兜底——备用交换器Alternate Exchange。思路很简单给交换器声明一个属性alternate-exchange凡是这个交换器路由不到队列的消息统一丢给备用交换器。备用交换器后面再绑定一个专门收集死消息的队列即可MapString, Object args new HashMap(); args.put(alternate-exchange, unrouted.exchange); // 声明主交换器路由失败的消息会进入备用交换器 channel.exchangeDeclare(order.exchange, direct, true, false, args); // 声明备用交换器类型通常用 fanout避免二次路由失败 channel.exchangeDeclare(unrouted.exchange, fanout, true, false, null); channel.queueDeclare(unrouted.queue, true, false, false, null); channel.queueBind(unrouted.queue, unrouted.exchange, );有了备用交换器哪怕某条消息真的发错路由键它也会落在unrouted.queue里你只需要写个消费者定时扫这个队列把消息捞出来分析原因、重新投递。这个方案比mandatory更省心的地方在于它不依赖生产者侧每一条消息都写 return 监听而是从交换器层面统一兜底适合多业务方共用一个接入层的场景。2.4 事务模式与 confirm 怎么选老一点的资料里还会提到 RabbitMQ 的事务机制发送前txSelect()发送后txCommit()事务提交失败就回滚。这确实能保证消息不丢但代价非常大。事务模式下每条消息都要等待 Broker 完成磁盘同步和事务提交吞吐量会急剧下降实测往往连 confirm 模式的十分之一都不到而且代码还要包一层事务边界复杂度也不低。我的建议很直接生产环境一律用 publisher confirm事务模式只适合那种发送量极小但绝对不容有失的边缘场景。官方自己也更推荐 confirm它的名字叫轻量级事务兼可靠性与性能。现在主流客户端都已经把waitForConfirms封装得很好了没有理由再走事务的老路。3. Broker 这关durable 不是免死金牌落盘才是3.1 队列、消息、交换器三层持久化要配齐很多人以为把队列声明成 durable 就够了但消息持久化其实有四个独立的点交换器要 durable、队列要 durable、消息本身要持久化、绑定关系也要能重建。任何一个环节缺失Broker 重启后都可能造成丢消息。先说队列。queueDeclare的第二个参数durable为true只是说队列的元数据包括绑定关系会持久化Broker 重启后队列还存在。但队列里已有的消息能不能恢复取决于消息本身有没有持久化。如果发送消息时没有指定deliveryMode 2消息还是存在内存里的机器重启就是一波清零。下面是正确的声明和发送组合// 交换器持久化 channel.exchangeDeclare(order.exchange, direct, true); // 队列持久化 MapString, Object queueArgs new HashMap(); channel.queueDeclare(order.queue, true, false, false, queueArgs); // 绑定关系 channel.queueBind(order.queue, order.exchange, order.created); // 发送时消息也持久化 channel.basicPublish(order.exchange, order.created, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));还有一个很隐蔽的坑生产环境里如果某个队列已经以非 durable 方式存在你再想用 durable 方式重新声明RabbitMQ 会直接报错inequivalent arg。因为队列属性一旦创建就不能原地修改只能删除重建。所以这些可靠性相关的属性一定要在项目初始化阶段就定好上线之后再改代价很大。3.2 消息从内存到磁盘中间发生了什么RabbitMQ 对持久化消息的处理并不是简单地把每条消息立刻 fsync 到队列文件里。真实流程是Broker 收到持久化消息后先把它写入一份预写日志transaction log后台再异步合并到队列索引中。fsync 也不是每条消息都触发一次而是按批次批量刷盘这就带来一个微小的确认时间窗口。理解这一点很重要publisher confirm 保证的是消息已经被 Broker 接收并完成持久化写入但持久化写入的具体粒度受磁盘刷盘策略和磁盘性能影响。如果你把disk_free_limit配得太激进或者机器本身 IO 性能很差确认的延迟会明显拉高甚至出现发送超时。所以可靠性配置不是越严格越好要在性能和安全之间找平衡。实操中我一般这样判断消息量不大、对时延不敏感的业务直接把每个环节都配到最严格消息量大的场景就要靠后面的批量发送和压测来确定刷盘间隔是否合理。3.3 内存水位和磁盘水位Broker 的自保机制很多人不知道RabbitMQ 自带一层保命机制内存和磁盘达到阈值后它会主动阻塞所有生产者连接防止进程 OOM 或者磁盘写爆。默认的内存水位是物理内存的 40%磁盘剩余空间阈值传统版本默认在 50MB 左右新版本也支持按磁盘总容量的百分比配置。这层机制平时不起眼但关键时刻能防止更严重的连锁丢失。试想一下如果磁盘满了而 RabbitMQ 还在继续接收消息落盘必然失败消息就真的找不回来了。有了水位阻塞Broker 会把生产者的连接置为 blocked你在客户端日志里能看到类似Connection was blocked的提示这时候就应该触发告警去清理磁盘了。运维层面要给这几个指标配上监控rabbitmq_queue_memory、disk_free、还有节点上阻塞连接数。很多莫名其妙丢消息的线上事故追到最后都跟磁盘写满导致的落盘失败有关。3.4 持久化带来的吞吐代价与应对最后说一个大家容易忽略的问题持久化 confirm 是有性能代价的。非持久化消息配合自动 ack单机吞吐可以轻松破万一旦全链路开启持久化和发布确认吞吐会明显下降因为每条消息都涉及磁盘写入和确认往返。应对方法有两个方向。一是批量发布在同一个 channel 里连续发多条消息再统一waitForConfirms把多次磁盘确认合并成一次实测吞吐能恢复不少二是评估业务对消息时延的敏感度比如日志类消息完全没必要持久化只有状态变化、资金相关的消息才需要严格的落盘保证。把不同可靠级别的消息分到不同交换器上比一刀切全部持久化要合理得多。4. 消费者这关手动 ack 加合理 prefetch别让消息死在最后一步4.1 autoAck 在什么情况下会丢消息很多刚接触 RabbitMQ 的人喜欢图省事消费的时候用autoAck true。这个参数的本意是Broker 把消息交给消费者就算确认完成但它在可靠性上有一个致命漏洞如果消费者收到消息后、业务逻辑还没执行完进程就崩溃或者被 kill 了这条消息就永久消失了Broker 不会重新投递因为按协议它已经被确认了。举一个我见过的真实场景某服务从队列里取消息后会调用一个外部接口更新缓存。因为外部接口偶尔超时服务重启时正好有一批消息处于已读取未处理完的状态。由于是 autoAck这批消息全部丢失。这类问题在代码层面完全看不出来只有数据对不上账的时候才会暴露。在 Spring AMQP 里要特别留意ackMode的三种取值ackMode对应行为可靠性NONE等价于原生 autoAck消息到达消费者就确认最危险AUTO监听方法正常返回才算确认抛异常则重新入队较可靠MANUAL开发者手动调用 basicAck / basicNack最可控4.2 先业务后 ack 的正确姿势手动 ack 的核心原则只有一句话先确保业务处理成功再告诉 Broker 可以删除这条消息。反过来先 ack 再处理业务跟 autoAck 就没有本质区别了。原生客户端的写法channel.basicQos(10); DeliverCallback deliverCallback (consumerTag, delivery) - { long deliveryTag delivery.getEnvelope().getDeliveryTag(); try { // 1. 先处理业务 handleOrder(delivery.getBody()); // 2. 业务成功确认消息 channel.basicAck(deliveryTag, false); } catch (Exception e) { // false 表示不批量拒绝true 表示重新入队 channel.basicNack(deliveryTag, false, true); } }; channel.basicConsume(order.queue, false, deliverCallback, consumerTag - {});Spring Boot 里的手动确认写法也类似只是要把ackMode显式声明出来RabbitListener(queues order.queue, ackMode MANUAL) public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); try { handleOrder(message); channel.basicAck(deliveryTag, false); } catch (Exception e) { channel.basicNack(deliveryTag, false, true); } }这里要说一个容易循环踩的坑basicNack最后一个参数requeue设为true消息会重新放回队列如果业务一直是坏的消息就会无限循环消费把日志刷爆。所以建议给队列配一个死信交换器把重试了一定次数仍然失败的消息扔进死信队列由人工或者补偿任务去处理。否则不丢是做到了但卡死也同样麻烦。4.3 prefetch 为什么重要确认机制还牵扯出一个资源问题如果一个消费者处理得慢而 Broker 又不断给它推送新消息消息会先在消费者进程的内存里积压。这个积压对 RabbitMQ 本身没有影响但会把消费者所在的业务服务内存吃满严重的时候业务进程 OOM反而把没处理完的消息连锅端了。basicQos(n)解决的就是这个水位问题。它表示这个消费者在收到 n 条未确认消息之前Broker 不会再推新的过来。对于处理耗时的业务我习惯把prefetch设置为 1 到几十之间具体值取决于单条消息的处理耗时。不要设置太大也不要完全依赖默认值——默认不设 limit 的时候Broker 会尽量塞满消费者。在 Spring Boot 里可以这样配spring: rabbitmq: listener: simple: prefetch: 10 acknowledge-mode: manual4.4 重复消费不是 bug幂等要靠自己手动 ack 加重新入队带来的一个必然结果就是至少一次投递。消息可能会在消费者处理完但还没来得及 ack 时因为连接断开而被 Broker 重新投递给另一个消费者实例。也就是说你的消费端代码必须做好幂等否则库存扣减、积分发放这类操作会重复执行。幂等方案按成本从低到高排列业务表里加唯一约束比如order_id唯一索引用 Redis 的SETNX做消费标记或者维护一张专门的消息消费记录表落的是全局消息 ID。我的习惯是每条消息在生产者那边生成一个全局唯一的messageId从发送到消费全程透传消费端先查记录、再执行业务、再记录结果这样即使重复投递也能安全跳过。5. 单机不丢只是入门集群挂点才是真正的考试5.1 镜像队列的复制逻辑与短板单机环境下把持久化、confirm、手动 ack 全配齐链路已经比较稳了但单机毕竟有宕机风险。早期 RabbitMQ 的集群方案是镜像队列一个主节点加多个镜像节点队列内容在这些节点之间全量复制。主节点挂了可以从镜像里提升一个新的主节点继续服务。镜像队列听着不错实际短板也很明显。它的复制是基于整条队列的同步期间性能会打折扣更麻烦的是网络分区时如果处理不当主节点和镜像节点之间可能出现脑裂造成消息不一致。官方后来对镜像队列的态度一直是能用但不够理想并在后续大版本里逐渐移除这套经典方案。现在新项目如果还在用镜像队列我建议直接用更现代的仲裁队列替代。5.2 仲裁队列用多数派协议换可靠性仲裁队列Quorum Queue是 RabbitMQ 3.8 引入、后续版本主推的高可靠队列类型。它的底层是 Raft 多数派协议队列有多个副本消息写入时只要超过半数的节点确认就算写入成功之后即使有部分节点宕机数据仍然完整。声明仲裁队列很简单只需要在队列参数里指定类型MapString, Object args new HashMap(); args.put(x-queue-type, quorum); channel.queueDeclare(order.queue, true, false, false, args);对比镜像队列仲裁队列有几个明显的可靠性优势一是发布确认要等多数副本落盘后才会返回确认的含金量更高二是节点故障恢复后会自动补齐落后的数据不容易出现脑裂导致的数据分叉三是它对消费者确认同样有日志复制机制消费进度本身也具备一致性。代价是延迟比普通队列略高但这对于不能丢消息的业务来说完全可以接受。5.3 网络分区与客户端重连确认丢失不等于消息丢失集群环境下还要理清一个概念客户端的确认丢失不等于消息本身丢失。比如生产者把消息发出去了Broker 也落盘并返回了确认但如果网络抖动导致确认回包丢了生产者的视角是这条消息不确定是否成功于是重发结果就是同样的消息进了队列两次。所以集群环境下的可靠性必然要配合幂等和去重来做。消费者端看到重复消息不必惊慌这是消息系统在故障恢复时的正常现象关键是链路里要有全局唯一 ID任何一次投递都能被识别成已处理过。我见过太多团队在单机测试时链路完好一上集群就开始出现重复消息最后都是靠幂等表兜住的。6. 故障演练三步验证你的链路是不是真的不丢6.1 演练一杀掉 Broker验证持久化恢复配置写得再漂亮不如实际崩一次看看。第一个演练最简单向某个队列发 1000 条持久化消息先不让任何消费者消费然后直接kill -9掉 RabbitMQ 进程等几秒再启动查一下队列里的消息数。如果还是 1000说明队列持久化和消息持久化都生效了如果少了回到第 3 节去查队列是不是 durable消息是不是deliveryMode 2有没有在声明阶段就埋了隐患。这里要提醒一句演练一定要在专用的测试环境做而且要把消费记录也一并清理干净否则你根本分不清消息是丢了还是被消费了。6.2 演练二断网模拟生产者发送失败第二个演练针对生产者侧。用 confirm 模式跑一个循环发送程序每发一条就记录一条然后在中途把网络断开或者直接把 RabbitMQ 停掉观察哪些消息收到了确认、哪些没有。凡是没收到确认的消息重发逻辑有没有生效重新连接之后数量能不能对上这个演练最能暴露的一个问题是很多人把waitForConfirms写成了等待确认超时就抛异常异常抛了但消息没有重发最后还是丢。正确的做法是超时或 nack 之后把消息标记为待重发由单独的补偿任务去处理。6.3 演练三消费者异常时的 ack 与重投策略第三个演练针对消费端。写一个消费者处理消息时故意抛异常然后分别测试三种情况requeuetrue时消息会不会回来、回来了多少次、日志有没有被打爆requeuefalse时消息会不会进死信队列死信队列里的消息能不能被人工脚本再捞回来重新投递。测完你就知道自己的重试策略到底靠不靠谱了。真实的线上故障往往不是消息丢了而是消息一直在一个死循环里转或者坏消息把正常消息堵住了。6.4 对账表才是最后防线最后的建议其实是所有环节之外的一层兜底对账。无论 RabbitMQ 的可靠性配得多完美我仍然建议在核心业务里维护两张表。一张是发送记录表字段至少包括全局消息 ID、业务主键、发送时间、confirm 状态另一张是消费记录表记录消息 ID、消费时间、处理结果。再起一个定时任务定期找出发送成功但超过 N 分钟没有消费记录的消息触发告警或者自动重发。这套对账方案看起来啰嗦但它解决的是配置层面所有手段都失效之后的最终一致性问题。传输链路再可靠也只是把丢失概率降得很低而不是降到零对账表能保证哪怕真的漏了一条也能在几分钟内发现并从业务层面补偿。我现在做消息相关项目第一件事永远是先设计对账任务再谈各种花式配置。最后分享一个我自己的习惯每次改完 RabbitMQ 的可靠性相关配置我都会强制自己跑一遍上面三个演练并把结果截图存档。因为配置文件里的一个参数看起来只是多了一个 true 或者 false实际影响却可能在半年后的一次宕机里才暴露出来。与其那时候在告警群里手忙脚乱不如现在就把崩一次当成例行公事。