最终一致性方案：本地消息表、事务消息、Saga模式

引言

先讲一个我亲身踩过的坑。

2023年，我当时所在的团队负责一个电商中台的订单服务重构。业务逻辑并不复杂：用户下单后，需要扣减库存、锁定优惠券、增加积分。最初是单机应用，一个本地事务包住所有操作，干干净净。

微服务拆分后，订单、库存、优惠券、积分各成独立服务，各自拥有独立数据库。问题来了：用户下单成功后，如果库存扣减失败怎么办？如果优惠券锁定成功但积分增加失败怎么办？

当时我们天真地采用了先更新订单状态，再依次调用其他服务的方案。上线第一个月，就出现了大量数据不一致：有订单显示已支付但库存没扣、有优惠券被锁定但订单已取消。客服工单爆满，凌晨三点被叫起来排查数据。

后来我们调研并落地了最终一致性方案，才真正解决了这个问题。

本文将以订单场景为主线，深入剖析三种主流方案：本地消息表、事务消息、Saga模式。我会给出源码级分析和完整可运行的代码示例，并对比它们的优缺点和适用场景。

核心概念

生活类比：支付宝转账的"异步记账"

想象一个场景：你在餐厅吃完饭，用支付宝扫码付款。收银台系统需要做两件事：扣你的钱、记餐厅的账。

如果这两个操作必须同时成功或同时失败（强一致），那么支付宝和餐厅系统之间就需要一个分布式事务协调者，整个过程耗时可能从毫秒级变成秒级——你肯定不愿意在收银台等3秒。

现实中的做法是：支付宝先扣你的钱，生成一条转账记录（本地事务），然后异步通知餐厅的账户系统入账。如果通知失败，后台会有补偿任务不断重试，直到餐厅入账成功。你扫码后立刻能走人，餐厅晚几秒收到钱也无所谓。

这就是最终一致性：系统不保证任意时刻数据完全一致，但保证在一段时间后，所有副本最终达到一致状态。

技术定义

在分布式系统中，跨服务的业务操作无法用单一数据库的ACID事务来保证。最终一致性方案的核心思想是：

1. 将一个大事务拆分为多个本地事务，每个服务在自己的数据库内完成一个本地事务
2. 通过消息或事件驱动，将本地事务的结果异步传播给其他服务
3. 引入补偿机制，当后续步骤失败时，回滚前面已成功的操作

三种主流方案的区别在于：如何保证"本地事务"与"消息发送"的原子性，以及如何设计补偿逻辑。

源码/原理深度分析

方案一：本地消息表

这是最经典、也是实现成本最低的方案，由 eBay 提出。核心思想是：将业务操作和消息写入放在同一个本地事务中。

#### 核心流程

graph TD A[订单服务] -->|1. 开启本地事务| B[(订单表)] A -->|2. 写入| C[(本地消息表)] B --> D[事务提交成功] C --> D D -->|3. 异步发送消息| E[消息队列] E -->|4. 消费消息| F[库存服务] F -->|5. 扣减库存| G[(库存表)] G -->|6. 处理成功| H[确认消息] H -->|7. 删除本地消息记录| C C -.->|8. 定时任务扫描未确认消息| A A -.->|9. 重新发送| E

#### 关键源码

/**
 * 本地消息表方案 - 订单创建服务
 * 
 * 核心思想：订单数据和消息数据在同一个本地事务中写入，
 * 保证"业务操作"和"消息发送"的原子性。
 */
@Service
public class OrderService {
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private LocalMessageMapper messageMapper;
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    /**
     * 创建订单并写入本地消息表
     * 
     * @param order 订单信息
     * @return 订单ID
     */
    @Transactional(rollbackFor = Exception.class)
    public Long createOrder(Order order) {
        // 1. 插入订单数据（本地事务的一部分）
        orderMapper.insert(order);
        
        // 2. 构造本地消息记录
        LocalMessage message = LocalMessage.builder()
            .messageId(UUID.randomUUID().toString())
            .bizType("ORDER_CREATED")
            .bizId(order.getId())
            .payload(JSON.toJSONString(order))
            .status(MessageStatus.PENDING)  // 待发送状态
            .retryCount(0)
            .createTime(new Date())
            .build();
        
        // 3. 插入本地消息表（同一个本地事务！）
        messageMapper.insert(message);
        
        // 4. 注意：这里不直接发送MQ消息！
        // 事务提交后，由定时任务或事务回调来发送
        return order.getId();
    }
    
