如何保证事务消息的最终一致性？

一、核心原理：事务消息最终一致性的底层逻辑

1. 第一阶段：生产者发送「半消息」（Broker 暂存，不可投递），确保消息能成功写入 Broker；
2. 第二阶段：执行本地事务，根据结果向 Broker 发送「提交 / 回滚」指令：
   • 提交：Broker 投递消息给消费者，保证「本地事务成功 → 消息必投递」；
   • 回滚：Broker 删除半消息，保证「本地事务失败 → 消息不投递」；
3. 兜底机制：若生产者宕机 / 网络异常导致 Broker 未收到指令，Broker 定时回查生产者，确认本地事务状态后补全操作，避免「状态丢失」。

二、生产端：保证「本地事务 ↔ 消息状态」一致

1. 本地事务状态必须持久化（核心！）

• 问题：若本地事务状态仅存于内存，生产者宕机后，Broker 回查时无法获取真实状态，导致一致性破坏；
• 解决方案：将本地事务状态（成功 / 失败）写入数据库 / 事务日志，回查时从持久化存储读取，而非内存。
  java

  运行

  // 示例：本地事务执行后，记录事务日志（关键）
  private boolean doLocalBiz(String orderId) {
      // 1. 执行本地事务（如扣减库存）
      boolean success = deductStock(orderId);
      // 2. 记录事务日志（落地到DB，用于回查）
      saveTxLog(orderId, success ? "SUCCESS" : "FAIL");
      return success;
  }
  
  // 事务回查时，从DB读取状态
  private boolean checkLocalBiz(String orderId) {
      TxLog txLog = getTxLogByOrderId(orderId);
      return "SUCCESS".equals(txLog.getStatus());
  }
  

2. 本地事务必须是「幂等 / 可重试」的

• 问题：Broker 回查可能多次触发本地事务状态校验，若本地事务非幂等，重复执行会导致数据错误；
• 解决方案：
  • 本地事务操作（如扣减库存、更新订单）基于「业务唯一 ID（如订单 ID）」做幂等；
  • 避免本地事务包含「非幂等操作」（如新增记录、调用第三方接口），可通过「状态机 + 乐观锁」控制。
  java

  运行

  // 示例：本地事务幂等实现（乐观锁）
  private boolean deductStock(String orderId) {
      // 仅当库存>0且订单未处理时，扣减库存（乐观锁）
      int affected = jdbcTemplate.update(
          "UPDATE t_stock SET num = num - 1 WHERE order_id = ? AND num > 0 AND status = 'UNDO'",
          orderId
      );
      return affected > 0;
  }
  

3. 事务回查机制配置优化

4. 生产端异常处理

• 本地事务执行异常：返回 UNKNOWN 触发回查，而非直接回滚（避免因临时异常误判）；
• 半消息发送失败：开启生产者重试（retry-times-when-send-failed=3），确保半消息能写入 Broker；
• 提交 / 回滚指令发送失败：无需处理，依赖 Broker 回查兜底。

三、消费端：保证「消息消费 ↔ 业务结果」一致

1. 消费端必须实现幂等（核心！）

• 问题：Broker 重试机制、网络异常会导致消息重复投递，重复消费会破坏数据一致性；
• 解决方案：基于「业务唯一 ID（如订单 ID）」或「RocketMQ 消息 ID」做幂等校验：
  java

  运行

  @Override
  public void onMessage(String msg) {
      OrderPayDTO dto = JSON.parseObject(msg, OrderPayDTO.class);
      String orderId = dto.getOrderId();
      
      // 1. 幂等校验：查消费日志，已消费则直接返回
      if (checkConsumed(orderId)) {
          return;
      }
      
      // 2. 执行业务逻辑（如创建物流单）
      createLogistics(orderId);
      
      // 3. 记录消费日志（落地DB，完成幂等闭环）
      saveConsumeLog(orderId);
  }
  

2. 消费失败必须触发重试（兜底）

• 默认机制：RocketMQ 消费失败后会自动重试（默认 16 次），重试间隔逐渐延长；
• 优化配置：根据业务场景调整重试次数（如 retry-times-when-consume-failed=3），避免过多重试导致消息堆积；
• 重试耗尽处理：消息进入「死信队列（DLQ）」，需监控死信队列并人工介入处理（避免消息丢失）。

3. 消费逻辑必须是「可重试」的

• 消费逻辑避免依赖「一次性资源」（如临时文件、第三方接口单次调用）；
• 若调用第三方接口失败，可通过「定时任务补偿」，确保业务最终执行成功。

四、全局兜底：异常场景的一致性保障

1. 消息状态监控与告警

• 监控 Broker 中「待回查的半消息数量」：若数量激增，说明生产端可能异常，及时告警；
• 监控死信队列：若有消息进入，立即触发告警，人工介入处理；
• 监控消费堆积：若消费进度落后，及时扩容消费线程或排查消费逻辑。

2. 定时补偿任务（最终兜底）

• 针对「本地事务成功但消息未投递」「消息投递但消费失败」的场景，开发定时补偿任务：
  • 扫描「本地事务成功但无消费记录」的订单，主动触发消费逻辑；
  • 扫描死信队列，重试消费失败的消息。
  java

  运行

  // 示例：定时补偿任务（每5分钟执行一次）
  @Scheduled(fixedRate = 300000)
  public void compensateTxMsg() {
      // 1. 扫描本地事务成功但未消费的订单
      List<String> unConsumedOrderIds = getUnConsumedOrderIds();
      // 2. 重新触发消费逻辑
      for (String orderId : unConsumedOrderIds) {
          createLogistics(orderId);
          saveConsumeLog(orderId);
      }
  }
  

3. 数据一致性校验

• 若「生产端事务成功」但「消费端无消费记录」，触发补偿；
• 若「生产端事务失败」但「消费端有消费记录」，执行回滚操作（如恢复库存、撤销物流单）。

五、核心避坑点（一致性破坏的常见原因）

六、核心总结：最终一致性的闭环保障

1. 生产端：半消息保证消息能写入 Broker，本地事务状态持久化 + 回查兜底保证「本地事务结果 ↔ 消息状态」一致；
2. 消费端：幂等消费 + 失败重试保证「消息投递 → 业务必执行成功」；
3. 全局兜底：监控告警 + 定时补偿 + 数据对账，解决极端异常场景的一致性问题。