自以为掌握的分布式数据一致性问题，面试还是被问懵了……

dbaplus社群 07月16日 23:25

本文探讨了在分布式系统中保障数据一致性的挑战，分析了CAP理论与BASE理论，并结合实际场景，详细阐述了多种解决一致性失效问题的方案，包括调用写RPC、消息发送和本地消息表等。文章深入分析了不同方案的优缺点，为开发者提供了在不同业务场景下选择合适解决方案的参考。

💡CAP理论指出在分布式系统中，一致性(C)、可用性(A)和分区容错性(P)最多只能同时满足两个，而BASE理论则强调在无法保证强一致性的情况下，通过基本可用性、允许短暂中间状态等机制来实现最终一致性。

📢调用写RPC时，若出现成功或失败的情况，可采用重试机制，并结合幂等处理来确保数据一致性。对于RPC接口超时问题，应配置告警或抛出特定异常。

✉️消息发送方面，同步消息配合重试机制是常用的方式。对于一致性要求高的场景，RocketMQ事务消息可以确保消息与本地事务的原子性，但实现复杂且性能较低；事务事件+同步消息则是一种更优选择，能在事务提交后发送消息，且性能损失较小。

💾本地消息表方案通过将分布式处理事件以本地消息日志存储，异步重试，确保上下游一致性。虽然实现简单且可靠性高，但在高并发场景下可能出现数据积压和延迟问题。

勇者 2025-07-11 07:16 广东

结合部分场景探讨不同方式的优缺点。

分享概要

一、引言

二、CAP理论与BASE理论

三、一致性失效场景及其解决方案

1、调用写RPC

2、消息发送

3、本地消息表

四、总结

一、引言

在经典的CAP理论中一致性是指分布式或多副本系统中数据在任一时刻均保持逻辑与物理状态的统一，这是确保业务逻辑正确性和系统可靠性的核心要素。在单体应用单一数据库中可以直接通过本地事务(ACID)保证数据的强一致性。

然而随着微服务架构的普及和业务场景的复杂化，原来的原子性操作会随着系统拆分而无法保障原子性从而产生一致性问题，但业务实际又需要保障一致性，为此BASE理论提出了最终一致性来解决这类问题。那么如何在跨服务、跨数据库的事务中保证数据最终一致性。

二、CAP理论与BASE理论

在经典的CAP理论中提到一个分布式系统中，一致性(C)、可用性(A)、分区容错性(P)最多只能同时实现两点，不可能三者兼顾。实际上这是一个伪命题，必须从 A 和 C 选择一个和 P 组合，更进一步基本上都会选择 A，相比一致性，系统一旦不可用或不可靠都可能会造成整个站点崩溃，所以一般都会选择 AP。

BASE理论源于对大规模互联网分布式系统实践的总结，作为CAP定理中一致性与可用性矛盾的实践性补充逐步演化形成。该理论主张在无法保证强一致性的场景下，系统可基于业务特性灵活调整架构设计，通过基本可用性保障、允许短暂中间状态等机制，确保数据最终达成一致性状态，从而在分布式环境中实现可靠服务能力与业务需求的平衡。

三、一致性失效场景及其解决方案

这里有一个简化的仓库上架的流程（在实际业务中可能还会涉及到履约，仓储库存等等），体现分布式系统中可能出现的一致性问题，在分布式系统中的处理流程可能如下所示：

操作员操作商品仓库上架

商品在仓储系统(WMS)中上架，写入仓储数据库

仓储系统通知中央库存系统(SCI)添加可用库存

仓储系统通知交易该商品可以进行售卖

多服务协作交互示例

简化代码示例：

@Transactional
public void upper(upperRequest request) {
    // 1. 写入仓储数据库
    UpperDo upperDo = buildUpperDo(request);
    wmsService.upper(upperDo);
    // 2. 调用rpc添加中央库存系统库存
    SciAInventoryRequest sciInventoryRequest = buildSciAInventoryRequest(request);
    sciRpcService.addInventory(sciInventoryRequest)
    // 3. 发送商品可以售卖的消息
    TradeMessageRequest tradeMessage = buildTradeMessageRequest(request);
    sendMessageToDealings(tradeMessage);
    // 4. 其他处理
    recordLog(buildLogRequest(request))
    return;
}

整个时序逻辑拆解到事务层面执行流程如下：

发送消息

在第5步添加sci库存之前任意一步出现问题，事务都会回滚，对其他系统的影响为0，所以不存在一致性问题。

但是，在此之后出现问题都有可能会出现事务问题。

1、调用写RPC

在分布式系统中，调用RPC一般可以分为着两类：

1）读RPC：当前数据结构不完整，需要通过其他服务补充数据，对其他服务无影响。

