目錄

一、引言

二、CAP理論與BASE理論

三、一致性失效場(chǎng)景及其解決方案

    1. 調(diào)用寫(xiě)RPC

    2. 消息發(fā)送

    3. 本地消息表

四、總結(jié)

一、引言

在經(jīng)典的CAP理論中一致性是指分布式或多副本系統(tǒng)中數(shù)據(jù)在任一時(shí)刻均保持邏輯與物理狀態(tài)的統(tǒng)一，這是確保業(yè)務(wù)邏輯正確性和系統(tǒng)可靠性的核心要素。在單體應(yīng)用單一數(shù)據(jù)庫(kù)中可以直接通過(guò)本地事務(wù)(ACID)保證數(shù)據(jù)的強(qiáng)一致性。

然而隨著微服務(wù)架構(gòu)的普及和業(yè)務(wù)場(chǎng)景的復(fù)雜化，原來(lái)的原子性操作會(huì)隨著系統(tǒng)拆分而無(wú)法保障原子性從而產(chǎn)生一致性問(wèn)題，但業(yè)務(wù)實(shí)際又需要保障一致性，為此BASE理論提出了最終一致性來(lái)解決這類(lèi)問(wèn)題。那么如何在跨服務(wù)、跨數(shù)據(jù)庫(kù)的事務(wù)中保證數(shù)據(jù)最終一致性。

二、CAP理論與BASE理論

在經(jīng)典的CAP理論中提到一個(gè)分布式系統(tǒng)中，一致性(C)、可用性(A)、分區(qū)容錯(cuò)性(P)最多只能同時(shí)實(shí)現(xiàn)兩點(diǎn)，不可能三者兼顧。實(shí)際上這是一個(gè)偽命題，必須從 A 和 C 選擇一個(gè)和 P 組合，更進(jìn)一步基本上都會(huì)選擇 A，相比一致性，系統(tǒng)一旦不可用或不可靠都可能會(huì)造成整個(gè)站點(diǎn)崩潰，所以一般都會(huì)選擇 AP。

圖片

BASE理論源于對(duì)大規(guī)?；ヂ?lián)網(wǎng)分布式系統(tǒng)實(shí)踐的總結(jié)，作為CAP定理中一致性與可用性矛盾的實(shí)踐性補(bǔ)充逐步演化形成。該理論主張?jiān)跓o(wú)法保證強(qiáng)一致性的場(chǎng)景下，系統(tǒng)可基于業(yè)務(wù)特性靈活調(diào)整架構(gòu)設(shè)計(jì)，通過(guò)基本可用性保障、允許短暫中間狀態(tài)等機(jī)制，確保數(shù)據(jù)最終達(dá)成一致性狀態(tài)，從而在分布式環(huán)境中實(shí)現(xiàn)可靠服務(wù)能力與業(yè)務(wù)需求的平衡。

三、一致性失效場(chǎng)景及其解決方案

這里有一個(gè)簡(jiǎn)化的倉(cāng)庫(kù)上架的流程（在實(shí)際業(yè)務(wù)中可能還會(huì)涉及到履約，倉(cāng)儲(chǔ)庫(kù)存等等），體現(xiàn)分布式系統(tǒng)中可能出現(xiàn)的一致性問(wèn)題，在分布式系統(tǒng)中的處理流程可能如下所示：

操作員操作商品倉(cāng)庫(kù)上架 

   商品在倉(cāng)儲(chǔ)系統(tǒng)(WMS)中上架，寫(xiě)入倉(cāng)儲(chǔ)數(shù)據(jù)庫(kù) 

   倉(cāng)儲(chǔ)系統(tǒng)通知中央庫(kù)存系統(tǒng)(SCI)添加可用庫(kù)存 

   倉(cāng)儲(chǔ)系統(tǒng)通知交易該商品可以進(jìn)行售賣(mài)

多服務(wù)協(xié)作交互示例

簡(jiǎn)化代碼示例：

