RocketMQ消息丢失场景及全面解决方案

RocketMQ 作为一款高性能、分布式的消息中间件，被广泛应用于电商、金融、物流、在线游戏等对高可用、高性能、强一致性要求较高的场景。然而，在实际生产环境中，消息丢失问题仍时有发生，影响系统的可靠性与数据一致性。本文将从常见消息丢失场景、原因分析、全面解决方案等方面入手，通过图解流程与代码示例，帮助你彻底理解并解决 RocketMQ 的消息丢失问题。

一、前言

在分布式系统中，消息队列承担着“解耦”“异步解耦”“流量削峰”等重要角色。消息一旦丢失，可能会导致订单丢失、库存扣减不一致、用户通知漏发等严重业务问题。因此，对于 RocketMQ 这样的企业级中间件来说，确保消息可靠投递与消费至关重要。本文重点剖析以下内容：

1. 常见的消息丢失场景：生产者端、Broker 端、消费者端、事务消息、延迟消息等多种原因导致的消息丢失。
2. 原因详细分析：从网络、磁盘、并发、代码逻辑等角度剖析根本原因。
3. 全面解决方案：针对不同场景给出从生产端到消费端、配置、监控、运维等全链路的优化措施，并提供 Java 代码示例和 Mermaid 流程图。

二、常见消息丢失场景

下面罗列了在实际生产中最容易遇到的几种 RocketMQ 消息丢失场景：

1. 生产者端发送失败未重试

   • 场景：生产者发起消息发送时，因网络抖动、Broker 不可用等导致发送返回超时或失败；如果开发者没有开启重试或未捕获发送异常，消息可能直接丢失。

2. Broker 存储异常或宕机，Message 尚未持久化

   • 场景：Broker 接收到消息并返回发送成功，随后在刷盘之前发生宕机，导致消息未写入磁盘；如果使用异步刷盘且刷盘回调未生效，重启后该消息就会丢失。

3. 消费端处理异常造成偏移量（offset）提前提交

   • 场景：消费者收到消息后，在处理业务逻辑（如写数据库）过程中出现异常，导致消费失败；如果消费框架采用自动提交 offset 的方式，且提交时机在业务处理之前，Broker 会认为该消息已经消费，后续消费者将跳过该条消息，造成消息“丢失”。

4. 消息重复消费后丢弃导致数据不一致感知为丢失

   • 场景：消费者做幂等性保护不当，对重复消息进行了静默丢弃。虽然消息实际上到达过消费端，但因业务判断为“已消费”，不会再次处理，导致某些数据未恢复预期结果，表现为“消息丢失”。

5. 事务消息半消息回查超时导致丢失

   • 场景：事务消息发送后，Producer 端本地事务未及时提交或回滚，导致 Broker 长时间等待回查；如果超出指定回查次数且条件判断不当，造成最终该半消息被丢弃。

6. 延迟消息/定时消息由于 Broker 配置或消费逻辑错误失效

   • 场景：配置了延迟级别的消息，但 Broker 与 Consumer 未正确识别延迟队列导致过期消息提前投递，或 Consumer 端过滤条件错误将其直接舍弃。

7. Broker Master-Slave 同步延迟，消费者从 Slave 同步延迟敏感场景下读取旧数据

   • 场景：开启了半同步刷盘模式，若 Master 刚收到消息还未同步到 Slave，消费者恰好从 Slave 拉取，可能读不到最新消息，表现为“丢失”。

8. 消费端负载均衡瞬间抖动，Topic/Queue 重平衡导致少量消息跳过

   • 场景：当消费者组实例数量调整时（增减实例），Broker 会重新分配 Queue。若消费者在 Rebalance 过程中提交 Offset 有误或拉取不到新分配的队列，可能会错过部分消息。

三、原因分析

针对以上场景，我们逐一拆解根本原因：

3.1 生产者发送层面

