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

分布式系统的最终一致性方案——从本地消息表到事务消息的渐进式演进一、背景与问题在微服务拆分后跨服务的数据一致性问题变得无处不在。以电商系统的下单减库存为例订单服务创建订单记录后需要通知库存服务扣减库存。如果库存服务宕机订单已经创建库存未扣减——这是一个典型的一致性问题。分布式事务如 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 Messagebr/MQ 投递消息给消费者] B3 --|失败| B5[发送 Rollback Messagebr/MQ 丢弃消息] end subgraph 方案三: 事件溯源Event Sourcing C1[订单服务产生事件: OrderCreated] -- C2[事件写入 Event Store] C2 -- C3[库存服务订阅事件流] C3 -- C4[处理事件扣减库存] end选型决策矩阵维度本地消息表事务消息事件溯源开发成本低纯业务代码中需接入 RocketMQ高架构层面改造运维成本中需维护定时任务低RocketMQ 内置能力高Event Store 运维一致性保证最终一致最终一致更高可靠性最终一致性能定时轮询有延迟低延迟低延迟适用场景中小规模、简单业务高并发、核心链路事件驱动架构三、实战演示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 ListEventMessage 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 { // 构建事务消息 MessageString 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/RollbackRocketMQ 会在一定时间后向 Producer 发起回查。回查的关键注意事项幂等性回查可能被多次调用必须保证幂等。检查逻辑的准确性不能仅凭订单是否存在判断——如果订单被其他流程删除回查逻辑需考虑这种边缘情况。回查频率默认 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); } } }五、总结最终一致性的方案选择遵循够用就好的原则本地消息表适合 QPS 500、开发成本敏感的中小规模业务。部署简单不依赖特定 MQ。RocketMQ 事务消息适合 QPS 1000、一致性要求高的核心链路。依赖 RocketMQ开发成本中等。事件溯源适合事件驱动架构、需要审计追溯的业务。技术门槛和运维成本最高。在实际项目中三条核心原则始终有效消息的原子性由事务保证消息的可靠性由回查保证消费的幂等性由唯一 ID 保证。只要这三层防线设计到位最终一致性就是安全可靠的。作者程序员鸭梨李然Java 架构师专注分布式系统一致性与微服务治理实践。欢迎留言交流你的最终一致性方案。