@Transactional
public void upper(upperRequest request) {


    // 1. 寫(xiě)入倉(cāng)儲(chǔ)數(shù)據(jù)庫(kù)
    UpperDo upperDo = buildUpperDo(request);
    wmsService.upper(upperDo);


    // 2. 調(diào)用rpc添加中央庫(kù)存系統(tǒng)庫(kù)存
    SciAInventoryRequest sciInventoryRequest = buildSciAInventoryRequest(request);
    sciRpcService.addInventory(sciInventoryRequest)


    // 3. 發(fā)送商品可以售賣(mài)的消息
    TradeMessageRequest tradeMessage = buildTradeMessageRequest(request);
    sendMessageToDealings(tradeMessage);


    // 4. 其他處理
    recordLog(buildLogRequest(request))
    return;
}

整個(gè)時(shí)序邏輯拆解到事務(wù)層面執(zhí)行流程如下：

發(fā)送消息

在第5步添加sci庫(kù)存之前任意一步出現(xiàn)問(wèn)題，事務(wù)都會(huì)回滾，對(duì)其他系統(tǒng)的影響為0，所以不存在一致性問(wèn)題。

但是，在此之后出現(xiàn)問(wèn)題都有可能會(huì)出現(xiàn)事務(wù)問(wèn)題。

調(diào)用寫(xiě)RPC

在分布式系統(tǒng)中，調(diào)用RPC一般可以分為著兩類(lèi)： 

1.讀RPC：當(dāng)前數(shù)據(jù)結(jié)構(gòu)不完整，需要通過(guò)其他服務(wù)補(bǔ)充數(shù)據(jù)，對(duì)其他服務(wù)無(wú)影響。 

2.寫(xiě)RPC：當(dāng)前業(yè)務(wù)操作、數(shù)據(jù)變更需要通知其他服務(wù)，對(duì)其他服務(wù)有影響。

調(diào)用寫(xiě)RPC添加sci可用庫(kù)存可能出現(xiàn)的問(wèn)題：

• 調(diào)用處理成功，返回成功?！緮?shù)據(jù)一致】
• 調(diào)用處理成功，返回失敗?！緮?shù)據(jù)不一致】

對(duì)于這種情況，最簡(jiǎn)單的做法是直接操作重試，但是需要下游冪等處理，保證同樣的請(qǐng)求效果一致。這里重試的方式，即重新操作上架，此外也可以直接在rpc方法中異步重試機(jī)制（這種方式不會(huì)阻塞整體流程，但是增大了數(shù)據(jù)不一致的風(fēng)險(xiǎn)）。如果重試失敗可能需要研發(fā)介入排查具體失敗的原因（對(duì)于寫(xiě)RPC的接口超時(shí)問(wèn)題，需要研發(fā)關(guān)注，配置告警或拋出特定異常等）。

針對(duì)RPC方法重試，可以考慮采用本地消息表的方式實(shí)現(xiàn)，具體參考3.3.本地消息表。

消息發(fā)送

寫(xiě)RPC調(diào)用成功后，會(huì)給trade服務(wù)發(fā)送消息，而后提交事務(wù)，整個(gè)流程結(jié)束。

Rocket消息發(fā)送有多種方式，不同的方式適用場(chǎng)景不一，一般業(yè)務(wù)邏輯使用同步發(fā)送消息配合重試機(jī)制即可，對(duì)于一致性要求高的場(chǎng)景，可以考慮事務(wù)消息確保消息與本地事務(wù)的原子性。

圖片

同步消息+重試

同步消息比異步消息更可靠，比事務(wù)消息性能更高是一種廣泛采用的方式。

