如何基于RocketMQ的事務消息特性實現分布式系統的最終一致性？

前言

在這篇文章中我們將介紹RocketMQ的事務消息相關的內容，并通過一些實踐和大家一起來探索下事務消息如何解決分布式系統中的分布式事務問題。

事務消息原理

事務消息特性可以看作是兩階段協議的消息實現方式，用以確保在以消息中間件解耦的分布式系統中本地事務的執行和消息的發送，可以以原子的方式進行。

舉個例子，以某互聯網公司的用戶余額充值為例，因為有充返活動（充值100元贈送20元），優惠比較大，用戶Joe禁不住誘惑用支付寶向自己的余額賬戶充值了100元，支付成功后Joe的余額賬戶有了120元錢。

而該公司的關于用戶余額充值的系統設計是這樣的：

在這個設計流程中，該公司通過自建支付系統完成用戶Joe的支付寶扣款操作，成功后需要更新支付流水的狀態，因為用戶的余額賬戶系統與支付系統之間通過MQ解耦了，所以支付系統在完成支付流水狀態更新后需要通過發送MQ消息到消息中間件服務，然后用戶余額系統作為消費者通過消息消費的方式完成用戶余額的增加操作。

這里有個問題：“支付系統如何確保這筆余額充值消息一定會成功發送到MQ，并且用戶余額系統一定能處理成功呢”？如果支付系統在完成支付訂單狀態更新后，MQ消息發送失敗或者用戶余額系統消息處理失敗的話，都會導致Joe支付扣款成功，而自己的余額賬戶卻沒到賬的情況發生。

為了解決這個問題，按照目前的系統設計是需要“支付系統-MQ服務-用戶余額系統”三者的處理滿足數據的一致性要求。例如，如果支付系統感知到消息發送失敗后還可以進行重新投遞，從而確保支付系統與用戶余額數據的最終一致性。

而上述問題就是事務消息要解決的問題，在具體了解RocketMQ提供的事務消息機制之前，我們先來看下在RocketMQ的早期版本不支持事務消息，或者因為歷史原因選擇的消息中間件本身就不支持事務消息的情況下，一些大公司是怎么解決這個問題的？

早期為了實現基于MQ異步調用的多個服務間，業務邏輯執行要么一起成功、要么一起失敗，具備事務特點，通常會采用可靠消息最終一致性方案，來實現分布式事務。還是以Joe充值這件事來舉例，可靠消息方案實現過程如下：

在可靠消息最終一致性方案中，為了實現分布式事務，需要確保上游服務本地事務的處理與MQ消息的投遞具有原子性，也就是說上游服務本地事務處理成功后要確保消息一定要成功投遞到MQ服務，否則消息就不應該被投遞到MQ服務；同樣，被成功投遞到MQ服務的消息，也一定要被下游服務成功處理，否則就需要重新投遞MQ消息。

為了實現雙向的原子性，可靠消息服務需要對消息進行狀態標記，與此同時還需要對消息進行狀態檢查，從而實現重新投遞及消息狀態的最終一致性。核心流程說明如下：

1、上游服務（支付系統）如何確保完成自身支付成功狀態更新后消息100%的能夠投遞到下游服務（用戶余額系統）指定的Topic中？

在這個流程中上游服務在進行本地數據庫事務操作前，會先發送一個狀態為“待確認”的消息至可靠消息服務，而不是直接將消息投遞到MQ服務的指定Topic。可靠消息服務此時會將該消息記錄到自身服務的消息數據庫中（消息狀態為->待確認），完成后可靠消息服務會回調上游服務表示收到了消息，你們可以進行本地事務的操作了。

之后上游服務就會開啟本地數據庫事務執行業務邏輯操作，這里支付系統就會將該筆支付訂單狀態更新為“已成功”。（注意，這里只是舉個示例場景，在真正的實踐中一般是不會把支付訂單本身的狀態與業務端回調放在一個事務流程中的，關于這部分的詳細說明我們在下面的場景說明中再討論）。

如果上游服務本地數據庫事務執行成功，則繼續向可靠消息服務發送消息確認消息，此時可靠消息服務就會正式將消息投遞到MQ服務，并且同時更新消息數據庫中的消息狀態為“已發送”。（注意，這里可靠消息服務更新消息狀態與投遞消息至MQ也必須是在一個原子操作中，即消息投遞成功則一定要將消息狀態更新為“已發送”，所以在編程的細節中，可靠消息服務一般會先更新消息狀態，然后再進行消息投遞，這樣即使消息投遞失敗，也可以對消息狀態進行回滾->“待確認”，相反如果先進行消息投遞再更新消息狀態，可能就不好控制了）。

相反，如果上游本地數據庫事務執行失敗，則需要向可靠消息服務發送消息刪除消息，可靠消息服務此時就會將消息刪除，這樣就意味著事務在上游消息投遞過程中就被回滾了，而流程也就此結束了，此時上游服務可以需要通過業務邏輯的設計進行重發，這個就不再分布式事務的討論范疇了。

說到這里，大家可能會有疑問了！因為在上述描述中，即使上游服務本地數據庫事務執行成功了，但是在發送確認消息至可靠消息服務的過程中，以及可靠消息服務在投遞消息至MQ服務的過程中，還是會存在失敗的風險，這樣的話還是會導致支付服務更新了狀態，但是用戶余額系統連消息都沒有收到的情況發生？

實際上，實現數據一致性是一個復雜的活。在這個方案中可靠消息服務作為基礎性的服務除了執行正常的邏輯外，還得處理復雜的異常場景。在實現過程中可靠消息服務需要啟動相應的后臺線程，不斷輪訓消息的狀態，這里會輪訓消息狀態為“待確認”的消息，并判斷該消息的狀態的持續時間是否超過了規定的時間，如果超過規定時間的消息還處于“待確認”的狀態，就會觸發上游服務狀態詢問機制。

可靠消息服務就會調用上游服務提供的相關借口，詢問這筆消息的處理情況，如果這筆消息在上游服務處理成功，則后臺線程就會繼續觸發上圖中的步驟5，更新消息狀態為“已發送”并投遞消息至MQ服務；反之如果這筆消息上游服務處理失敗，可靠消息服務則會進行消息刪除。通過這樣以上機制就確保了“上游服務本地事務成功處理+消息成功投遞”處于一個原子操作了。

2、下游服務（用戶余額系統）如何確保對MQ服務Topic消息的消費100%都能處理成功？

在1的過程中，確保了上游服務邏輯處理與MQ消息的投遞具備原子性，那么當消息被成功投遞到了MQ服務的指定Topic后，下游服務如何才能確保消息的消費一定能被成功處理呢？

在正常的流程中，下游服務等待消費Topic的消息并進行自身本地數據庫事務的處理，如果處理成功則會主動通知可靠消息服務，可靠消息服務此時就會將消息的狀態更新為“已完成”；反之，處理失敗下游服務就無法再主動向可靠消息服務發送通知消息了。

此時，與消息投遞過程中的異常邏輯一樣，可靠消息服務也會啟動相應的后臺線程，輪詢一直處于“已發送”狀態的消息，判斷狀態持續時間是否超過了規定時間，如果超時，可靠消息服務就會再次向MQ服務投遞此消息，從而確保消息能被再次消費處理。（注意，也可能出現下游服務處理成功，但是通知消息發送失敗的情況，所以為了確保冪等，下游服務也需要在業務邏輯上做好相應的防重處理）。

RocketMQ事務消息機制

在