一、为什么消息会丢失？

RabbitMQ 的消息从 Producer 到 Consumer，中间经过 4 个环节：

Producer → Broker（Exchange → Binding → Queue）→ Consumer → ACK

每个环节都有丢消息的风险：

环节	丢消息原因	发生概率
Producer → Broker	网络抖动 + Broker 未确认 + 消息未持久化	中
Broker 内部	消息在内存中，Broker 崩溃	低
Broker → Consumer	Consumer 拿到消息后未处理完就挂了	高
Consumer 挂掉	autoAck=true 时消息被标记删除	高

---

二、场景1：Producer 发消息，Broker 没收到

根因： Producer 发出去就认为成功了，但 Broker 根本没收到。

解决方案：持久化 + Publisher Confirms

// 1. 开启 Publisher Confirms（AMQP 0-9-1 协议的生产可靠性机制）
channel.confirmSelect();

// 2. 发消息时保证消息持久化
channel.basicPublish(
    exchange, 
    routingKey, 
    MessageProperties.PERSISTENT_TEXT_PLAIN,  // 消息持久化
    body.getBytes()
);

// 3. 等待 Broker 确认（同步方式，生产环境推荐异步回调）
boolean ack = channel.waitForConfirmsOrDie(5_000);
System.out.println("消息已确认到达 Broker");

解释：

• MessageProperties.PERSISTENT_TEXT_PLAIN = 消息持久化到磁盘
• waitForConfirmsOrDie(5000) = 5秒内等不到 Broker 的 ACK 就抛异常
• 金融跨行转账场景：这一步是必须的，不能省

更优方案：异步回调（生产级）

channel.confirmSelect();
ConcurrentNavigableMap<Long, String> outstanding = new ConcurrentSkipListMap<>();

channel.addConfirmListener(
    // ACK 回调：消息安全到达 Broker
    (sequenceNumber, multiple) -> {
        if (multiple) {
            outstanding.headMap(sequenceNumber, true).clear();
        } else {
            outstanding.remove(sequenceNumber);
        }
    },
    // NACK 回调：消息发送失败，需要重试
    (sequenceNumber, multiple) -> {
        String body = outstanding.remove(sequenceNumber);
        System.err.println("消息发送失败，需要重试: " + body);
        // 落库 + 延迟重投
    }
);

long seqNo = channel.getNextPublishSeqNo();
outstanding.put(seqNo, message);
channel.basicPublish(exchange, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

---

三、场景2：Broker 崩溃，内存消息丢失

根因： 消息还在内存里，没来得及刷盘，Broker 挂了。

解决方案：消息持久化 + 队列持久化 + 多副本

// 声明持久化队列
channel.queueDeclare(
    "my-queue", 
    true,       // durable = true：队列持久化
    false,      // exclusive = false
    false,      // autoDelete = false
    null
);

// 发消息时指定持久化
channel.basicPublish(
    "", 
    "my-queue", 
    MessageProperties.PERSISTENT_TEXT_PLAIN,  // 消息持久化
    body.getBytes()
);

关键点：必须同时满足以下条件，消息才能真正持久化：

1. Queue 声明时 durable=true
2. 消息属性设为 PERSISTENT
3. Broker 需要开启持久化（默认开启）

进阶：Quorum Queue（推荐生产环境使用）

普通 Queue 靠单节点内存存储；Quorum Queue 基于 Raft 协议，多节点复制，任意节点宕机不影响消息。

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

金融场景注意： Quorum Queue 是 2023 年后的新特性，有些银行科技部门的 RabbitMQ 版本较老不支持，需要提前确认 Broker 版本。

---

四、场景3：Consumer 拿到消息，处理一半就崩了

根因： autoAck=true 时，消息一发给 Consumer 就标记为已删除，Consumer 还没处理完就挂了，消息丢了。

解决方案：手动 ACK + 正确时机提交 ACK

// 1. 关闭自动 ACK
boolean autoAck = false;
channel.basicConsume("my-queue", autoAck, deliverCallback, cancelCallback);

// 2. prefetch 控制同时处理的消息数
channel.basicQos(1);  // 同时只处理 1 条，防止积压过多

// 3. 正确时机提交 ACK：业务处理完再确认
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    try {
        // 业务逻辑
        processMessage(message);
        // 业务处理成功，才 ACK
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 业务处理失败，拒绝消息（重新入队 or 进 DLQ）
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
    }
};