1. 同步发送不用重试

   • RocketMQ 的 Producer 支持同步、异步、单向三种发送模式。调用 producer.send(msg) 若发生网络抖动或 Broker 不可用时会抛出 MQClientException、RemotingException、MQBrokerException、InterruptedException 等异常。如果开发者未捕获或未配置 retryTimesWhenSendFailed（同步发送默认重试 2 次），出现一次发送失败即可造成消息丢失。

2. 异步发送回调失败后未再次补偿

   • 异步发送接口 producer.send(msg, SendCallback) 只会将发送请求放到网络层，如果网络断开或 Broker 拒收，回调会触发 onException(Throwable)。若开发者在该回调内未进行二次补偿（比如重试或将消息持久化到本地 DB），则异步发送失败的消息会被丢弃。

3. 事务消息业务逻辑与消息返回不一致

   • 事务消息分为“半消息发送”和“本地事务执行”。如果开发者没有正确实现 TransactionListener 中的 executeLocalTransaction 或 checkLocalTransaction 逻辑，当本地事务异常后，Broker 会根据 TransactionCheckMax 参数多次回查，但如果回查策略配置不当或超时，该“半消息”最终可能被 Broker 丢弃。

3.2 Broker 存储层面

1. 刷盘/同步策略不当

   • RocketMQ 默认刷盘模式为异步刷盘（ASYNC\_FLUSH），即消息先写到内存，稍后后台线程刷到磁盘。在高并发或磁盘 IO 高峰时，会导致内存中的消息尚未刷盘就被认为已发送成功。一旦 Broker 崩溃，这部分未刷盘记录会丢失。
   • 如果使用同步刷盘（SYNC\_FLUSH）模式，虽然可避免上述风险，但会牺牲吞吐量并有可能导致高延迟。

2. 主从同步配置不当

   • 在集群模式下，Master 接收到消息后需要同步给 Slave。如果设置为“异步双写”（异步复制到 Slave），Master 一旦崩溃，而 Slave 尚未同步到最新数据，就会导致接收过但未同步的消息丢失。
   • 若设置为“同步双写”（SYNC\_DUP 和 SLAVE\_TYPE\_SYNC：404），Master 会等待至少一个 Slave 返回 ACK 后才认为写入成功，但性能开销较大，且在某些极端网络抖动场景下依旧有窗口丢失。

3. Broker 配置不足导致持久化失败

   • 存储目录磁盘空间不足、文件句柄耗尽、文件系统错误等，都可能导致 RocketMQ 无法正常持久化消息。此时，Broker 会抛出 DiskFullException 或相关异常，如果监控与告警未及时触发，就会出现消息写入失败而丢失。

3.3 消费者消费层面

1. 自动提交 Offset 时机不当

   • 默认消费模型中，DefaultMessageListenerConcurrently 在消费成功之后，会自动提交 Offset。如果消费者在业务逻辑异常时仍然让消费框架认为“已消费”，则该消息跳过，不会重试，彻底丢失。
   • 反过来，如果采用手动提交 Offset，若提交时机放在业务逻辑之前，也会导致相同问题。

2. 消费者业务端未做幂等性

   • 假设消费端在处理过程中出现异常，但依旧把这条消息标记为“已消费”并提交 Offset。再次启动时，没有该消息可消费，如果消费端对业务系统幂等保障不足，可能导致某些更新未落盘，表现为“丢失”。

3. rebalance 高峰期漏拉取消息

   • 当消费者组扩容或缩容时，Broker 会触发 Rebalance 逻辑，将部分队列从一个实例迁移到另一个实例。如果 Rebalance 过程中，没有正确获取到最新 Queue 列表或偏移量变更发生错误，极端情况下会跳过某些消息。

4. 消息过滤/Tag 配置错误

   • 如果 Consumer 端订阅主题时指定了 Tag 或使用了消息过滤插件，但实际生产者发送的消息没有打上匹配 Tag，消费者会“看不到”本该消费的消息，导致消息似乎丢失。

3.4 事务消息与延迟消息

