摘要：在對事實性要求沒有那么高的情況下，可以用基于最大努力交付消息隊列以及消息存儲來解決最終一致性?？煽肯⒎?wù)和消息組件，協(xié)調(diào)上下游消息的傳遞，并確保上下游數(shù)據(jù)的一致性。下游應(yīng)用通知可靠消息服務(wù)該消息已經(jīng)成功消費。

本文對比 二階段事務(wù)、最大努力交付以及消息最終一致性，并給出部分解決方案,最終一致性方案參考阿里RockMQ事務(wù)消息：http://blog.csdn.net/chunlong...）

項目git地址：https://github.com/vvsuperman...

分布式系統(tǒng)最終一致性有N種方案，比如2PC（2階段事務(wù)） ，以及三段提交等等，但開銷較大，實現(xiàn)起來復(fù)雜，比如2階段事務(wù)為例，需要引入一個協(xié)調(diào)者（Coordinator）來統(tǒng)一掌控所有參與者（Participant）的操作結(jié)果

以開會為例：
甲乙丙丁四人要組織一個會議，需要確定會議時間，不妨設(shè)甲是協(xié)調(diào)者，乙丙丁是參與者。
投票階段：
（1）甲發(fā)郵件給乙丙丁，周二十點開會是否有時間；
（2）甲回復(fù)有時間；
（3）乙回復(fù)有時間；
（4）丙遲遲不回復(fù)，此時對于這個活動，甲乙丙均處于阻塞狀態(tài)，算法無法繼續(xù)進行；
（5）丙回復(fù)有時間（或者沒有時間）；
提交階段：
（1）協(xié)調(diào)者甲將收集到的結(jié)果反饋給乙丙?。ㄊ裁磿r候反饋，以及反饋結(jié)果如何，在此例中取決與丙的時間與決定）；
（2）乙收到；
（3）丙收到；
（4）丁收到；
不僅要鎖住參與者的所有資源，而且要鎖住協(xié)調(diào)者資源，開銷大。一句話總結(jié)就是：2PC效率很低，分布式事務(wù)很難做。

在對事實性要求沒有那么高的情況下，可以用基于最大努力交付 && 消息隊列以及消息存儲來解決最終一致性。

所謂最大努力交付，就是俺反正用最大努力做，能不能成功，不做完全保證
會涉及到三個模塊

上游應(yīng)用，發(fā)消息到 MQ 隊列。

下游應(yīng)用（例如短信服務(wù)、郵件服務(wù)），接受請求，并返回通知結(jié)果。

最大努力通知服務(wù)，監(jiān)聽消息隊列，將消息存儲到數(shù)據(jù)庫中，并按照通知規(guī)則調(diào)用下游應(yīng)用的發(fā)送通知接口。

具體流程如下

上游應(yīng)用發(fā)送 MQ 消息到 MQ 組件內(nèi)，消息內(nèi)包含通知規(guī)則和通知地址

最大努力通知服務(wù)監(jiān)聽到 MQ 內(nèi)的消息，解析通知規(guī)則并放入延時隊列等待觸發(fā)通知

最大努力通知服務(wù)調(diào)用下游的通知地址，如果調(diào)用成功，則該消息標記為通知成功，如果失敗則在滿足通知規(guī)則（例如 5 分鐘發(fā)一次，共發(fā)送 10 次）的情況下重新放入延時隊列等待下次觸發(fā)。

最大努力通知服務(wù)表示在不影響主業(yè)務(wù)的情況下，盡可能地確保數(shù)據(jù)的一致性。它需要開發(fā)人員根據(jù)業(yè)務(wù)來指定通知規(guī)則，在滿足通知規(guī)則的前提下，盡可能的確保數(shù)據(jù)的一致，以達到最大努力的目的。

實現(xiàn)上也比較簡單，目前主流消息隊列都有ack機制，當沒收到ack的時候用規(guī)則做定時重發(fā)即可。
優(yōu)點：實現(xiàn)簡單
缺點：無補償機制，不保證能夠送達
實現(xiàn)要點： 保證消息發(fā)送失敗之后能夠和業(yè)務(wù)一起回滾；消息接受方保證冥等性；定時重發(fā)機制，采用一定的重發(fā)策略，例如說指數(shù)增長，據(jù)說阿里采用redis的zset來完成，參考https://zhuanlan.zhihu.com/p/...
消息進到zset后，DelayQ會通過timer觸發(fā)(比如秒級)，fork相應(yīng)的消費線程去處理zset里ExecuteTime大于當前時間的消息。DelayQ拿到一條消息后，解析其中的callbackurl，并組裝參數(shù)，push業(yè)務(wù)消息給Consumer.
Consumer返回處理成功，那么zrem Codis里的消息。如果處理失敗，則計算其下次嘗試時間，并更新其ExecuteTime.

此方案涉及 3 個模塊：

上游應(yīng)用，執(zhí)行業(yè)務(wù)并發(fā)送 MQ 消息。

可靠消息服務(wù)和 MQ 消息組件，協(xié)調(diào)上下游消息的傳遞，并確保上下游數(shù)據(jù)的一致性。

下游應(yīng)用，監(jiān)聽 MQ 的消息并執(zhí)行自身業(yè)務(wù)。

上游應(yīng)用將本地業(yè)務(wù)執(zhí)行和消息發(fā)送綁定在同一個本地事務(wù)中，保證要么本地操作成功并發(fā)送 MQ 消息，要么兩步操作都失敗并回滾。

