RocketMQ如何保证消息不丢失

线上突然报警：消息堆积，排查后发现是某Broker宕机，部分消息神秘“消失”……
如果你不想经历这种惊魂时刻，那么理解RocketMQ如何保证消息不丢失，就是你的必修课。

消息可靠性是消息队列的立身之本。尤其在金融、交易等核心业务场景，丢一条消息可能就意味着一次严重的事故。今天，我们就来深入剖析RocketMQ的可靠性保障机制

可以把RocketMQ想象成一个快递公司，保证消息不丢，就是要确保：

1. 快递员收件后，包裹不会在运输中丢失（生产者 -> Broker）
2. 包裹在仓库里不会因为失火、停电而损毁（Broker自身）
3. 收件人必须成功签收，否则包裹要退回或重送（Broker -> 消费者）

围绕这三点，RocketMQ构建了一套完整的防御体系。

---

第一道防线：Broker的持久化机制（保护“仓库中的包裹”）

风险：Broker（消息存储服务器）突然宕机，内存中还未写入磁盘的消息会全部丢失。

解决方案：消息刷盘

这就像快递公司收到包裹后，是随便扔在临时工棚（内存）里，还是立即锁进坚固的保险库（磁盘）。RocketMQ提供了两种“入库”策略：

1. 同步刷盘 (SYNC_FLUSH) - 最安全

• 工作方式：每收到一个消息，必须立即成功写入磁盘后，才向生产者返回成功响应。
• 优点：绝对可靠，即使机器突然断电，消息也已安全落盘。
• 缺点：性能有损耗，因为每次写入都有一次磁盘IO的等待时间。
• 配置：在Broker的配置文件中设置 flushDiskType = SYNC_FLUSH

2. 异步刷盘 (ASYNC_FLUSH) - 高性能（默认）

• 工作方式：收到消息后，先存入内存就立即返回成功响应，然后由后台线程定期批量将内存中的消息刷到磁盘。
• 优点：吞吐量极高，性能好。
• 缺点：如果Broker突然宕机，内存中尚未刷盘的消息就会丢失。
• 配置：flushDiskType = ASYNC_FLUSH

生产环境建议：

• 对消息可靠性要求极高的场景（如资金变动），使用同步刷盘。
• 对性能要求更高、允许毫秒级消息丢失的场景（如日志收集），使用异步刷盘。

---

第二道防线：主从复制机制（建设“异地备援仓库”）

风险：即使采用了同步刷盘，如果整个Broker服务器所在的机器磁盘损坏了，消息还是会全部丢失。

解决方案：主从架构 (Master-Slave)

这就像不光有一个主仓库，还在另一个地方建立了一个备援仓库。主仓库每收到一个包裹，都会实时或准实时地同步给备援仓库。

1. 同步复制 (SYNC_MASTER) - 高可靠

• 工作方式：主节点（Master）收到消息后，必须立即将其成功复制到至少一个从节点（Slave）后，才向生产者返回成功响应。
• 优点：数据有实时备份，主节点宕机，从节点顶上，数据不丢。
• 缺点：增加了写入延迟。
• 配置：在Broker的配置文件中设置 brokerRole = SYNC_MASTER

2. 异步复制 (ASYNC_MASTER) - 高性能

• 工作方式：主节点收到消息后立即返回成功，然后异步地将消息复制给从节点。
• 优点：性能好，延迟低。
• 缺点：如果主节点在消息复制完成前宕机，可能会丢失少量消息。
• 配置：brokerRole = ASYNC_MASTER

生产环境建议：

• 通常将 同步刷盘 + 同步复制 组合，作为金融级的高可靠性配置（代价是性能最低）。
• 将 异步刷盘 + 异步复制 组合，作为高性能配置（代价是可靠性最低）。

---

第三道防线：生产者确认机制（确保“包裹成功入库”）

风险：生产者把消息发出去了，但网络闪断或Broker瞬时故障，导致消息根本没到达Broker。

解决方案：发送状态确认

这就像快递员收件后，必须给你一张盖了章的“收货回执”，你才放心。RocketMQ的生产者发送消息后，会收到一个SendResult。

• 代码示例：

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.start();try {Message msg = new Message("OrderTopic", "TagA", "OrderID001", "Hello World".getBytes());// 发送消息，并获取结果SendResult sendResult = producer.send(msg);// 根据发送结果判断是否成功if (sendResult.getSendStatus() == SendStatus.SEND_OK) {System.out.println("消息发送成功，已确认送达Broker！");} else {System.out.println("消息发送失败，需要进行重试！");// 这里应加入重试逻辑}
} catch (Exception e) {e.printStackTrace();// 发生异常，同样需要进行重试producer.send(msg, 3); // 例如，自动重试3次
}

• 最佳实践：务必在发送消息后检查SendStatus，并实现重试机制（setRetryTimesWhenSendFailed），确保在网络波动等情况下消息能最终送达。

---

第四道防线：消费者确认与重试机制（确保“包裹成功签收”）

风险：消费者拉取到消息，业务逻辑还没执行完（比如还没扣款），程序就崩溃了。此时Broker如果认为消息已消费而删除它，就会导致消息丢失且业务未执行。

解决方案：消费确认 (ACK) 与重试

这就像收件人必须当面拆包验货，确认商品完好无损后再签字。如果验货失败或者收件人不在家，快递员就需要改天再送（重投）。

• 工作方式：

1. 1. 消费者拉取消息。
   2. 执行业务逻辑（如操作数据库）。
   3. 业务成功后，返回 CONSUME_SUCCESS 告知Broker“可以删除消息了”。
   4. 如果业务失败或抛出异常，返回 RECONSUME_LATER，Broker会稍后重新投递这条消息（重试机制）。

• 代码示例：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
consumer.subscribe("OrderTopic", "*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {try {// 1. 这里是你的核心业务逻辑String orderId = processMessage(new String(msg.getBody()));System.out.println("处理订单成功: " + orderId);// 2. 业务成功，确认消费return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;} catch (BusinessException e) {// 3. 业务逻辑处理失败，稍后重试System.out.println("处理失败，等待重试");return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});
consumer.start();

• 重试策略：RocketMQ自带重试机制，消息消费失败后，会在延迟一段时间后重新投递，默认最多重试16次。

---

总结与面试实战

“为了保证RocketMQ消息不丢失，我们构建了一个从Broker到客户端的全方位保障体系：

1. 在Broker层面，通过持久化机制（刷盘） 来保证单机可靠性，通过主从复制机制来保证集群的高可用。生产上可以根据业务在性能和可靠性之间做权衡，比如选择同步刷盘还是异步刷盘。
2. 在生产端，我们依赖发送状态确认和内部重试机制，确保消息一定能成功送达Broker。
3. 在消费端，最关键的是采用手动ACK确认机制，只有在业务逻辑真正执行成功后，才告知Broker消息已消费。如果处理失败，Broker会通过重试机制重新投递，直到消费成功。

通过这套 ‘刷盘 + 主从 + 确认 + 重试’ 的组合拳，我们就能在99.99%的场景下保证消息的零丢失。”

一张图帮你理解全流程：

生产者 --(发送确认&重试)--> Broker主节点 --(同步/异步复制)--> Broker从节点|(同步/异步刷盘)|V磁盘|
消费者 <--(ACK确认&重试机制)-- Broker主节点