1. 事务消息回查超时

   • 事务消息发送后处于“半消息”状态，Broker 会等待 transactionCheckMax（默认 15 次）轮询回查。但如果开发者在 checkLocalTransaction 中出现了长时间阻塞或未知异常，Broker 判断超时后会丢弃该半消息。

2. 延迟消息过期或 Broker/brokerFilter 未启用

   • 延迟消息依赖 Broker 的定时轮询，如果 Broker 配置 messageDelayLevel 不正确，或者定时队列写入到错误的 Topic，导致延迟时间计算错乱，消费者会提早拉取或根本收不到，表现为“消息丢失”。

四、全面解决方案

针对上述各种导致消息丢失的场景，应当从生产端、Broker 端、消费端、监控与运维四个维度进行全链路保障。下面详述各环节的优化手段。

4.1 生产者端保障

4.1.1 同步发送 + 重试策略

• 配置重试次数
  对于同步发送方式，可通过以下方式配置发送失败时的重试：

  DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup");
  producer.setNamesrvAddr("127.0.0.1:9876");
  // 如果 send() 抛异常，则会重试 retryTimesWhenSendFailed 次（默认 2 次）
  producer.setRetryTimesWhenSendFailed(3);
  producer.start();

• 捕获异常并补偿
  即使开启了重试，也要在 send(...) 出现异常时捕获并做补偿（例如写入 DB、落盘本地文件，以便后续补发）：

  try {
      SendResult result = producer.send(msg);
      if (result.getSendStatus() != SendStatus.SEND_OK) {
          // 保存消息到本地持久化，如 DB，以便后续补偿
          saveToLocal(msg);
      }
  } catch (Exception e) {
      // 记录并持久化消息供定时补偿
      saveToLocal(msg);
      log.error("同步发送异常，消息已持久化待重发", e);
  }

4.1.2 异步发送 + 回调补偿

• 异步发送能提高吞吐，但需要在 onException 回调中做好补偿逻辑：

  producer.send(msg, new SendCallback() {
      @Override
      public void onSuccess(SendResult sendResult) {
          // 可记录日志或统计指标
          log.info("异步发送成功：{}", sendResult);
      }
  
      @Override
      public void onException(Throwable e) {
          // 此处需要将消息持久化到本地 DB 或消息表，用定时任务补偿
          saveToLocal(msg);
          log.error("异步发送失败，消息已持久化待重发", e);
      }
  });

• 补偿机制

  • 定时扫描本地持久化库，重新调用 send（同步/异步）发送，直到成功为止。
  • 当重试次数超出预设阈值，可以发邮件/报警人工介入。

4.1.3 幂等性与消息唯一 ID

• 在消息体中添加唯一业务 ID（如订单号），消费者在处理时先检查该 ID 是否已在业务 DB 中存在，若已存在则直接幂等忽略。这样即使发生生产端重试或重复发送，也不会导致业务系统重复消费或数据不一致。

  Message msg = new Message("TopicOrder", "TagNewOrder", orderId, bodyBytes);
  producer.send(msg);

• 消费端在处理前需查询幂等表：

  public void onMessage(MessageExt message) {
      String orderId = message.getKeys();
      if (orderExists(orderId)) {
          log.warn("幂等检测：订单 {} 已处理，跳过", orderId);
          return;
      }
      // 处理逻辑...
      markOrderProcessed(orderId);
  }

4.1.4 事务消息

• 如果应用场景需要“先写 DB，再发送消息”或“先发送消息，再写 DB”的强一致性逻辑，可以使用 RocketMQ 的事务消息。事务消息分为两步：

  1. 发送 Half 消息（prepare 阶段）：RocketMQ 会先发送半消息，此时 Broker 不会将该消息投递给消费者。
  2. 执行本地事务：开发者在 executeLocalTransaction 中执行 DB 写入或其他本地事务。
  3. 提交/回滚：若本地事务成功，调用 TransactionMQProducer.commitTransaction 通知 Broker 提交消息；若本地事务失败，则 rollbackTransaction 使 Broker 丢弃半消息。