ACK 时机口诀： "业务处理完才 ACK，处理失败才 NACK"。

---

五、场景4：消息重复消费（幂等性）

根因： Consumer 处理成功，但 ACK 时网络抖动没发出去，Broker 重发了消息。

RabbitMQ 本身无法防止重复，需要消费者端做幂等设计。

解决方案：数据库唯一键 + Redis 去重

// 方案1：数据库唯一键（金融行业最常用）
public void processMessage(Message msg) {
    String msgId = msg.getMessageId();  // 消息唯一ID
    try {
        jdbcTemplate.update(
            "INSERT INTO processed_messages(msg_id, content) VALUES(?, ?)",
            msgId, msg.getContent()
        );
        // 业务处理
    } catch (DuplicateKeyException e) {
        // 消息已处理过，直接返回（幂等）
        log.info("消息 {} 已处理，跳过", msgId);
    }
}

// 方案2：Redis 去重（性能更高）
public void processMessageWithRedis(Message msg) {
    String msgId = msg.getMessageId();
    String key = "msg:processed:" + msgId;
    if (redis.setIfAbsent(key, "1", 24, TimeUnit.HOURS)) {
        // 首次处理，正常业务逻辑
    } else {
        // 已处理过
    }
}

金融行业场景： 跨行转账的流水号就是天然唯一键，直接用 流水号 做数据库唯一约束，任何重复消息都会被拦截。

---

六、场景5：Broker 队列积压爆满，新消息被丢弃

根因： 消费者挂了或处理太慢，队列积压满，新消息被 RabbitMQ 直接丢弃（默认行为）。

解决方案：

// 1. 队列加最大长度限制（超过就reject）
Map<String, Object> args = new HashMap<>();
args.put("x-max-length", 100000);  // 最多10万条
args.put("x-overflow", "reject-publish");  // 超出则拒绝新消息
channel.queueDeclare("my-queue", true, false, false, args);

// 2. 监控告警（超过80%就告警）
// rabbitmqctl set_policy limit_alert "my-queue" '{"max-length": 100000}'

// 组合使用：惰性队列 + 最大长度限制（金融行业推荐配置）
Map<String, Object> args = new HashMap<>();
args.put("x-queue-type", "lazy");
args.put("x-max-length", 500000);         // 最多50万条
args.put("x-overflow", "reject-publish"); // 超出拒绝新消息而不是覆盖旧消息
channel.queueDeclare("my-audit-queue", true, false, false, args);

---

七、场景6：TTL 队列消息过期后进了 DLQ，但 DLQ 也没处理

根因： 三档 TTL 设计时，消息进 DLQ 后没有消费者处理，永久积压。

解决方案：DLQ 消费者 + 死信监控

// DLQ 消费者：专门处理超时消息
@RabbitListener(queues = "dlq.my-queue")
public void handleDLQ(Message msg) {
    log.warn("DLQ 收到消息，可能业务处理超时，需要人工介入检查");
    // 重新入队 or 告警 + 落库
}

金融行业必做： DLQ 必须有监控，消息进 DLQ = 业务异常，不是正常流程，需要立即告警。

---

八、完整链路可靠性配置清单

配置项	配置值	说明
Queue durable	true	队列持久化
消息属性	PERSISTENT	消息持久化
Consumer autoAck	false	手动 ACK
prefetch	1（或业务合理值）	控制积压
Publisher Confirms	开启	生产端确认
Quorum Queue	推荐	多副本容错
DLQ 监控	必做	告警 + 落库

---

九、总结：一句话答案

当面试官问"消息丢失怎么办"，标准答案结构：

"消息丢失发生在 4 个环节：Producer→Broker、Broker 内部、Broker→Consumer、Consumer 处理中。对应的解法是：持久化 + Publisher Confirms（生产端）；Quorum Queue（Broker 端）；手动 ACK + 正确时机（消费端）；幂等设计（重复消费）。金融行业还必须加 DLQ 监控 + 告警，因为 DLQ 进消息 = 业务异常，不是正常流程。"