分布式系统的最终一致性方案——从本地消息表到事务消息的渐进式演进

一、背景与问题

在微服务拆分后，跨服务的数据一致性问题变得无处不在。以电商系统的"下单减库存"为例：订单服务创建订单记录后，需要通知库存服务扣减库存。如果库存服务宕机，订单已经创建，库存未扣减——这是一个典型的一致性问题。

分布式事务（如 Seata AT/TCC）能解决强一致性问题，但引入的性能开销和复杂度不一定值得。对于大多数业务场景，"最终一致性"是性价比更高的选择：允许短暂的数据不一致，在可接受的时间窗口内达到一致状态。本文梳理从本地消息表到 RocketMQ 事务消息的演进路径，以及在真实项目中的选型决策。

二、方案设计

最终一致性有两种实现范式：

• 最大努力通知（Best-Effort Notification）：发起方尽最大努力通知接收方，不保证一定到达。
• 可靠消息最终一致（Reliable Message Eventual Consistency）：通过消息中间件的事务能力，保证消息与业务操作的原子性。

flowchart TD subgraph "方案一: 本地消息表" A1["订单服务执行业务SQL"] --> A2["在同一事务中写入本地消息表"] A2 --> A3["定时任务扫描未发送消息"] A3 --> A4["发送MQ消息"] A4 --> A5["消息发送成功后删除/标记本地消息"] end subgraph "方案二: 事务消息（RocketMQ）" B1["订单服务发送 Half Message"] --> B2["MQ 返回 Half 确认"] B2 --> B3["订单服务执行本地事务"] B3 -->|成功| B4["发送 Commit Message<br/>MQ 投递消息给消费者"] B3 -->|失败| B5["发送 Rollback Message<br/>MQ 丢弃消息"] end subgraph "方案三: 事件溯源（Event Sourcing）" C1["订单服务产生事件: OrderCreated"] --> C2["事件写入 Event Store"] C2 --> C3["库存服务订阅事件流"] C3 --> C4["处理事件：扣减库存"] end

选型决策矩阵

三、实战演示

3.1 本地消息表实现

