摘要：一小小推廣講座本話題已收入視頻講座分布式事務(wù)解決方案大家不妨圍觀下開源項(xiàng)目我們利用消息隊(duì)列實(shí)現(xiàn)了分布式事務(wù)的最終一致性解決方案，請大家圍觀。

講座

本話題已收入視頻講座《Spring Cloud分布式事務(wù)解決方案》大家不妨圍觀下

開源項(xiàng)目

我們利用消息隊(duì)列實(shí)現(xiàn)了分布式事務(wù)的最終一致性解決方案，請大家圍觀。可以參考Github CoolMQ源碼，項(xiàng)目支持網(wǎng)站: http://rabbitmq.org.cn，最新文章或?qū)崿F(xiàn)會更新在上面

阿里2017云棲大會《破解世界性技術(shù)難題！GTS讓分布式事務(wù)簡單高效》中，阿里聲稱提出了一種破解世界性難題之分布式事務(wù)的終極解決方案，無論是可靠性、還是處理速率都領(lǐng)先于市面上所有的技術(shù)。但令人遺憾的是一來項(xiàng)目未開源，二來還必須依賴阿里云的分布式數(shù)據(jù)庫。畢竟，吃飯的家伙可不能輕易示人嘛。

雖然如此，但《世界難題...》一文中對事務(wù)還是歸納的還是蠻到位的:“一個(gè)看似簡單的功能，內(nèi)部可能需要調(diào)用多個(gè)“服務(wù)”并操作多個(gè)數(shù)據(jù)庫或分片來實(shí)現(xiàn)，單一技術(shù)手段和解決方案已無法滿足這些復(fù)雜應(yīng)用場景。因此，分布式系統(tǒng)架構(gòu)中分布式事務(wù)是一個(gè)繞不過去的挑戰(zhàn)。

什么是分布式事務(wù)？簡單的說，就是一次大操作由不同小操作組成，這些小操作分布在不同服務(wù)器上，分布式事務(wù)需要保證這些小操作要么全部成功，要么全部失敗?！?/p>

舉個(gè)栗子：

你上Taobao買東西，需要先扣錢，然后商品庫存-1吧。但扣款和庫存分別屬于兩個(gè)服務(wù)，這兩個(gè)服務(wù)中間要經(jīng)過網(wǎng)絡(luò)、網(wǎng)關(guān)、主機(jī)等一系列中間層，萬一任何一個(gè)地方出了問題，比如網(wǎng)絡(luò)抖動、突發(fā)異常等待，都會導(dǎo)致不一致，比如扣款成功了，但是庫存沒-1，就會出現(xiàn)超賣的現(xiàn)象，而這就是分布式事務(wù)需要解決的問題

2階段提交是分布式事務(wù)傳統(tǒng)解決方案，先進(jìn)為止還廣泛存在。當(dāng)一個(gè)事務(wù)跨越多個(gè)節(jié)點(diǎn)時(shí)，為了保持事務(wù)ACID特性，需要引入一個(gè)作為協(xié)調(diào)者來統(tǒng)一掌控所有節(jié)點(diǎn)(稱作參與者)的操作結(jié)果并最終指示這些節(jié)點(diǎn)是否要把操作結(jié)果進(jìn)行真正的提交(比如將更新后的數(shù)據(jù)寫入磁盤等等)。因此，二階段提交的算法思路可以概括為：參與者將操作成敗通知協(xié)調(diào)者，再由協(xié)調(diào)者根據(jù)所有參與者的反饋情報(bào)決定各參與者是否要提交操作還是中止操作。

以開會為例

甲乙丙丁四人要組織一個(gè)會議，需要確定會議時(shí)間，不妨設(shè)甲是協(xié)調(diào)者，乙丙丁是參與者。

投票階段

甲發(fā)郵件給乙丙丁，周二十點(diǎn)開會是否有時(shí)間；

甲回復(fù)有時(shí)間；

乙回復(fù)有時(shí)間；

丙遲遲不回復(fù)，此時(shí)對于這個(gè)活動，甲乙丙均處于阻塞狀態(tài)，算法無法繼續(xù)進(jìn)行；

丙回復(fù)有時(shí)間（或者沒有時(shí)間）；

提交階段

協(xié)調(diào)者甲將收集到的結(jié)果反饋給乙丙?。ㄊ裁磿r(shí)候反饋，以及反饋結(jié)果如何，在此例中取決與丙的時(shí)間與決定）；

乙收到；

丙收到；

丁收到；

不僅要鎖住參與者的所有資源，而且要鎖住協(xié)調(diào)者資源，開銷大。一句話總結(jié)就是：2PC效率很低，對高并發(fā)很不友好。

引用《世界性難題...》一文原話 "國外具有幾十年歷史和技術(shù)沉淀的基于XA模型的商用分布式事務(wù)產(chǎn)品，在相同軟硬件條件下，開啟分布式事務(wù)后吞吐經(jīng)常有數(shù)量級的下降。"