    /**
     * 事务提交后的回调 - 发送消息
     * 使用 TransactionSynchronizationManager 注册回调，
     * 确保只有在事务真正提交后才发送消息。
     */
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void sendMessageOnCommit(OrderCreatedEvent event) {
        LocalMessage message = messageMapper.selectById(event.getMessageId());
        if (message != null && message.getStatus() == MessageStatus.PENDING) {
            sendMessageWithRetry(message);
        }
    }
    
    /**
     * 发送消息并更新状态
     */
    private void sendMessageWithRetry(LocalMessage message) {
        try {
            // 发送到MQ
            kafkaTemplate.send("order-topic", message.getPayload()).get(3, TimeUnit.SECONDS);
            // 发送成功，更新状态为已发送
            messageMapper.updateStatus(message.getMessageId(), MessageStatus.SENT);
        } catch (Exception e) {
            log.error("消息发送失败，messageId: {}", message.getMessageId(), e);
            // 发送失败，状态保持PENDING，等待定时任务重试
        }
    }
    
    /**
     * 定时任务 - 扫描并重发超时未确认的消息
     * 
     * 每隔30秒扫描一次，找出超过1分钟仍未发送成功的消息进行重试
     */
    @Scheduled(fixedDelay = 30000)
    public void retryPendingMessages() {
        Date deadline = new Date(System.currentTimeMillis() - 60_000);
        List<LocalMessage> pendingMessages = 
            messageMapper.selectPendingMessages(deadline, 100);
        
        for (LocalMessage message : pendingMessages) {
            // 重试次数限制，超过10次则标记为死信
            if (message.getRetryCount() >= 10) {
                messageMapper.updateStatus(message.getMessageId(), MessageStatus.DEAD);
                continue;
            }
            sendMessageWithRetry(message);
            messageMapper.incrementRetryCount(message.getMessageId());
        }
    }
}

#### 消费端处理（幂等性保证）

/**
 * 消费端 - 库存服务
 * 
 * 关键点：消费端必须实现幂等，防止消息重复投递导致的数据问题
 */
@Component
public class InventoryConsumer {
    
    @Autowired
    private InventoryMapper inventoryMapper;
    
    @Autowired
    private ConsumedMessageMapper consumedMessageMapper;
    
    /**
     * 消费订单创建消息，扣减库存
     * 
     * 幂等方案：使用消费记录表 + 唯一索引
     */
    @KafkaListener(topics = "order-topic", groupId = "inventory-group")
    public void handleOrderCreated(String payload) {
        Order order = JSON.parseObject(payload, Order.class);
        
        // 使用消费记录表做幂等（也可以用Redis SETNX）
        String consumeKey = "ORDER_CREATED:" + order.getId();
        try {
            // 尝试插入消费记录，如果主键冲突说明已处理过
            consumedMessageMapper.insert(ConsumedMessage.builder()
                .consumeKey(consumeKey)
                .consumeTime(new Date())
                .build());
        } catch (DuplicateKeyException e) {
            // 已消费过，直接返回
            log.info("消息重复消费，跳过：{}", consumeKey);
            return;
        }
        
        // 执行实际的库存扣减业务
        int affected = inventoryMapper.deductStock(order.getProductId(), 
                                                   order.getQuantity());
        if (affected == 0) {
            // 库存不足，需要告警并触发人工处理
            // 注意：这里不能直接抛异常，否则消息会重试导致死循环
            // 应该记录失败状态，发送告警，由人工介入
            log.error("库存不足，productId: {}, quantity: {}", 
                      order.getProductId(), order.getQuantity());
        }
    }
}