/** * 基于本地消息表的可靠消息投递服务。 * * 核心设计： * 1. 业务操作与消息写入在同一本地事务中 * 2. 定时任务扫描未发送消息并重试 * 3. 消息状态机：PENDING -> SENT / FAILED */ @Service @Slf4j public class LocalMessageService { @Autowired private OrderMapper orderMapper; @Autowired private EventMessageMapper eventMessageMapper; @Autowired private RabbitTemplate rabbitTemplate; /** * 创建订单并写入本地消息表——同一事务保证原子性。 * * @param order 订单信息 * @throws OrderException 如果执行失败 */ @Transactional(rollbackFor = Exception.class) public void createOrderWithMessage(Order order) { try { // 1. 写入订单表（业务操作） orderMapper.insert(order); log.info("订单创建成功: orderId={}", order.getId()); // 2. 写入本地消息表（同一个事务） EventMessage message = new EventMessage(); message.setMessageId(UUID.randomUUID().toString()); message.setTopic("order.events"); message.setTag("order.created"); message.setBody(JSON.toJSONString(order)); message.setStatus("PENDING"); message.setRetryCount(0); message.setCreateTime(LocalDateTime.now()); eventMessageMapper.insert(message); log.info("本地消息写入成功: messageId={}", message.getMessageId()); } catch (Exception e) { log.error("创建订单并写入本地消息失败: orderId={}", order.getId(), e); throw new OrderException("订单创建失败: " + e.getMessage(), e); } // 事务提交后，消息进入"待发送"状态 } /** * 定时任务：扫描并发送待处理消息。 * 配置为每 10 秒执行一次。 * * 重试策略：最多重试 5 次，指数退避（1m, 2m, 4m, 8m, 16m）。 */ @Scheduled(fixedDelay = 10000) public void sendPendingMessages() { // 查询待发送消息（状态 PENDING，且重试次数 ≤ 5） List<EventMessage> pendingMessages = eventMessageMapper .selectPendingMessages(5); if (pendingMessages.isEmpty()) { return; } log.info("扫描到 {} 条待发送消息", pendingMessages.size()); for (EventMessage message : pendingMessages) { try { // 发送到 RabbitMQ rabbitTemplate.convertAndSend( message.getTopic(), message.getTag(), ConvertUtil.toRabbitMessage(message) ); // 更新为已发送 eventMessageMapper.updateStatus( message.getMessageId(), "SENT"); log.info("消息发送成功: messageId={}", message.getMessageId()); } catch (Exception e) { log.error("消息发送失败: messageId={}, retry={}", message.getMessageId(), message.getRetryCount(), e); // 增加重试次数，标记失败 eventMessageMapper.incrementRetryAndUpdateStatus( message.getMessageId(), "FAILED", message.getRetryCount() + 1 ); } } } }

3.2 RocketMQ 事务消息实现

/** * RocketMQ 事务消息生产者——保证本地事务与消息发送的原子性。 * * 流程： * 1. 发送 Half Message（对消费者不可见） * 2. 执行本地事务 * 3. 根据本地事务结果，发送 Commit 或 Rollback * 4. RocketMQ 定期回查（checkLocalTransaction）处理未知状态 */ @Component @Slf4j public class OrderTransactionProducer { @Autowired private RocketMQTemplate rocketMQTemplate; /** * 发送订单创建的事务消息。 * * @param order 订单信息 * @throws OrderException 如果发送失败 */ public void sendOrderCreatedMessage(Order order) { try { // 构建事务消息 Message<String> message = MessageBuilder .withPayload(JSON.toJSONString(order)) .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString()) .build(); // 发送事务消息 TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction( "order-tx-producer-group", "order.created", message, order // 传入参数，传递给本地事务执行器 ); log.info("事务消息发送结果: status={}, msgId={}", result.getLocalTransactionState(), result.getMsgId()); if (result.getLocalTransactionState() == LocalTransactionState.ROLLBACK_MESSAGE) { throw new OrderException("事务消息回滚: orderId=" + order.getId()); } } catch (OrderException e) { throw e; } catch (Exception e) { log.error("事务消息发送异常: orderId={}", order.getId(), e); throw new OrderException("事务消息发送失败: " + e.getMessage(), e); } } /** * 本地事务监听器：执行本地事务并返回执行结果。 */ @Component @RocketMQTransactionListener(txProducerGroup = "order-tx-producer-group") public static class OrderTransactionListenerImpl implements RocketMQLocalTransactionListener { @Autowired private OrderMapper orderMapper; @Override public RocketMQLocalTransactionState executeLocalTransaction( Message msg, Object arg) { try { Order order = (Order) arg; // 执行本地事务：插入订单 orderMapper.insert(order); log.info("本地事务执行成功——订单已创建: orderId={}", order.getId()); return RocketMQLocalTransactionState.COMMIT; } catch (Exception e) { log.error("本地事务执行失败: ", e); return RocketMQLocalTransactionState.ROLLBACK; } } /** * 回查方法：当 Half Message 超过一定时间未收到 Commit/Rollback 时， * RocketMQ 会调用此方法确认本地事务的最终状态。 * * @param msg 消息对象 * @return 本地事务的最终状态 */ @Override public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { String orderJson = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8); Order order = JSON.parseObject(orderJson, Order.class); // 查询订单是否存在来确认本地事务结果 Order dbOrder = orderMapper.selectById(order.getId()); if (dbOrder != null) { log.info("回查确认——订单存在，提交消息: orderId={}", order.getId()); return RocketMQLocalTransactionState.COMMIT; } else { log.warn("回查确认——订单不存在，回滚消息: orderId={}", order.getId()); return RocketMQLocalTransactionState.ROLLBACK; } } } }

四、深度解析

4.1 本地消息表的定时扫描策略设计

定时扫描的频率需要在延迟和数据库压力之间折中。我们的实践：

• 正常模式：10 秒间隔扫描，消息发送延迟中位数 5 秒。
• 降级模式：当积压消息超过 1000 条时，缩短间隔到 1 秒，直到积压消除。
• 上限保护：每次扫描最多处理 200 条消息，防止单次处理过久导致后续间隔推迟。

此外，消息的"已完成"记录不应直接物理删除，而应标记为ARCHIVED状态并定期归档（如每天凌晨迁移到历史表），避免主表无限膨胀。

4.2 事务消息的回查机制原理

RocketMQ 事务消息的可靠性核心在于回查机制。当 Half Message 发送后，如果 Producer 宕机或因网络问题无法回复 Commit/Rollback，RocketMQ 会在一定时间后向 Producer 发起回查。

回查的关键注意事项：

1. 幂等性：回查可能被多次调用，必须保证幂等。
2. 检查逻辑的准确性：不能仅凭"订单是否存在"判断——如果订单被其他流程删除，回查逻辑需考虑这种边缘情况。
3. 回查频率：默认 60 秒首次回查，最多回查 15 次。总时间窗口约 30 分钟，之后消息会被丢弃。

4.3 消费者幂等性设计

消息可能被重复投递，消费者必须实现幂等处理。常见策略：

/** * 消费者幂等处理——基于消息 ID 的唯一约束防重。 */ @Service @Slf4j public class OrderEventConsumer { @Autowired private InventoryMapper inventoryMapper; @Autowired private MessageRecordMapper messageRecordMapper; @RabbitListener(queues = "order.events.queue") @Transactional(rollbackFor = Exception.class) public void handleOrderCreated(Message message) { String messageId = message.getMessageProperties().getMessageId(); try { // 幂等检查：消息是否已处理 if (messageRecordMapper.existsByMessageId(messageId)) { log.info("消息已处理，跳过: messageId={}", messageId); return; } Order order = parseOrder(message); // 扣减库存 inventoryMapper.deduct(order.getProductId(), order.getQuantity()); // 记录消息处理状态（防重） messageRecordMapper.insert(new MessageRecord(messageId, "PROCESSED")); log.info("订单事件处理完成: orderId={}", order.getId()); } catch (DuplicateKeyException e) { log.info("并发检测到重复消息，忽略: messageId={}", messageId); } catch (Exception e) { log.error("订单事件处理失败: messageId={}", messageId, e); throw new MessageHandleException("消费处理异常: " + e.getMessage(), e); } } }

五、总结

最终一致性的方案选择，遵循"够用就好"的原则：

1. 本地消息表：适合 QPS < 500、开发成本敏感的中小规模业务。部署简单，不依赖特定 MQ。
2. RocketMQ 事务消息：适合 QPS > 1000、一致性要求高的核心链路。依赖 RocketMQ，开发成本中等。
3. 事件溯源：适合事件驱动架构、需要审计追溯的业务。技术门槛和运维成本最高。

在实际项目中，三条核心原则始终有效：消息的原子性由事务保证，消息的可靠性由回查保证，消费的幂等性由唯一 ID 保证。只要这三层防线设计到位，最终一致性就是安全可靠的。

作者：程序员鸭梨（李然），Java 架构师，专注分布式系统一致性与微服务治理实践。欢迎留言交流你的最终一致性方案。