Spring Cloud分佈式事務終極解決方案探討

一 小小推廣

講座

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

開源項目

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

二 前言

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

雖然如此，但《世界難題...》一文中對事務仍是概括的仍是蠻到位的:「一個看似簡單的功能，內部可能須要調用多個「服務」並操做多個數據庫或分片來實現，單一技術手段和解決方案已沒法知足這些複雜應用場景。所以，分佈式系統架構中分佈式事務是一個繞不過去的挑戰。

什麼是分佈式事務？簡單的說，就是一次大操做由不一樣小操做組成，這些小操做分佈在不一樣服務器上，分佈式事務須要保證這些小操做要麼所有成功，要麼所有失敗。」

舉個栗子：

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

三 2階段提交(2PC, 3PC等)

2階段提交是分佈式事務傳統解決方案，先進爲止還普遍存在。當一個事務跨越多個節點時，爲了保持事務ACID特性，須要引入一個做爲協調者來統一掌控全部節點(稱做參與者)的操做結果並最終指示這些節點是否要把操做結果進行真正的提交(好比將更新後的數據寫入磁盤等等)。所以，二階段提交的算法思路能夠歸納爲：參與者將操做成敗通知協調者，再由協調者根據全部參與者的反饋情報決定各參與者是否要提交操做仍是停止操做。

以開會爲例

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

投票階段

1. 甲發郵件給乙丙丁，週二十點開會是否有時間；
2. 甲回覆有時間；
3. 乙回覆有時間；
4. 丙遲遲不回覆，此時對於這個活動，甲乙丙均處於阻塞狀態，算法沒法繼續進行；
5. 丙回覆有時間（或者沒有時間）；

提交階段

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

不只要鎖住參與者的全部資源，並且要鎖住協調者資源，開銷大。一句話總結就是：2PC效率很低，對高併發很不友好。

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

此外還有三階段提交

你們有興趣的不妨研究下

四 柔性事務

所謂柔性事務是相對強制鎖表的剛性事務而言。流程入下：服務器A的事務若是執行順利，那麼事務A就先行提交，若是事務B也執行順利，則事務B也提交，整個事務就算完成。可是若是事務B執行失敗，事務B自己回滾，這時事務A已經被提交，因此須要執行一個補償操做，將已經提交的事務A執行的操做做反操做，恢復到未執行前事務A的狀態。

缺點是業務侵入性太強，還要補償操做，缺少廣泛性，無法大規模推廣。

五 消息最終一致性解決方案之RocketMQ

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

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

1. 上游應用發送待確認消息到可靠消息系統
2. 可靠消息系統保存待確認消息並返回
3. 上游應用執行本地業務
4. 上游應用通知可靠消息系統確認業務已執行併發送消息。

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

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

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

1. 下游應用監聽 MQ 消息組件並獲取消息
2. 下游應用根據 MQ 消息體信息處理本地業務
3. 下游應用向 MQ
4. 確認消息被消費
5. 下游應用通知可靠消息系統消息被成功消費，可靠消息將該消息狀態更改成已完成

RocketMQ貌似是一種先進的實現方案了，但問題是缺少文檔，不管是在Apache項目主頁，仍是在阿里的頁面上，最多隻告訴你如何用，而原理性或者指導性的東西很是缺少。

固然，若是你在阿里雲上專門購買了RocketMQ服務，想必是另當別論了。但若是你試圖在本身的服務環境中部署和使用，想必要歷經至關大的學習曲線。畢竟是人家吃飯的傢伙嘛

六 消息最終一致性解決方案之RabbitMQ實現

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

6.1 RabbitMQ消息確認機制

---

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);
        }
      }
    };

6.2 異常

---

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

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但願經過創新改變數據一致性處理的行業現狀。但遺憾的是並未開源，並且須要結合阿里雲服務來使用。