此外還有三階段提交

大家有興趣的不妨研究下

所謂柔性事務(wù)是相對強(qiáng)制鎖表的剛性事務(wù)而言。流程入下：服務(wù)器A的事務(wù)如果執(zhí)行順利，那么事務(wù)A就先行提交，如果事務(wù)B也執(zhí)行順利，則事務(wù)B也提交，整個(gè)事務(wù)就算完成。但是如果事務(wù)B執(zhí)行失敗，事務(wù)B本身回滾，這時(shí)事務(wù)A已經(jīng)被提交，所以需要執(zhí)行一個(gè)補(bǔ)償操作，將已經(jīng)提交的事務(wù)A執(zhí)行的操作作反操作，恢復(fù)到未執(zhí)行前事務(wù)A的狀態(tài)。

缺點(diǎn)是業(yè)務(wù)侵入性太強(qiáng)，還要補(bǔ)償操作，缺乏普遍性，沒法大規(guī)模推廣。

目前基于消息隊(duì)列的解決方案有阿里的RocketMQ，它實(shí)現(xiàn)了半消息的解決方案,有點(diǎn)類似于Paxos算法，具體流程如下

第一階段：上游應(yīng)用執(zhí)行業(yè)務(wù)并發(fā)送MQ消息

上游應(yīng)用發(fā)送待確認(rèn)消息到可靠消息系統(tǒng)

可靠消息系統(tǒng)保存待確認(rèn)消息并返回

上游應(yīng)用執(zhí)行本地業(yè)務(wù)

上游應(yīng)用通知可靠消息系統(tǒng)確認(rèn)業(yè)務(wù)已執(zhí)行并發(fā)送消息。

可靠消息系統(tǒng)修改消息狀態(tài)為發(fā)送狀態(tài)并將消息投遞到 MQ 中間件

第二階段：下游應(yīng)用監(jiān)聽 MQ 消息并執(zhí)行業(yè)務(wù)

下游應(yīng)用監(jiān)聽 MQ 消息并執(zhí)行業(yè)務(wù)，并且將消息的消費(fèi)結(jié)果通知可靠消息服務(wù)。

下游應(yīng)用監(jiān)聽 MQ 消息組件并獲取消息

下游應(yīng)用根據(jù) MQ 消息體信息處理本地業(yè)務(wù)

下游應(yīng)用向 MQ

確認(rèn)消息被消費(fèi)

下游應(yīng)用通知可靠消息系統(tǒng)消息被成功消費(fèi)，可靠消息將該消息狀態(tài)更改為已完成

RocketMQ貌似是一種先進(jìn)的實(shí)現(xiàn)方案了，但問題是缺乏文檔，無論是在Apache項(xiàng)目主頁，還是在阿里的頁面上，最多只告訴你如何用，而原理性或者指導(dǎo)性的東西非常缺乏。

當(dāng)然，如果你在阿里云上專門購買了RocketMQ服務(wù)，想必是另當(dāng)別論了。但如果你試圖在自己的服務(wù)環(huán)境中部署和使用，想必要?dú)v經(jīng)相當(dāng)大的學(xué)習(xí)曲線。畢竟是人家吃飯的家伙嘛

RabbitMQ遵循了AMQP規(guī)范，用消息確認(rèn)機(jī)制來保證：只要消息發(fā)送，就能確保被消費(fèi)者消費(fèi)來做到了消息最終一致性。而且開源，文檔還異常豐富，貌似是實(shí)現(xiàn)分布式事務(wù)的良好載體

rabbitmq的整個(gè)發(fā)送過程如下

1. 生產(chǎn)者發(fā)送消息到消息服務(wù)

2. 如果消息落地持久化完成，則返回一個(gè)標(biāo)志給生產(chǎn)者。生產(chǎn)者拿到這個(gè)確認(rèn)后，才能放心的說消息終于成功發(fā)到消息服務(wù)了。否則進(jìn)入異常處理流程。

    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        //try to resend msg
    } else {
        //delete msg in db
        }
    });

3. 消息服務(wù)將消息發(fā)送給消費(fèi)者

4. 消費(fèi)者接受并處理消息，如果處理成功則手動確認(rèn)。當(dāng)消息服務(wù)拿到這個(gè)確認(rèn)后，才放心的說終于消費(fèi)完成了。否則重發(fā)，或者進(jìn)入異常處理。

    final Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    String message = new String(body, "UTF-8");

    System.out.println(" [x] Received "" + message + """);
    try {
      doWork(message);
    } finally {
       //確認(rèn)收到消息
      channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };

我們來看看可能發(fā)送異常的四種

1. 直接無法到達(dá)消息服務(wù)

