企业官网建设流程全解析

RabbitMQ 消息丢失，几乎是每个用过它的人都会头疼的问题。我见过不少团队，凌晨被线上告警叫起来，最后发现不是网络抖动，不是消费者代码 Bug，而是消息在某个环节被“默认行为”悄悄扔掉了——生产者以为发出去了，消费者也没收到，数据就这么消失了。RabbitMQ 本身对可靠性是有完整机制的，问题在于这些机制默认不是全开的，你得在正确的位置做正确的配置。

这篇文章适合正在维护 RabbitMQ 生产集群、或者项目里准备把 RabbitMQ 作为核心消息中间件的同学。你可以把它当成一套完整的防丢检查清单：消息从生产者发出，到交换机路由，到队列存储，再到消费者确认，每个环节到底丢在哪、怎么防、怎么排查，我会逐个拆开讲。内容偏实战，不会堆概念，我会尽量把底层原理和踩坑细节都说清楚。

1. 先把“消息去哪了”这件事彻底理清

1.1 一条消息从发到收，要经过哪些关卡

要防止数据丢失，首先得知道一条消息要过几道门。RabbitMQ 的模型不算复杂：生产者把消息发给交换机（Exchange），交换机按照绑定关系（Binding）和路由键（Routing Key）把消息投递到队列（Queue），消费者再从队列里取消息处理。这中间其实只有三段链路：

第一段是生产端到服务端。消息从你的应用进程出发，经 TCP 连接到达 RabbitMQ 节点。这一段最容易丢，因为网络是异步的，你调用发送接口只是把数据交给了本地套接字，服务器到底收没收到，默认情况下你是不知道的。

第二段是服务端内部的路由与存储。消息到了交换机，如果路由不到任何队列，默认会被直接丢弃；就算进了队列，如果队列和消息都没做持久化，节点一重启，积压的数据就全部清零；磁盘满了、内存达到阈值被阻塞，也有一系列不可控行为。

第三段是消费端确认。消费者拉取消息后，如果用了自动 ACK，那无论业务处理成功还是抛异常，消息都会被标记为已消费。等你想重试的时候，消息已经不在队列里了。

所以“数据丢失”其实从来没有单一原因，绝大多数是多环节叠加出来的。后面所有配置和方案，本质都是把这三段路上的默认宽松策略收紧。

1.2 三类典型的丢消息场景

我习惯把丢消息场景分成三类，排查的时候思路会非常清晰：

这里有个经验：很多人以为 RabbitMQ 是“消息中间件就肯定不会丢”，这个观念要不得。RabbitMQ 的默认策略是追求性能和灵活性，可靠性需要你主动配置。接下来每一节，我都会按生产端、服务端、消费端的顺序展开，最后再给一套可以直接抄的配置清单。

2. 生产者端：消息没进队列，一切都无从谈起

2.1 先确认：你发出去的每一条消息，真的有回执吗

生产端最大的坑，就是以为调用发送 API 返回了就是成功了。 RabbitMQ 的普通发送是 fire-and-forget 模式，你的客户端把消息写进 TCP 缓冲区后，方法就返回了。如果发送过程中连接断开、Broker 拒绝、网络闪断，这条消息就无声无息地消失了。

要解决这个问题，必须开启Publisher Confirm 机制。开启后，Broker 每收到一条消息，都会给生产者返回一个确认（Basic.Ack）或者失败（Basic.Nack）。在 Spring Boot 里设置如下：

spring: rabbitmq: publisher-confirm-type: correlated publisher-returns: true

correlated表示每条消息都会有一个 correlationData 与确认结果对应。然后你需要在发送时传入 correlationData，并在回调里处理确认结果：