同步消息通過(guò)confirm機(jī)制能保證消息發(fā)送成功：生產(chǎn)者發(fā)送同步消息后，等待Broker返回確認(rèn)結(jié)果（SendResult）。如果 Broker 成功接收并存儲(chǔ)消息，返回成功狀態(tài)；否則返回失敗狀態(tài)。消息發(fā)送失敗時(shí)，Rocket默認(rèn)自動(dòng)重試2次，支持手動(dòng)設(shè)置，提高消息發(fā)送的可靠性。

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup");
producer.setRetryTimesWhenSendFailed(3); // 設(shè)置重試次數(shù)為 3 次
producer.start();
Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
SendResult sendResult = producer.send(msg); // 同步發(fā)送
if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
    log.info("Send Success: " + sendResult);
} else {
    log.warn("Send Failed: " + sendResult);
}

同步消息+重試機(jī)制能盡可能的保證消息成功發(fā)送，但是在這種情況下仍可能出現(xiàn)一致性問(wèn)題：消息成功發(fā)送，在提交事務(wù)之前，依然可能出現(xiàn)問(wèn)題（第8步出現(xiàn)問(wèn)題），導(dǎo)致事務(wù)回滾，但是下游的消息是無(wú)法回滾的。

為此在RocketMQ中提供了事務(wù)消息作為一種解決方案。

RocketMQ事務(wù)消息

RocketMQ 的分布式事務(wù)消息功能，在普通消息基礎(chǔ)上，支持二階段的提交能力。將二階段提交和本地事務(wù)綁定，實(shí)現(xiàn)全局提交結(jié)果的一致性。

發(fā)送事物消息

Rocket的事務(wù)消息可以確保消息和本地事務(wù)的原子性，但是實(shí)現(xiàn)起來(lái)很復(fù)雜，性能也比較低，特別是需要實(shí)現(xiàn)回查本地事務(wù)狀態(tài)，這是一個(gè)比較復(fù)雜的問(wèn)題，需要case by case，每一個(gè)消息都需要單獨(dú)寫(xiě)邏輯，還必須確保消息體中的數(shù)據(jù)支持回查本地事務(wù)狀態(tài)，對(duì)代碼入侵度較高。

在筆者的了解中我司事務(wù)消息的使用情況不多，對(duì)于低并發(fā)且強(qiáng)一致性的場(chǎng)景可以考慮使用這種方式。在這個(gè)業(yè)務(wù)場(chǎng)景中使用事務(wù)消息可以解決3.2.1中出現(xiàn)的消息發(fā)送成功但事務(wù)回滾的問(wèn)題，但是這個(gè)場(chǎng)景使用這種方式并不太合適。最終結(jié)果可能是整體數(shù)據(jù)一致性提升2%-3%，但是業(yè)務(wù)性能下降20%-30%。

spring提供給了一種事件發(fā)布-訂閱機(jī)制可以解決事務(wù)回滾但消息依然發(fā)送成功的問(wèn)題，并且性能損失幾乎可以忽略。

事務(wù)事件+同步消息

事務(wù)事件是指在事務(wù)執(zhí)行的不同階段觸發(fā)的事件。這些事件通常用于處理次要邏輯，例如發(fā)送領(lǐng)域事件、消息或者郵件等。

spring通過(guò)事務(wù)管理@Transactional和事件發(fā)布機(jī)制ApplicationEventPublisher，可以實(shí)現(xiàn)類(lèi)似事務(wù)事件的功能。事件發(fā)布后事件廣播器（SimpleApplicationEventMulticaster）接收事件，根據(jù)事件類(lèi)型匹配所有的監(jiān)聽(tīng)者（getApplicationListeners）。

@Service
public class wmsService {
    @Autowired
    private ApplicationEventPublisher eventPublisher;