• 示例代码：

  // 1. 定义事务监听器
  public class TransactionListenerImpl implements TransactionListener {
  
      @Override
      public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
          String orderId = msg.getKeys();
          try {
              // 执行本地事务（比如写订单表、库存表）
              saveOrderToDB(orderId);
              // 业务成功，提交事务
              return LocalTransactionState.COMMIT_MESSAGE;
          } catch (Exception e) {
              // 本地事务失败，回滚
              return LocalTransactionState.ROLLBACK_MESSAGE;
          }
      }
  
      @Override
      public LocalTransactionState checkLocalTransaction(MessageExt msg) {
          String orderId = msg.getKeys();
          // 查询本地事务是否成功
          if (isOrderSaved(orderId)) {
              return LocalTransactionState.COMMIT_MESSAGE;
          }
          return LocalTransactionState.UNKNOW; // 继续等待或下次回查
      }
  }
  
  // 2. 发送事务消息
  TransactionMQProducer producer = new TransactionMQProducer("ProducerTxGroup");
  producer.setNamesrvAddr("127.0.0.1:9876");
  producer.setTransactionListener(new TransactionListenerImpl());
  producer.start();
  
  Message msg = new Message("TopicTxOrder", "TagTx", orderId, bodyBytes);
  producer.sendMessageInTransaction(msg, null);

• 注意事项：

  • checkLocalTransaction 方法需要保障幂等性，并对 UNKNOW 状态进行多次回查。
  • transactionCheckMax、transactionCheckInterval 等参数需根据业务特点进行合理配置，避免过度丢弃半消息。

4.2 Broker 层面保障

4.2.1 刷盘与同步配置

• 同步刷盘（SYNC\_FLUSH）
  在 Broker 端 broker.conf 或通过 BrokerController 代码配置：

  flushDiskType=SYNC_FLUSH

  或者在 Java 配置中：

  BrokerConfig brokerConfig = new BrokerConfig();
  brokerConfig.setBrokerName("broker-a");
  brokerConfig.setEnableDLegerCommitLog(false);
  brokerConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);

  • 优点：Master 在返回消息发送成功前，必须将消息刷盘并同步到至少一个 Slave，保证了高可靠。
  • 缺点：吞吐降低（约 20%\~30%），网络延迟增加。

• 同步双写（SYNC\_MASTER\_SLAVE）
  如果需要 Master-Slave 之间强同步，也可在集群模式下配置 brokerRole=ASYNC_MASTER（异步复制）或 SYNC_MASTER（同步复制），示例：

  brokerRole=SYNC_MASTER
  brokerId=0

  注意：在 SYNC_MASTER 模式下，需要至少在另一台机器上配置对应 Slave，且网络延迟要可控，否则会严重影响写入吞吐。

4.2.2 磁盘预警与多副本策略

• 磁盘阈值告警
  在 Broker 配置文件中，可设置磁盘空间阈值，当剩余空间低于阈值时，会阻止新的消息写入并触发告警：

  diskMaxUsedRatio=75   # 磁盘使用率超过 75% 即进入警戒状态

  同时，可结合监控平台（如 Prometheus + Alertmanager、Zabbix、ELK）对 Broker 磁盘利用率进行实时监控，避免磁盘耗尽导致消息无法持久化。

• 多副本方案
  通过在 Broker 集群中部署多个 Slave，实现多副本持久化。即使 Master 崩溃，Slave 可以接管并保证数据可靠性。可以结合 Proxy 模式或 NameServer 动态路由，尽量避免某台 Broker 宕机导致整体服务不可用。

4.2.3 Broker 容错与灰度扩容

• 负载均衡与分片机制
  将 Topic 切分为多个队列（Queue），分布在不同 Broker 上，既能水平扩展吞吐，又能保证单队列顺序或无序场景下的高可用。
