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

講座

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

開源項目

我們利用消息隊列實現了分布式事務的最終一致性解決方案，請大家圍觀。可以參考Github CoolMQ源碼，項目支持網站: http://rabbitmq.org.cn，最新文章或實現會更新在上面

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

雖然如此，但《世界難題...》一文中對事務還是歸納的還是蠻到位的:“一個看似簡單的功能，內部可能需要調用多個“服務”并操作多個數據庫或分片來實現，單一技術手段和解決方案已無法滿足這些復雜應用場景。因此，分布式系統架構中分布式事務是一個繞不過去的挑戰。

什么是分布式事務？簡單的說，就是一次大操作由不同小操作組成，這些小操作分布在不同服務器上，分布式事務需要保證這些小操作要么全部成功，要么全部失敗。”

舉個栗子：

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

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

以開會為例

甲乙丙丁四人要組織一個會議，需要確定會議時間，不妨設甲是協調者，乙丙丁是參與者。

投票階段

甲發郵件給乙丙丁，周二十點開會是否有時間；

甲回復有時間；

乙回復有時間；

丙遲遲不回復，此時對于這個活動，甲乙丙均處于阻塞狀態，算法無法繼續進行；

丙回復有時間（或者沒有時間）；

提交階段

協調者甲將收集到的結果反饋給乙丙丁（什么時候反饋，以及反饋結果如何，在此例中取決與丙的時間與決定）；

乙收到；

丙收到；

丁收到；

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

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

此外還有三階段提交

大家有興趣的不妨研究下

所謂柔性事務是相對強制鎖表的剛性事務而言。流程入下：服務器A的事務如果執行順利，那么事務A就先行提交，如果事務B也執行順利，則事務B也提交，整個事務就算完成。但是如果事務B執行失敗，事務B本身回滾，這時事務A已經被提交，所以需要執行一個補償操作，將已經提交的事務A執行的操作作反操作，恢復到未執行前事務A的狀態。

缺點是業務侵入性太強，還要補償操作，缺乏普遍性，沒法大規模推廣。

目前基于消息隊列的解決方案有阿里的RocketMQ，它實現了半消息的解決方案,有點類似于Paxos算法，具體流程如下

第一階段：上游應用執行業務并發送MQ消息

上游應用發送待確認消息到可靠消息系統

可靠消息系統保存待確認消息并返回

上游應用執行本地業務

上游應用通知可靠消息系統確認業務已執行并發送消息。

可靠消息系統修改消息狀態為發送狀態并將消息投遞到 MQ 中間件

第二階段：下游應用監聽 MQ 消息并執行業務

下游應用監聽 MQ 消息并執行業務，并且將消息的消費結果通知可靠消息服務。

下游應用監聽 MQ 消息組件并獲取消息

下游應用根據 MQ 消息體信息處理本地業務

下游應用向 MQ

確認消息被消費

下游應用通知可靠消息系統消息被成功消費，可靠消息將該消息狀態更改為已完成

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

當然，如果你在阿里云上專門購買了RocketMQ服務，想必是另當別論了。但如果你試圖在自己的服務環境中部署和使用，想必要歷經相當大的學習曲線。畢竟是人家吃飯的家伙嘛

RabbitMQ遵循了AMQP規范，用消息確認機制來保證：只要消息發送，就能確保被消費者消費來做到了消息最終一致性。而且開源，文檔還異常豐富，貌似是實現分布式事務的良好載體

rabbitmq的整個發送過程如下

1. 生產者發送消息到消息服務

2. 如果消息落地持久化完成，則返回一個標志給生產者。生產者拿到這個確認后，才能放心的說消息終于成功發到消息服務了。否則進入異常處理流程。

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

3. 消息服務將消息發送給消費者

4. 消費者接受并處理消息，如果處理成功則手動確認。當消息服務拿到這個確認后，才放心的說終于消費完成了。否則重發，或者進入異常處理。

    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 {
       //確認收到消息
      channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };

我們來看看可能發送異常的四種

1. 直接無法到達消息服務

網絡斷了，拋出異常，業務直接回滾即可。如果出現connection closed錯誤，直接增加 connection數即可

    connectionFactory.setChannelCacheSize(100);

2. 消息已經到達服務器，但返回的時候出現異常

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

    /**confirmcallback用來確認消息是否有送達消息隊列*/     
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        //try to resend msg
    } else {
        //delete msg in db
    }
    });
     /**若消息找不到對應的Exchange會先觸發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. 消息送達后，消息服務自己掛了

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

4. 未送達消費者

消息服務收到消息后，消息會處于"UNACK"的狀態，直到客戶端確認消息

    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 {
       //確認收到消息
      channel.basicAck(envelope.getDeliveryTag(), false);
    }
      }
    };
    boolean autoAck = false;
    channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

5. 確認消息丟失

消息返回時假設確認消息丟失了，那么消息服務會重發消息。注意，如果你設置了autoAck= false，但又沒應答channel.baskAck也沒有應答channel.baskNack，那么會導致非常嚴重的錯誤：消息隊列會被堵塞住，所以，無論如何都必須應答

6. 消費者業務處理異常

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

《世界性難題...》一文中對分布式事務的幾種實現方式進行了形象歸納

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

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

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

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