    @Transactional
    public void upper(upperRequest request) {


        // 1. 寫(xiě)入倉(cāng)儲(chǔ)數(shù)據(jù)庫(kù)
        UpperDo upperDo = buildUpperDo(request);
        wmsService.upper(upperDo);


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


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


        // 1. 調(diào)用rpc添加中央庫(kù)存系統(tǒng)庫(kù)存
        SciAInventoryRequest sciInventoryRequest = buildSciAInventoryRequest(event);
        sciRpcService.addInventory(sciInventoryRequest)




        // 2. 發(fā)送商品可以售賣(mài)的消息
        TradeMessageRequest tradeMessage = buildTradeMessageRequest(event);
        sendMessageToDealings(tradeMessage);


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

上述流程在寫(xiě)完DB，調(diào)用寫(xiě)RPC之后，發(fā)布上架完成的事件并提交事務(wù)。upperFinishEventListener訂閱上架完成的事件，并發(fā)送可以售賣(mài)的消息。

通過(guò)這種方式可以在事務(wù)提交之后再發(fā)送消息。通過(guò)事務(wù)事件保證事務(wù)提交，通過(guò)重試機(jī)制和confirm機(jī)制確保生產(chǎn)者發(fā)送消息成功。

本地消息表

在上述過(guò)程中我們選擇使用事務(wù)事件+同步消息可以來(lái)替代事務(wù)消息，但是事務(wù)事件對(duì)RPC調(diào)用并不太友好，本地事務(wù)提交之后，調(diào)用寫(xiě)RPC就一定要成功，不然一致性問(wèn)題就無(wú)法保證。

為此可以考慮使用本地消息表這個(gè)方案：將需要分布式處理的事件通過(guò)本地消息日志存儲(chǔ)的方式來(lái)異步執(zhí)行，通過(guò)異步線程或者自動(dòng)Job發(fā)起重試，確保上下游一致。

圖片

將上述流程抽象為代碼可以實(shí)現(xiàn)一個(gè)一致性框架，通過(guò)注解實(shí)現(xiàn)無(wú)侵入、策略化、通用性和高復(fù)用性的能力。然后本地消息表的方式仍然存在一些問(wèn)題：

• 高并發(fā)場(chǎng)景不適用，寫(xiě)本地消息會(huì)帶來(lái)延遲可能出現(xiàn)數(shù)據(jù)積壓，影響系統(tǒng)的吞吐量。
• 業(yè)務(wù)邏輯過(guò)程會(huì)長(zhǎng)時(shí)間的占用事務(wù)，造成大事務(wù)問(wèn)題。
• 本地消息報(bào)文巨大，難以存儲(chǔ)等。

四、總結(jié)

本文分析的場(chǎng)景都是解決生產(chǎn)者端的一致性問(wèn)題。結(jié)合部分場(chǎng)景探討不同方式的優(yōu)缺點(diǎn)。

1. 事務(wù)事件+普通消息&重試 ：適合對(duì)實(shí)時(shí)一致性要求不高、需要異步處理的場(chǎng)景、適合高并發(fā)場(chǎng)景，可靠性一般，實(shí)現(xiàn)簡(jiǎn)單但需手動(dòng)處理重試和冪等性。
2. 事務(wù)消息 ：適合一致性要求較高的場(chǎng)景（如金融交易），性能較低，實(shí)現(xiàn)復(fù)雜但能確保消息與事務(wù)的原子性。
3. 本地消息表 ：適合跨服務(wù)事務(wù)、異步任務(wù)處理和最終一致性場(chǎng)景，高并發(fā)場(chǎng)景可能出現(xiàn)數(shù)據(jù)積壓，實(shí)現(xiàn)簡(jiǎn)單且可靠性高，但存在延遲性和資源占用問(wèn)題。

在分布式系統(tǒng)中，很難有能100%保證一致性的方案，正如《人月神話(huà)》中說(shuō)的“沒(méi)有不存在缺陷的軟件，只是尚未發(fā)現(xiàn)缺陷”。

在上面提到的各種方案中，筆者所在團(tuán)隊(duì)高并發(fā)場(chǎng)景很少，所以一般都采用本地詳細(xì)表的方式來(lái)處理一致性問(wèn)題，這既可以處理寫(xiě)RPC的調(diào)用問(wèn)題，也能通過(guò)消息狀態(tài)顯示的統(tǒng)一失敗情況，統(tǒng)一進(jìn)行重試。