• 故障转移（Failover）
  客户端可配置 tryLockQueueEnable、brokerSuspendMaxTimeMillis 等参数，当一个 Broker 不可用时，消费者会在备份队列中拉取消息，减少由于单点故障导致的消息“丢失”窗口。

4.3 消费者端保障

4.3.1 手动 Ack 与业务幂等

• 关闭自动提交 Offset，使用手动提交
  在 Spring Boot + RocketMQ 的 @RocketMQMessageListener 注解中，可以设置 consumeMode = ConsumeMode.ORDERLY 或 ConsumeMode.CONCURRENTLY，并开启手动 ack 模式：

  @RocketMQMessageListener(
      topic = "TopicOrder",
      consumerGroup = "cg-order",
      consumeMode = ConsumeMode.CONCURRENTLY,
      consumeThreadMax = 8,
      messageModel = MessageModel.CLUSTERING
  )
  public class OrderConsumer implements RocketMQListener<MessageExt> {
  
      @Override
      public void onMessage(MessageExt message) {
          String body = new String(message.getBody(), StandardCharsets.UTF_8);
          String orderId = message.getKeys();
          try {
              // 1. 幂等检测
              if (orderExists(orderId)) {
                  return;
              }
              // 2. 处理业务逻辑，如写 DB、调用外部接口等
              processOrder(orderId, body);
              // 3. 手动提交消费成功（如果使用原生 API）或通过返回结果通知框架
          } catch (Exception e) {
              // 4. 消费失败则抛出异常，RocketMQ 会根据配置进行重试
              throw new RuntimeException("Order 消费失败，稍后重试", e);
          }
      }
  }

• 幂等设计
  消费前先在业务数据库或 Redis 中做唯一性检查：

  public boolean orderExists(String orderId) {
      // 查询幂等表或订单表
      return orderDao.existsById(orderId);
  }
  
  public void processOrder(String orderId, String body) {
      // 将订单写入 DB，同时在幂等表中标记 orderId
      orderDao.save(new Order(orderId, body));
      idempotentDao.mark(orderId);
  }

• 重试 & 死信队列

  • 当消费出现异常时，RocketMQ 会对消息进行重试（默认 16 次），间隔策略从 10 秒逐步增长（Level 1,2,3...）。
  • 若最终仍然失败，消息会进入死信队列（DLQ），可通过监控获取该队列信息并做人工介入或二次补偿。

4.3.2 顺序消费与并发消费

• 顺序消费
  对于需要严格按顺序处理的业务，可使用 Orderly 模式，在每个队列内部保证单线程顺序消费。

  @RocketMQMessageListener(
      topic = "TopicOrder",
      consumerGroup = "cg-order",
      consumeMode = ConsumeMode.ORDERLY
  )
  public class OrderlyConsumer implements RocketMQListener<List<MessageExt>> {
      @Override
      public void onMessage(List<MessageExt> msgs) {
          for (MessageExt msg : msgs) {
              // 按消息在队列中的顺序依次处理
          }
      }
  }

• 并发消费
  对于无序场景，可采用并发方式提高吞吐。需注意：并发消费时，要避免多线程环境下对同一业务 ID 的 并发操作冲突，推荐使用分布式锁或将数据写入同一分区分库目标。

4.3.3 优化 Rebalance 逻辑

• 减小 Rebalance 造成的抖动

  • 通过设置 rebalanceDelayTimeMillisWhenException 和 consumeTimeout 等参数，降低重平衡时跳过队列的风险。
  • 同时，可在 Consumer 启动或关闭时，将应用实例置于维护模式，短暂停止拉取新队列，待 Rebalance 完成后再恢复正常消费。