rabbitTemplate.convertAndSend(exchange, routingKey, message, new CorrelationData(id));

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { if (!ack) { // 记录日志，走补偿或告警 log.error("消息发送失败，id={}, cause={}", correlationData.getId(), cause); } });

注意，ack=true只能说明消息被 Broker 接收了，不代表已经持久化到磁盘。但至少这一层能防住网络层面的丢失。还有waitForConfirms这种同步等待方式，性能较差，高吞吐场景不建议，但小流量内部系统可以用。

提示：开启 confirm 之后，一定要给发送失败留一条补偿路径。常见做法是：确认失败的 ID 落库，定时任务扫描后重发；如果大数据量场景不想落库，至少要打到日志中心并接告警。

2.2 路由失败怎么办：mandatory 与备份交换机

另一个容易忽略的是路由失败。消息到了交换机，如果没有任何队列匹配它的路由键，默认情况下消息会被直接丢弃，而且生产者完全不知道。你发成功了吗？从 TCP 层面看确实发出去了，但队列里根本没有这条消息。

这时候需要给发送加mandatory标志。mandatory 为 true 时，如果消息无法路由，Broker 会通过 Basic.Return 把消息退回给生产者。在 Spring Boot 里publisher-returns: true开启后，再设置：

rabbitTemplate.setReturnsCallback(returned -> { log.error("消息路由失败，exchange={}, routingKey={}, replyText={}", returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()); });

但靠回调处理 return 还是略显被动，更稳妥的方案是配置备份交换机（Alternate Exchange）。给主交换机声明参数alternate-exchange，所有无法路由的消息会自动进入备份交换机，你再让备份交换机绑定一个死信队列，这样消息就不会凭空消失，而是进入“待人工处理”的队列：

Map<String, Object> args = new HashMap<>(); args.put("alternate-exchange", "exchange.ae"); channel.exchangeDeclare("exchange.main", "direct", true, false, args);

2.3 为什么我不推荐用事务来保证发送

RabbitMQ 支持 AMQP 事务机制，txSelect()开启事务，txCommit()提交。事务模式下，消息确认变成同步阻塞操作，吞吐量会下降几个数量级。我自己只在极端的对账类场景里用过事务，常规业务我更推荐 confirm 模式，因为 confirm 是异步的，性能和可靠性之间的平衡点更好。

另外注意，事务和 Confirm 不能同时用。如果你看到网上有些老代码又txSelect()又confirmSelect()，那多半是互相冲突的，发消息时会直接报错。

3. Broker 端：队列没扛住，消息照样丢

3.1 持久化三件套：交换机、队列、消息

消息进了队列，你以为就安全了？不一定。RabbitMQ 的持久化要三个东西同时满足，缺一个都会在节点重启时丢数据：

• 交换机要声明为 durable。交换机丢了，绑定关系就全没了，消息来了也投不出去。
• 队列要声明为 durable。队列不是持久化的，节点重启后队列直接被删掉，里面的消息不管是不是持久化的，全部蒸发。
• 消息本身要设置 deliveryMode=2，也就是持久化消息。

在 Spring Boot 里，声明队列时可以这样写：

@Bean public Queue orderQueue() { return QueueBuilder.durable("order.queue").build(); }

原生客户端发送消息时，设置消息属性：

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .deliveryMode(2) .build(); channel.basicPublish(exchange, routingKey, true, props, body);

这里有个容易被忽视的细节：消息写入队列后，RabbitMQ 是先在内存里缓冲，再异步刷盘的。就算你设置了 deliveryMode=2，在极端断电场景下，仍然可能丢失极少量刚刚写入但尚未落盘的消息。要缩小这个窗口，可以调整queue_explicit_acks、msg_store_sync_interval等参数，但会影响性能。大多数业务能容忍这个毫秒级窗口，不能容忍的话需要做对账补偿，而不是纯依赖中间件。

3.2 单节点扛不住：镜像队列与仲裁队列怎么选

单节点的 RabbitMQ 就算持久化全做了，物理机宕机或者磁盘损坏，数据照样会丢。所以要上集群，但集群本身也有讲究。

早期的方案是镜像队列（Mirrored Queue），通过ha-mode策略把队列复制到多个节点。镜像队列有个隐患：它本质上是主从异步复制，主节点宕机时，如果从节点还没同步完最新消息，就可能丢。另外它和网络分区相处得不好，脑裂时会有分队各持一部分数据的问题。

我现在的建议是，新项目直接用仲裁队列（Quorum Queue）。仲裁队列基于 Raft 协议实现，写入需要多数节点确认才算成功，天然保证多副本一致性，比镜像队列可靠得多。声明方式是在队列参数里加x-queue-type=quorum：

Map<String, Object> args = new HashMap<>(); args.put("x-queue-type", "quorum"); channel.queueDeclare("queue.quorum", true, false, false, args);

仲裁队列虽然在吞吐性能上比普通队列略低，但它换来的是一致性保障。对核心业务跟交易相关的消息，强烈建议直接上仲裁队列。如果用的是旧版本 RabbitMQ，也不要用镜像队列硬扛核心链路，至少把重要消息放在“镜像数量 = 全部节点”的策略下，降低同步延迟带来的丢失风险。

注意：镜像队列在 RabbitMQ 3.8 以后已经进入维护模式，官方主推仲裁队列和流队列。还在用镜像队列的老集群，迟早要迁移。

3.3 磁盘和内存水位，被 Block 时消息去哪了

RabbitMQ 有内存和磁盘保护机制。当内存使用超过vm_memory_high_watermark（默认 0.4，即节点内存的 40%），或者磁盘可用空间低于disk_free_limit（默认 50MB 或磁盘的 10% 中的一个），Broker 会进入 Block 状态，阻塞所有生产者的连接。

很多人以为 Block 只是暂停发送，问题不大。但实际上，如果 Block 持续太久，一些非持久化消息可能因为内存压力被换页到磁盘，严重时连接会被直接关闭。更麻烦的是，如果你的节点在磁盘告警状态下又收到大量写入，队列可能拒绝接受消息，对客户端表现为发送超时。消息有没有丢？丢失不一定，但发送失败是必然的。

这几个参数要提前规划：

# rabbitmq.conf vm_memory_high_watermark.relative = 0.6 disk_free_limit.absolute = 2GB

我通常会配合监控工具盯住rabbitmq_queue_consumers、rabbitmq_node_mem_used等指标，在到 80% 阈值之前就告警。不要等红色告警了才去处理。

3.4 队列长度与 TTL，溢出丢消息是“静默的”

RabbitMQ 队列可以设置最大长度x-max-length和溢出策略x-overflow。默认overflow=drop-head，意思是队列满了以后，新消息进来，队头的旧消息会被直接丢弃。如果你只设置了长度而没设置拒绝发布，那么看起来队列一直处于“满”的状态，实际上老消息正在被悄悄扔掉。

更隐蔽的是 TTL。消息设置了 TTL，过期后如果队列没有绑定死信交换机，过期消息会被直接删除。这个问题我踩过一次：给消息设了 30 分钟过期，结果某个消费者故障了 40 分钟，等恢复后所有积压消息已经全部过期被清空，业务对不上账才发现。

正确做法是：重要消息不设 TTL，或者设 TTL 的同时绑定死信队列，让过期消息有归宿。队列长度也一样，溢出策略要明确的变成“拒绝发布”（overflow=reject-publish）并配备死信，而不是默默丢队头。

4. 消费者端：ACK 用不对，业务等于没做

4.1 手动 ACK 是底线，prefetch 是调优

消费者丢消息，最常见的原因就是自动 ACK。自动 ACK 模式下，RabbitMQ 把消息推给消费者后就立刻从队列里删除，你的业务代码还没开始执行，消息已经标记为完成。如果处理中抛异常，这段业务数据就彻底丢了。

所以重要业务必须用手动 ACK。在 Spring Boot 里设置：

spring: rabbitmq: listener: simple: acknowledge-mode: manual prefetch: 10

消费端代码里，处理成功后调用basicAck，失败时调用basicNack：

@RabbitListener(queues = "order.queue") public void onMessage(Channel channel, Message message, @Header(AmqpHeaders.DELIVERY_TAG) long tag) { try { // 业务处理 channel.basicAck(tag, false); } catch (Exception e) { // 第三个参数 false 表示不重新入队，进入死信 channel.basicNack(tag, false, false); } }

这里prefetch也值得说。prefetch 决定消费者消费端能预取多少未确认消息。设得太高，比如默认的 250，一个消费者实例崩溃时，会有大量 unacked 消息需要重新投递，虽然不会丢，但会让其他消费者瞬间压力暴增。设置为 10~50 之间比较稳妥，既能保证吞吐，又不会造成大规模的重新排队。

经验：basicNack的第三个参数requeue要慎用。如果业务处理失败是因为代码 Bug，你 requeue 回队列，消息会立刻再次投递给同一个消费者，形成一个无限循环，把 CPU 打满不说，日志会被刷爆。只有明确是“临时性失败”才考虑 requeue，比如依赖的下游服务刚好重启。

4.2 消费失败别无限重试：死信与延迟重试

手动 ACK 挡住了一部分问题，但消费者处理失败的场景依然存在。多数框架默认会对失败消息进行本地重试，比如 Spring Boot 的@RabbitListener在抛出异常时，消息会回到队列头部，然后再次投递。如果业务代码永远是同样的异常，这个循环会一直转下去，消息永远不会被放弃，也不会进入其他兜底机制。

我的做法是三层策略：

1. 本地有限重试。Spring Boot 里配置spring.rabbitmq.listener.simple.retry.enabled=true，设置max-attempts=3，initial-interval=1s，倍增系数 2。本地重试只解决短暂的网络抖动。
2. 超过次数后进入死信队列。给业务队列设置x-dead-letter-exchange和x-dead-letter-routing-key，消息被重试失败后自动投递到专门的处理队列。
3. 对死信队列做告警和人工介入。死信队列里的消息代表“系统无法自动处理”，要有专门的任务扫描，或者至少夜深人静时能收到告警。

声明死信队列：

Map<String, Object> args = new HashMap<>(); args.put("x-dead-letter-exchange", "exchange.dlx"); args.put("x-dead-letter-routing-key", "dlx.routing"); channel.queueDeclare("queue.business", true, false, false, args);

4.3 幂等处理是最后一道保险

就算你把生产端、服务端、消费端全部配置到位，你依然要面对一个现实：RabbitMQ 的投递语义是at least once，也就是至少一次。为什么？因为生产者确认超时后重发，消费者处理成功但 ACK 在返回途中丢失，Broker 会再次投递。这两条路都会产生重复消息。

幂等不是 RabbitMQ 能帮你解决的，必须业务侧自己扛。常见的方案是：

• 用业务唯一键建唯一索引，重复插入直接冲突，不报错；
• 或者维护一张“已处理消息表”，消费前先查一次，处理完插入一行；
• 更简单粗暴的，把消息里的 eventId 去重。

我之前遇到过一个用户重复下单的问题，根源就是消费端没做幂等，一条下单指令被投递了两次，订单表里插了两条记录。RabbitMQ 配置再怎么调，也救不了幂等缺失。

5. 一套可落地的全链路防丢配置参考

5.1 Spring Boot 下的关键配置

如果你用的是 Spring Boot + RabbitMQ，这里是一份我偏好的最小可靠配置：

spring: rabbitmq: host: your-host port: 5672 username: your-user password: your-pass publisher-confirm-type: correlated publisher-returns: true listener: simple: acknowledge-mode: manual prefetch: 10 retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 2.0

然后再补充几个 Java 侧的要点：

• 发送消息时用CorrelationData带唯一 ID，并将 ID 缓存在本地，收到确认回调后清理；
• 设置ReturnsCallback，路由失败时告警；
• 核心队列全部通过QueueBuilder.durable()创建，并指定死信交换机；
• 消费者监听方法第一行就做幂等检查。

这套组合跑下来，常规业务不会丢消息。

5.2 原生客户端的发送与消费示例

不依赖 Spring Boot 的同学，原生客户端可以这样写：

ConnectionFactory factory = new ConnectionFactory(); factory.setHost("your-host"); Connection conn = factory.newConnection(); Channel channel = conn.createChannel(); // 开启确认 channel.confirmSelect(); // 注册 Return 监听 channel.addReturnListener((replyCode, replyText, exchange, routingKey, props, body) -> { // 消息路由失败，进入补偿 }); AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .deliveryMode(2) .build(); channel.basicPublish("exchange.name", "routing.key", true, props, "message".getBytes()); if (!channel.waitForConfirms(5000)) { // 发送失败，补偿 }

消费端手动 ACK：

channel.basicQos(10); channel.basicConsume("queue.name", false, (consumerTag, delivery) -> { try { // 业务处理 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (Exception e) { channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); } }, consumerTag -> {});

注意basicPublish的第三个参数mandatory=true要和 Return 监听配套使用，否则路由失败还是静默丢弃。

5.3 上线前的自检清单

每次新服务上线前，我都会拿这张表过一遍，配合命令行工具检查：

我自己在正式环境上线前，还会故意做一次“节点 kill -9 测试”，直接杀掉一个节点进程，观察队列恢复后消息有没有变少。只有经过这种破坏性测试，我才敢说这个链路的数据是安全的。

6. 实战中常见的丢失场景与排查记录

6.1 三个印象深刻的排查案例

先分享一个生产端案例。某天业务方反馈某个统计任务隔三差五少数据，查发送日志发现全部返回成功，但数据库里就是缺记录。排查了很久才想到是 RabbitMQ 连接在发送请求发出后、确认返回前断开了，那条消息根本没到 Broker。开启 Publisher Confirm 之后，把所有发送失败的消息统一补偿，问题才真正消失。

再说消费端。有个同事负责的异步任务总是“丢单”，我让他把acknowledge-mode改成 manual 后，发现其实是消费者在反序列化 JSON 时抛异常了，但自动 ACK 已经把消息吃掉。改成手动 ACK + 死信队列后，异常消息全部进死信，一条没丢，系统也从“莫名其妙少数据”变成了“所有异常可见”。

最后是 Broker 端的。某团队反馈节点重启后，一个核心队列里的积压消息全没了。我上去rabbitmqctl list_queues name durable messages一看，队列 durable 是 false，消息持久化属性也没有设置。典型的“开发环境当作生产环境用”的问题。这类配置错误，靠监控根本发现不了，只能在代码审查环节就拦住。

6.2 常见问题速查表

最后说一个我自己的习惯：每逢大版本升级或者集群迁移，我一定会在验证环境跑一遍完整的消息链路演练，把生产者的发送失败率、消费者 ack 成功数、最终落库的记录数三方对一遍。RabbitMQ 的可靠性从来不是靠某一个配置点撑起来的，而是每一层都留了兜底。你要做的，就是把每一层的兜底都接上，然后确保异常路径有人看见、有人处理。