上游應(yīng)用和可靠消息之間的業(yè)務(wù)交互圖如下：

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

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

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

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

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

以上每一步都可能出現(xiàn)失敗情況，分析一下這 5 步出現(xiàn)異常后上游業(yè)務(wù)和消息發(fā)送是否一致：

上游應(yīng)用執(zhí)行完成，下游應(yīng)用尚未執(zhí)行或執(zhí)行失敗時，此事務(wù)即處于 BASE 理論的 Soft State 狀態(tài)。

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

可靠消息的狀態(tài)需要和下游應(yīng)用的業(yè)務(wù)執(zhí)行保持一致，可靠消息狀態(tài)不是已完成時，確保下游應(yīng)用未執(zhí)行，可靠消息狀態(tài)是已完成時，確保下游應(yīng)用已執(zhí)行。

下游應(yīng)用和可靠消息服務(wù)之間的交互圖如下：

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

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

下游應(yīng)用向 MQ 組件自動發(fā)送 ACK 確認消息被消費

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

以上每一步都可能出現(xiàn)失敗情況，分析一下這 4 步出現(xiàn)異常后下游業(yè)務(wù)和消息狀態(tài)是否一致：

通過分析以上兩個階段可能失敗的情況，為了確保上下游數(shù)據(jù)的最終一致性，在可靠消息系統(tǒng)中，需要開發(fā) 消息狀態(tài)確認 和 消息重發(fā) 兩個功能以實現(xiàn) BASE 理論的 Eventually Consistent 特性。

可靠消息服務(wù)定時監(jiān)聽消息的狀態(tài)，如果存在狀態(tài)為待確認并且超時的消息，則表示上游應(yīng)用和可靠消息交互中的步驟 4 或者 5 出現(xiàn)異常。

可靠消息則攜帶消息體內(nèi)的信息向上游應(yīng)用發(fā)起請求查詢該業(yè)務(wù)是否已執(zhí)行。上游應(yīng)用提供一個可查詢接口供可靠消息追溯業(yè)務(wù)執(zhí)行狀態(tài)，如果業(yè)務(wù)執(zhí)行成功則更改消息狀態(tài)為已發(fā)送，否則刪除此消息確保數(shù)據(jù)一致。具體流程如下：

可靠消息查詢超時的待確認狀態(tài)的消息

向上游應(yīng)用查詢業(yè)務(wù)執(zhí)行的情況

業(yè)務(wù)未執(zhí)行，則刪除該消息，保證業(yè)務(wù)和可靠消息服務(wù)的一致性。業(yè)務(wù)已執(zhí)行，則修改消息狀態(tài)為已發(fā)送，并發(fā)送消息到 MQ 組件。

消息已發(fā)送則表示上游應(yīng)用已經(jīng)執(zhí)行，接下來則確保下游應(yīng)用也能正常執(zhí)行。

可靠消息服務(wù)發(fā)現(xiàn)可靠消息服務(wù)中存在消息狀態(tài)為已發(fā)送并且超時的消息，則表示可靠消息服務(wù)和下游應(yīng)用中存在異常的步驟，無論哪個步驟出現(xiàn)異常，可靠消息服務(wù)都將此消息重新投遞到 MQ 組件中供下游應(yīng)用監(jiān)聽。

下游應(yīng)用監(jiān)聽到此消息后，在保證冪等性的情況下重新執(zhí)行業(yè)務(wù)并通知可靠消息服務(wù)此消息已經(jīng)成功消費，最終確保上游應(yīng)用、下游應(yīng)用的數(shù)據(jù)最終一致性。具體流程如下：

可靠消息服務(wù)定時查詢狀態(tài)為已發(fā)送并超時的消息

可靠消息將消息重新投遞到 MQ 組件中

下游應(yīng)用監(jiān)聽消息，在滿足冪等性的條件下，重新執(zhí)行業(yè)務(wù)。

下游應(yīng)用通知可靠消息服務(wù)該消息已經(jīng)成功消費。

通過消息狀態(tài)確認和消息重發(fā)兩個功能，可以確保上游應(yīng)用、可靠消息服務(wù)和下游應(yīng)用數(shù)據(jù)的最終一致性。

我們在rabbitmq上肉身實戰(zhàn)了一下可靠消息，rabbitmq的發(fā)送過程如下

發(fā)送消息到消息服務(wù)

消息隊列將消息發(fā)送給監(jiān)聽

消息監(jiān)聽接受并處理消息

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

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

connectionFactory.setChannelCacheSize(100);

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

    /**confirmcallback用來確認消息是否有送達消息隊列*/     
   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);
    });
    

如果消息沒有到exchange,則confirm回調(diào),ack=false
如果消息到達exchange,則confirm回調(diào),ack=true
但如果是找不到exchange，則會先觸發(fā)returncallback

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

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

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

消息返回時假設(shè)確認消息丟失了，那么消息服務(wù)會重發(fā)消息。注意，如果你設(shè)置了autoAck= false，但又沒應(yīng)答 channel.baskAck也沒有應(yīng)答 channel.baskNack，那么會導(dǎo)致非常嚴重的錯誤：消息隊列會被堵塞住，可參考http://blog.sina.com.cn/s/blo...，所以，無論如何都必須應(yīng)答

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