• 配合 Consistent Hash 做队列分配
  在消费组队列分配策略中使用一致性 Hash(MixAll等），当消费者上下线时，只会造成极少量队列重新分配，降低 Rebalance 产生的“空洞”风险。

4.4 监控与运维保障

4.4.1 RocketMQ 自带监控 + 前端面板

• RocketMQ-console

  • RocketMQ 官方提供了一套图形化控制台 rocketmq-console（Java Web 应用）。
  • 启动后，可查看 Broker 列表、Topic 配置、Producer/Consumer 状态、延迟队列、死信队列和消息积压等关键指标，及时发现消息丢失或堆积风险。

• 指标采集与 Prometheus Exporter
  在 Broker 和 Consumer 端集成 Prometheus Exporter，将关键指标（消息入队速率、出队速率、延迟时间、存储 lat、消费失败次数、重试次数、死信队列大小）暴露给 Prometheus。然后通过 Grafana 仪表盘可视化：

  • Broker 端指标示例：

    rocketmq_broker_put_message_total
    rocketmq_broker_get_message_total
    rocketmq_broker_put_message_failed_total
    rocketmq_broker_get_message_failed_total

  • Consumer 端指标示例：

    rocketmq_consumer_pull_time_total
    rocketmq_consumer_consume_time_total
    rocketmq_consumer_consume_failed_total

4.4.2 日志预警与告警体系

• Broker 日志收集

  • 配置 logback-spring.xml 或 log4j2.xml，对 com.alibaba.rocketmq.broker、org.apache.rocketmq.store 等包级别日志做采集。
  • 当出现 DiskFullException、SlaveNotAvailableException、BrokerNotAvailableException 等关键异常时，通过 ELK/Graylog/Fluentd 将日志集中到日志平台，并触发告警。

• 生产者 & 消费者告警

  • 生产者端当连续 send() 异常超过阈值，可将告警信息推送到监控系统。
  • 消费者端若出现死信队列消息数量超过阈值、消费失败率过高，亦应触发告警邮件/钉钉通知。

4.4.3 灰度扩容与演练

• 分批灰度测试

  • 在线上新增 Broker 或 Consumer 副本时，应先在非关键 Topic 或流量较低的 Topic 进行灰度测试，验证配置与网络连通性，确保不会影响主业务。

• 灾备演练

  • 定期进行 Broker 宕机、网络抖动、磁盘满载等场景的模拟演练，验证同步刷盘、Slave 切换、消费者 Rebalance 的可靠性与容错能力。

五、图解：RocketMQ 消息流转与保全流程

5.1 生产者发送到 Broker 存储流程

flowchart TD
    subgraph Producer 端
        A1[构建消息 Message] --> A2[同步/异步 send() 调用]
        A2 --> A3{重试?}
        A3 -- 成功 --> A4[消息发往 Broker]
        A3 -- 失败且重试未成功 --> A5[本地持久化补偿]
    end

    subgraph Broker 端
        A4 --> B1[接收消息写入 CommitLog（内存）]
        B1 --> B2{刷盘模式?}
        B2 -- ASYNC --> B3[内存返回 Client；后台刷盘线程将 CommitLog 持久化]
        B2 -- SYNC --> B4[同步刷盘到磁盘；等待 Slave ACK；返回 Client]
        B3 --> B5[CommitLog 持久化完成后异步通知]
        B4 --> B5
        B5 --> B6[Flush ConsumerQueue 索引]
    end

• 要点

  • 同步发送 + 同步刷盘 + 同步 Slave ACK ⇒ 最可靠，但延迟最高。
  • 异步发送 + 异步刷盘 ⇒ 延迟最低，但有短暂窗口可能丢失。
  • 写入 CommitLog 后，Broker 会根据 topicQueueInfo 更新 ConsumeQueue 索引，令消费者可拉取该消息。

5.2 消费者拉取 & 消费流程

flowchart TD
    subgraph Consumer 端
        C1[ConsumerGroup 拉取消息] --> C2[按照负载策略选择 Broker 和 Queue]
        C2 --> C3[调用 PullMessageService 拉取请求]
        C3 --> C4{Message Ext 是否存在?}
        C4 -- 存在 --> C5[返回消息列表给 Consumer]
        C4 -- 不存在 ⇒ 暂无消息 --> C6[空轮询，等待下一次]
        C5 --> C7[消费端业务处理]
        C7 --> C8{处理成功?}
        C8 -- 是 --> C9[提交 Offset]
        C8 -- 否 --> C10[抛出异常，进入重试队列或死信队列]
    end

    subgraph Broker 端
        BQ1[Broker 持有 ConsumeQueue 索引] --> BQ2[按偏移量返回对应 CommitLog 消息]
        BQ2 --> C5
    end

• 要点

  • Pull 与 Push 模式：RocketMQ 默认采用 Pull 模式，Consumer 定时主动向 Broker 请求消息。
  • 消费成功后提交 Offset，否则 Consumer 将在下次拉取时重试。
  • 重试次数耗尽后，RocketMQ 会将该消息扔进死信队列，需人工或程序补偿。

六、代码示例

以下示例展示生产者、消费者在各自端如何实现可靠保证的关键逻辑。

6.1 生产者示例：同步 & 异步 + 本地补偿

package com.example.rocketmq.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ReliableProducer {

    private static final Logger log = LoggerFactory.getLogger(ReliableProducer.class);

    private final DefaultMQProducer producer;

    public ReliableProducer() throws MQClientException {
        producer = new DefaultMQProducer("ReliableProducerGroup");
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 重试 3 次
        producer.setRetryTimesWhenSendFailed(3);
        // 同步模式下的超时时间
        producer.setSendMsgTimeout(3000);
        producer.start();
    }

    public void sendSync(String topic, String body, String key) {
        try {
            Message msg = new Message(topic, "***".getBytes());
            msg.setBody(body.getBytes());
            msg.setKeys(key);
            // 同步发送
            SendResult result = producer.send(msg);
            log.info("同步发送结果：{}", result);
            if (result.getSendStatus() != SendResult.SendStatus.SEND_OK) {
                saveToLocalStorage(msg);
            }
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            // 本地补偿
            log.error("同步发送异常，持久化消息待补发", e);
            saveToLocalStorage(new Message(topic, key, body.getBytes()));
        }
    }

    public void sendAsync(String topic, String body, String key) {
        Message msg = new Message(topic, "***".getBytes());
        msg.setBody(body.getBytes());
        msg.setKeys(key);
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步发送成功：{}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.error("异步发送失败，持久化消息待补发", e);
                saveToLocalStorage(msg);
            }
        });
    }

    private void saveToLocalStorage(Message msg) {
        // TODO: 实际场景可持久化到 DB、文件，或发送到另一个可靠队列
        log.warn("持久化消息 Key={} Body={} 到本地，以便后续重发", msg.getKeys(), new String(msg.getBody()));
    }

    public void shutdown() {
        producer.shutdown();
    }
}