2）写RPC：当前业务操作、数据变更需要通知其他服务，对其他服务有影响。

调用写RPC添加sci可用库存可能出现的问题：

调用处理成功，返回成功。【数据一致】

调用处理成功，返回失败。【数据不一致】

对于这种情况，最简单的做法是直接操作重试，但是需要下游幂等处理，保证同样的请求效果一致。这里重试的方式，即重新操作上架，此外也可以直接在rpc方法中异步重试机制（这种方式不会阻塞整体流程，但是增大了数据不一致的风险）。如果重试失败可能需要研发介入排查具体失败的原因（对于写RPC的接口超时问题，需要研发关注，配置告警或抛出特定异常等）。

针对RPC方法重试，可以考虑采用本地消息表的方式实现，具体参考3.3.本地消息表。

2、消息发送

写RPC调用成功后，会给trade服务发送消息，而后提交事务，整个流程结束。

Rocket消息发送有多种方式，不同的方式适用场景不一，一般业务逻辑使用同步发送消息配合重试机制即可，对于一致性要求高的场景，可以考虑事务消息确保消息与本地事务的原子性。

1）同步消息+重试

同步消息比异步消息更可靠，比事务消息性能更高是一种广泛采用的方式。

同步消息通过confirm机制能保证消息发送成功：生产者发送同步消息后，等待Broker返回确认结果（SendResult）。如果 Broker 成功接收并存储消息，返回成功状态；否则返回失败状态。消息发送失败时，Rocket默认自动重试2次，支持手动设置，提高消息发送的可靠性。

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup");
producer.setRetryTimesWhenSendFailed(3); // 设置重试次数为 3 次
producer.start();
Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
SendResult sendResult = producer.send(msg); // 同步发送
if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
    log.info("Send Success: " + sendResult);
} else {
    log.warn("Send Failed: " + sendResult);
}

同步消息+重试机制能尽可能的保证消息成功发送，但是在这种情况下仍可能出现一致性问题：消息成功发送，在提交事务之前，依然可能出现问题（第8步出现问题），导致事务回滚，但是下游的消息是无法回滚的。

为此在RocketMQ中提供了事务消息作为一种解决方案。

2）RocketMQ事务消息

RocketMQ 的分布式事务消息功能，在普通消息基础上，支持二阶段的提交能力。将二阶段提交和本地事务绑定，实现全局提交结果的一致性。

发送事物消息

Rocket的事务消息可以确保消息和本地事务的原子性，但是实现起来很复杂，性能也比较低，特别是需要实现回查本地事务状态，这是一个比较复杂的问题，需要case by case，每一个消息都需要单独写逻辑，还必须确保消息体中的数据支持回查本地事务状态，对代码入侵度较高。

在笔者的了解中我司事务消息的使用情况不多，对于低并发且强一致性的场景可以考虑使用这种方式。在这个业务场景中使用事务消息可以解决3.2.1中出现的消息发送成功但事务回滚的问题，但是这个场景使用这种方式并不太合适。最终结果可能是整体数据一致性提升2%-3%，但是业务性能下降20%-30%。

spring提供给了一种事件发布-订阅机制可以解决事务回滚但消息依然发送成功的问题，并且性能损失几乎可以忽略。

3）事务事件+同步消息

事务事件是指在事务执行的不同阶段触发的事件。这些事件通常用于处理次要逻辑，例如发送领域事件、消息或者邮件等。

spring通过事务管理@Transactional和事件发布机制ApplicationEventPublisher，可以实现类似事务事件的功能。事件发布后事件广播器（SimpleApplicationEventMulticaster）接收事件，根据事件类型匹配所有的监听者（getApplicationListeners）。

@Service
public class wmsService {
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    @Transactional
    public void upper(upperRequest request) {


        // 1. 写入仓储数据库
        UpperDo upperDo = buildUpperDo(request);
        wmsService.upper(upperDo);


        // 3. 发布上架事件
        UpperFinishEvent upperFinishEvent = buildUpperFinishEvent(request)
        eventPublisher.publishEvent(upperFinishEvent);
        return;
    }
}


@Component
public class upperFinishEventListener {
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleUpperFinishEvent(UpperFinishEvent event) {
        // 处理事件


        // 1. 调用rpc添加中央库存系统库存
        SciAInventoryRequest sciInventoryRequest = buildSciAInventoryRequest(event);
        sciRpcService.addInventory(sciInventoryRequest)


        // 2. 发送商品可以售卖的消息
        TradeMessageRequest tradeMessage = buildTradeMessageRequest(event);
        sendMessageToDealings(tradeMessage);


        // 2. 其他处理
        recordLog(buildLogRequest(event))
    }
}