方案二：事务消息（RocketMQ）

RocketMQ 的事务消息解决了本地消息表方案中的两个痛点：消息表需要额外维护、定时任务有延迟。

#### 核心原理

graph TD A[订单服务] -->|1. 发送半消息| B[RocketMQ Broker] B -->|2. 半消息保存成功| C[执行本地事务] C -->|3a. 提交| D[Commit消息] C -->|3b. 回滚| E[Rollback消息] C -->|3c. 未知| F[Broker回查] F -->|4. 查询本地事务状态| C D -->|5. 消息可见| G[消费者] G -->|6. 消费| H[库存服务]

#### 核心源码

/**
 * 事务消息方案 - 使用RocketMQ实现
 * 
 * 核心思想：通过MQ的Half Message机制，将"发送消息"和"本地事务"解耦，
 * 由MQ Broker负责事务状态的最终确认。
 */
@Component
public class TransactionalOrderService {
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    /**
     * 创建订单 - 事务消息的入口
     */
    public Long createOrderWithTransactionalMessage(Order order) {
        // 1. 构造消息
        Message<String> message = MessageBuilder
            .withPayload(JSON.toJSONString(order))
            .setHeader("orderId", order.getId())
            .build();
        
        // 2. 发送事务消息
        // 注意：这里发送的是Half Message，对消费者不可见
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
            "order-topic",
            message,
            order,  // 作为参数传给本地事务执行器
            60_000   // 3秒超时（实际生产建议30秒）
        );
        