網(wǎng)絡(luò)斷了，拋出異常，業(yè)務(wù)直接回滾即可。如果出現(xiàn)connection closed錯(cuò)誤，直接增加 connection數(shù)即可

    connectionFactory.setChannelCacheSize(100);

2. 消息已經(jīng)到達(dá)服務(wù)器，但返回的時(shí)候出現(xiàn)異常

rabbitmq提供了確認(rèn)ack機(jī)制，可以用來確認(rèn)消息是否有返回。因此我們可以在發(fā)送前在db中(內(nèi)存或關(guān)系型數(shù)據(jù)庫)先存一下消息，如果ack異常則進(jìn)行重發(fā)

    /**confirmcallback用來確認(rèn)消息是否有送達(dá)消息隊(duì)列*/     
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        //try to resend msg
    } else {
        //delete msg in db
    }
    });
     /**若消息找不到對應(yīng)的Exchange會先觸發(fā)returncallback */
    rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
        try {
            Thread.sleep(Constants.ONE_SECOND);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    
        log.info("send message failed: " + replyCode + " " + replyText);
        rabbitTemplate.send(message);
    });

3. 消息送達(dá)后，消息服務(wù)自己掛了

如果設(shè)置了消息持久化，那么ack= true是在消息持久化完成后，就是存到硬盤上之后再發(fā)送的，確保消息已經(jīng)存在硬盤上，萬一消息服務(wù)掛了，消息服務(wù)恢復(fù)是能夠再重發(fā)消息

4. 未送達(dá)消費(fèi)者

消息服務(wù)收到消息后，消息會處于"UNACK"的狀態(tài)，直到客戶端確認(rèn)消息

    channel.basicQos(1); // accept only one unack-ed message at a time (see below)
    final Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");

    System.out.println(" [x] Received "" + message + """);
    try {
      doWork(message);
    } finally {
       //確認(rèn)收到消息
      channel.basicAck(envelope.getDeliveryTag(), false);
    }
      }
    };
    boolean autoAck = false;
    channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

5. 確認(rèn)消息丟失

消息返回時(shí)假設(shè)確認(rèn)消息丟失了，那么消息服務(wù)會重發(fā)消息。注意，如果你設(shè)置了autoAck= false，但又沒應(yīng)答channel.baskAck也沒有應(yīng)答channel.baskNack，那么會導(dǎo)致非常嚴(yán)重的錯(cuò)誤：消息隊(duì)列會被堵塞住，所以，無論如何都必須應(yīng)答

6. 消費(fèi)者業(yè)務(wù)處理異常

消息監(jiān)聽接受消息并處理，假設(shè)拋異常了，第一階段事物已經(jīng)完成，如果要配置回滾則過于麻煩，即使做事務(wù)補(bǔ)償也可能事務(wù)補(bǔ)償失效的情況，所以這里可以做一個(gè)重復(fù)執(zhí)行，比如guava的retry，設(shè)置一個(gè)指數(shù)時(shí)間來循環(huán)執(zhí)行，如果n次后依然失敗，發(fā)郵件、短信，用人肉來兜底。

《世界性難題...》一文中對分布式事務(wù)的幾種實(shí)現(xiàn)方式進(jìn)行了形象歸納

你每天上班，要經(jīng)過一條10公里的只有兩條車道的馬路到達(dá)公司。這條路很堵，經(jīng)常需要兩三個(gè)小時(shí)，上班時(shí)間沒有保證，這是2PC的問題-慢。

選擇一條很繞，長30公里但很少堵車的路，這是選b。上班時(shí)間有保證，但是必須早起，付出足夠的時(shí)間和汽油。這是柔性事務(wù)的問題，必須用具體業(yè)務(wù)來回滾，很難模塊化

選擇一條有點(diǎn)繞，長20公里的山路，路不平，只有suv可以走，這是事務(wù)消息最終一致性問題。引入了新的消息中間件，需要額外的開發(fā)成本。但我司開發(fā)的CoolMQ已經(jīng)對組件進(jìn)行了封裝，只需要發(fā)送，接受，就能滿足事務(wù)的要求。目前還有該方案的專題講座，大家可以根據(jù)自己的需要選用。

最后是GTS，GTS修了一條擁有4條車道的高架橋，沒有繞路，還是10公里。不堵車，對事務(wù)來說是高性能；不繞路，對事務(wù)來說是簡單易用，對業(yè)務(wù)無侵入，不用為事務(wù)而重構(gòu)；沒有車型限制，對事務(wù)來說是沒有功能限制，提供強(qiáng)一致事務(wù)。在沒有高架橋的時(shí)代，高架橋出現(xiàn)對交通來說就是一個(gè)顛覆性創(chuàng)新，很多以前看來無解的問題就迎刃而解了，同樣的，GTS希望通過創(chuàng)新改變數(shù)據(jù)一致性處理的行業(yè)現(xiàn)狀。但遺憾的是并未開源，而且需要結(jié)合阿里云服務(wù)來使用。