6.2 消费者示例：并发 & 死信队列处理

package com.example.rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class ReliableConsumer {

    private static final Logger log = LoggerFactory.getLogger(ReliableConsumer.class);

    private final DefaultMQPushConsumer consumer;

    public ReliableConsumer() throws Exception {
        consumer = new DefaultMQPushConsumer("ReliableConsumerGroup");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 设置从队列头开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 绑定 Topic 和 Tag
        consumer.subscribe("TopicOrder", "*");
        // 注册并发消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt message : list) {
                    String body = new String(message.getBody(), StandardCharsets.UTF_8);
                    String orderId = message.getKeys();
                    try {
                        // 幂等检查
                        if (orderExists(orderId)) {
                            log.warn("幂等检测：订单 {} 已处理，跳过", orderId);
                            continue;
                        }
                        // 处理业务逻辑
                        processOrder(orderId, body);
                        log.info("订单 {} 处理成功", orderId);
                    } catch (Exception e) {
                        log.error("订单 {} 处理失败，稍后重试", orderId, e);
                        // 返回稍后重试，RocketMQ 会根据配置重试或进入死信队列
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 全部消息成功消费，返回成功状态
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    private boolean orderExists(String orderId) {
        // TODO: 查询数据库/Redis 判断订单是否已处理
        return false;
    }

    private void processOrder(String orderId, String body) {
        // TODO: 执行业务逻辑，如写订单表、扣减库存、发通知等
        // 如果出现异常，则抛出，触发重试机制
    }

    public void shutdown() {
        consumer.shutdown();
    }
}

• 死信队列处理：当消息在重试次数耗尽后（默认 16 次），会被丢弃并发送到死信队列。你可以通过 RocketMQ 控制台或 API 拉取该死信队列，对消息做二次补偿或报警。死信队列 Topic 后缀默认为 %-RETRY-%d（消费重试队列）和 %-DLQ（死信队列）。例如消费者组 ReliableConsumerGroup 的死信队列为 TopicOrder-RETRY-ReliableConsumerGroup 与 TopicOrder-DLQ-ReliableConsumerGroup。

七、常见误区与注意事项

1. 误以为 send() 方法“只要不报错就一定写入磁盘”

   • 实际上，在异步刷盘场景下，send() 只保证写入 CommitLog 缓存，真正刷盘到磁盘要依赖后台刷盘线程，若此时发生宕机就会丢失。

2. 消费者自动提交 Offset 时机盲目

   • 切忌使用“默认自动提交 offset”再根据返回值判断消费成功的方法。推荐使用 RocketMQ 原生 API 或 Spring RocketMQ 的手动 ack 方式，确保业务处理完全成功后再提交 offset。

3. 过度依赖事务消息，忽略性能开销

   • 事务消息需要额外的回查开销，且会占用 Broker 半消息存储空间。仅在强一致性场景下使用事务消息，普通异步通知场景不推荐使用。

4. 只关注生产端，不关注 Broker 与 Consumer 状态

   • 如果缺少对 Broker 磁盘、网络、线程池等指标的监控，依赖经验设置刷盘与同步参数，往往在高峰期会出现不可预测的消息丢失。

5. 延迟消息未启用正确的延迟级别

   • RocketMQ 的延迟级别由 messageDelayLevel 参数统一管理，默认有 18 级（1s、5s、10s、30s、1m、2m...），如果想使用 2 分钟延迟，需要在 Broker 配置或客户端代码中指定合适的 level，否则会直接投递到消费者。

八、小结

消息丢失对业务系统的影响往往不可逆且难以挽回。本文从生产者、Broker、消费者三个层面深入剖析了 RocketMQ 在实际生产环境中最常见的消息丢失场景，并给出全面的解决方案：

1. 生产端：

   • 同步发送务必开启重试、捕获异常并补偿；
   • 异步发送在回调中做好落盘与补发；
   • 必要时使用事务消息保证“库 + 消息”强一致。

2. Broker 端：

   • 根据业务对可靠性要求选择刷盘与主从同步策略；
   • 配置磁盘预警、自动拒绝写入；
   • 部署多副本、灰度演练，保证机器宕机也不会产生数据丢失。

3. 消费者端：

   • 使用手动 ACK 或确保自动提交在业务逻辑之后；
   • 统一做幂等设计，避免重复或跳过；
   • 利用死信队列与重试队列进行补偿机制。

4. 监控与运维：

   • 部署 RocketMQ 控制台、Prometheus + Grafana 监控集群指标；
   • 日志告警及时发现异常；
   • 定期进行故障演练。

只要在各个环节合理配置、代码中做好异常捕获与补偿，并配合完善的监控与告警机制，就能大幅降低 RocketMQ 在生产环境中出现消息丢失的概率，打造高可靠分布式消息系统。

实践建议：

• 在开发初期，先按照高可靠扩展架构设计：同步双写+幂等消费+手动 ACK；
• 在测试环境压测后，根据吞吐量与延迟要求，逐步调整为异步刷盘或部分异步同步；
• 定期检查死信队列与重试队列，及时补偿生产与消费失败的消息。

愿本文能帮助你从根本上理解并避免 RocketMQ 的消息丢失问题，打造更稳定、可靠的分布式消息系统。