        if (result.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE) {
            return order.getId();
        } else {
            throw new RuntimeException("订单创建失败，事务已回滚");
        }
    }
    
    /**
     * 本地事务执行器
     * 
     * 这个方法在发送Half Message之后执行，
     * 与Half Message的发送是"伪原子"的。
     */
    @RocketMQTransactionListener
    class OrderTransactionListener implements RocketMQLocalTransactionListener {
        
        /**
         * 执行本地事务
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {
                Order order = (Order) arg;
                
                // 1. 插入订单数据
                orderMapper.insert(order);
                
                // 2. 如果业务成功，提交消息
                return LocalTransactionState.COMMIT_MESSAGE;
            } catch (Exception e) {
                log.error("本地事务执行失败", e);
                // 业务失败，回滚消息
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        
        /**
         * 事务回查
         * 
         * 当Broker长时间未收到Commit/Rollback时，会主动回查。
         * 这里需要根据业务数据判断事务状态。
         */
        @Override
        public LocalTransactionState checkLocalTransaction(Message msg) {
            String orderId = msg.getHeader("orderId");
            Order order = orderMapper.selectById(orderId);
            
            if (order != null) {
                // 订单存在，说明本地事务已提交
                return LocalTransactionState.COMMIT_MESSAGE;
            }
            // 订单不存在，事务已回滚
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
}

方案三：Saga模式

Saga 模式适合长事务场景，它把一个分布式事务拆分为一系列本地事务，每个本地事务都有对应的补偿操作。

#### 核心流程

graph LR A[开始] --> B[T1: 创建订单] B --> C[T2: 扣减库存] C --> D[T3: 锁定优惠券] D --> E[成功] B -. 补偿C1 .-> B1[撤销订单] C -. 补偿C2 .-> C1[回补库存] D -. 补偿C3 .-> D1[解锁优惠券]

#### 核心源码（基于Spring Boot + Saga框架）

/**
 * Saga模式 - 订单创建流程
 * 
 * 使用Seata的Saga模式实现（也可以用Camel或自研）
 * 
 * 关键设计点：
 * 1. 每个步骤都是独立的本地事务
 * 2. 每个步骤都有对应的补偿操作
 * 3. 通过状态机管理整个Saga流程
 */
@Service
public class SagaOrderService {
    
    @Autowired
    private OrderClient orderClient;
    
    @Autowired
    private InventoryClient inventoryClient;
    
    @Autowired
    private CouponClient couponClient;
    
    @Autowired
    private SagaStateMachine stateMachine;
    
    /**
     * 使用Saga状态机定义业务流程
     */
    public void createOrderWithSaga(OrderRequest request) {
        // 定义Saga流程
        SagaDefinition saga = new SagaDefinition()
            // 步骤1：创建订单（正向操作）
            .addStep(
                // 正向操作
                () -> orderClient.createOrder(request),
                // 补偿操作
                (orderId) -> orderClient.cancelOrder(orderId)
            )
            // 步骤2：扣减库存
            .addStep(
                () -> inventoryClient.deductStock(request.getProductId(), 
                                                  request.getQuantity()),
                (result) -> inventoryClient.increaseStock(request.getProductId(), 
                                                          request.getQuantity())
            )
            // 步骤3：锁定优惠券
            .addStep(
                () -> couponClient.lockCoupon(request.getCouponId(), 
                                              request.getUserId()),
                (result) -> couponClient.unlockCoupon(request.getCouponId())
            );
        
        // 执行Saga流程
        try {
            stateMachine.execute(saga);
        } catch (SagaException e) {
            log.error("Saga执行失败，已自动执行补偿", e);
            // 这里可以发起告警，或者记录失败状态等待人工处理
            throw new BusinessException("订单创建失败，已回滚");
        }
    }
}

#### 基于状态机的Saga实现（更可控）

/**
 * 状态机驱动的Saga实现
 * 
 * 比顺序编排更灵活，支持条件分支、并行执行、人工介入等高级特性
 */
@Component
public class OrderSagaStateMachine {
    
    private final StateMachine<OrderState, OrderEvent> stateMachine;
    
    public OrderSagaStateMachine() {
        // 构建状态机
        StateMachineBuilder.Builder<OrderState, OrderEvent> builder = 
            StateMachineBuilder.builder();
        
        // 定义状态转换和动作
        builder.external()
            .from(OrderState.INIT)
            .to(OrderState.ORDER_CREATED)
            .event(OrderEvent.CREATE_ORDER)
            .action(createOrderAction(), cancelOrderAction());
        
        builder.external()
            .from(OrderState.ORDER_CREATED)
            .to(OrderState.STOCK_DEDUCTED)
            .event(OrderEvent.DEDUCT_STOCK)
            .action(deductStockAction(), increaseStockAction());
        
        builder.external()
            .from(OrderState.STOCK_DEDUCTED)
            .to(OrderState.COUPON_LOCKED)
            .event(OrderEvent.LOCK_COUPON)
            .action(lockCouponAction(), unlockCouponAction());
        
        builder.external()
            .from(OrderState.COUPON_LOCKED)
            .to(OrderState.COMPLETED)
            .event(OrderEvent.COMPLETE)
            .action(completeOrderAction());
        
        this.stateMachine = builder.build();
    }
    
    /**
     * 执行状态机
     */
    public void execute(OrderContext context) {
        // 开始执行，如果中途失败，状态机会自动触发补偿
        StateMachine<OrderState, OrderEvent> sm = 
            stateMachine.startWith(OrderState.INIT);
        
        sm.sendEvent(OrderEvent.CREATE_ORDER, context);
        sm.sendEvent(OrderEvent.DEDUCT_STOCK, context);
        sm.sendEvent(OrderEvent.LOCK_COUPON, context);
        sm.sendEvent(OrderEvent.COMPLETE, context);
    }
}

方案对比

| 维度 | 本地消息表 | 事务消息（RocketMQ） | Saga模式 |

|------|-----------|---------------------|----------|

| 实现复杂度 | 中等，需要额外维护消息表 | 较低，依赖MQ的能力 | 较高，需要设计状态机和补偿 |

| 一致性保证 | 最终一致，有秒级延迟 | 最终一致，延迟更低 | 最终一致，可精确控制 |

| 适用场景 | 中小规模系统，对延迟不敏感 | 已经使用RocketMQ的团队 | 长事务、复杂业务流程 |

| 侵入性 | 需要修改业务代码 | 需要引入RocketMQ | 需要引入Saga框架 |

| 运维成本 | 需要维护定时任务 | 依赖MQ的高可用 | 需要监控Saga执行状态 |

| 隔离性 | 支持 | 支持 | 支持 |

| 幂等性 | 需要消费端自行保证 | 需要消费端自行保证 | 需要每个步骤自行保证 |

各方案优缺点详解

本地消息表

• 优点：不依赖特定中间件，任何MQ都能用；实现思路简单清晰

• 缺点：消息表会和业务表竞争数据库资源；定时任务有延迟；消息表膨胀后需要归档

事务消息

• 优点：RocketMQ解决了消息发送和本地事务的原子性问题；延迟更低；不需要定时任务

• 缺点：强依赖RocketMQ（目前只有RocketMQ支持）；RocketMQ的Broker需要额外配置

Saga模式

• 优点：适合长事务和跨多服务的复杂流程；可以精细控制每个步骤的补偿逻辑

• 缺点：实现复杂度最高；需要设计完整的补偿逻辑（有时补偿比正向逻辑还难写）；没有隔离性保证

最佳实践与避坑指南

最佳实践

1. 优先选择事务消息

如果你的团队已经在使用RocketMQ，事务消息是最优解。它同时解决了原子性和延迟问题，实现成本最低。

1. 幂等性是必须的

无论使用哪种方案，消费端幂等都是硬性要求。推荐使用唯一业务键 + 去重表的方式，比Redis SETNX更可靠。

1. 设置合理的重试策略

• 消息重试间隔：指数退避（1s → 2s → 4s ...）

• 最大重试次数：10-15次

• 超过重试次数：进入死信队列，触发告警

1. 监控和告警

• 监控消息积压量、消费延迟、死信队列数量

• 设置告警阈值，如：消息延迟超过5分钟、死信队列积压超过100条

1. 人工补偿通道

即使自动化方案再完善，也要保留人工干预的通道。提供管理后台，可以手动重发消息、手动执行补偿。

常见坑

1. 消息重复消费（幂等没做好）

• 症状：库存扣成负数、优惠券被重复锁定

• 解决方案：去重表 + 唯一索引，或者用业务ID做幂等

1. 消息顺序问题

• 症状：订单取消消息先于创建消息到达

• 解决方案：使用分区顺序消息，同一个订单ID发到同一个分区

1. 补偿操作不幂等

• 症状：补偿操作执行两次，导致数据错乱

• 解决方案：补偿操作也要做幂等，比如用补偿记录表

1. 死循环重试

• 症状：业务逻辑有问题，消费一直失败，消息无限重试

• 解决方案：设置最大重试次数，超过后进入死信队列，人工介入

1. 事务超时

• 症状：本地事务执行时间过长，超过MQ回查超时时间

• 解决方案：合理设置超时时间（建议30秒），不要在事务中调用远程服务

总结

分布式系统的事务一致性是一个"没有银弹"的问题。我在实际项目中经历过从"强一致"到"最终一致"的思维转变，这个转变对架构设计的影响是深远的。

回顾三种方案：

• 本地消息表是最朴素的思想，用"空间换时间"，适合对延迟不敏感的场景

• 事务消息是中间件层面的创新，用"消息代理"解决原子性问题，是当前的最佳性价比选择

• Saga模式是流程编排的思路，适合复杂的长事务场景，但需要精心的补偿设计

最后想说的是：不要为了分布式事务而分布式事务。如果你的业务可以接受将多个操作放在同一个服务中，用本地事务就够了。微服务拆分是有代价的，每个跨服务操作都需要考虑一致性方案，这会显著增加系统的复杂度。

在实际架构设计中，我通常遵循这样的决策路径：

1. 能用本地事务解决的，不要拆服务
2. 必须拆服务的，优先考虑事务消息
3. 事务消息搞不定的（如长事务），才考虑Saga

希望这篇文章能帮你在分布式系统的道路上少踩一些坑。如果你有更好的方案或不同的见解，欢迎在评论区讨论。