一、背景

在微服務架構中，我們常常使用非同步化的手段來提升系統的 吞吐量 和 解耦 上下游，而構建非同步架構最常用的手段就是使用 訊息佇列(MQ)，那非同步架構怎樣才能實現資料一致性呢？本文主要介紹如何使用RocketMQ的事務訊息來解決一致性問題。

RocketMQ 是阿里巴巴開源的分散式訊息中介軟體，目前已成為 Apache 的頂級專案。歷經多次天貓雙十一海量訊息考驗，具有高效能、低延時和高可靠等特性

PS：同步場景怎樣保證一致性？請看文章《Spring Cloud同步場景分散式事務怎樣做？試試Seata》

 

二、MQ選型

可以看到在 業務處理 方面來說 RocketMQ

優於其他對手，而且原生支援 事務訊息

PS：業務系統用的是其他 MQ 產品但是又需要 事務訊息 怎麼辦？學習原理自己開發實現！

 

三、什麼是事務訊息

例如下圖的場景：生成訂單記錄 -> MQ -> 增加積分

我們是應該先 建立訂單記錄，還是先 傳送MQ訊息 呢？

1. 先發送MQ訊息：這個明顯是不行的，因為如果訊息傳送成功，而訂單建立失敗的話是沒辦法把訊息收回來的

2. 先建立訂單記錄：如果訂單建立成功後MQ訊息傳送失敗 丟擲異常，因為兩個操作都在本地事務中所以訂單資料是可以 回滾 的

上面的 方式二 看似沒問題，但是 網路是不可靠的！如果 MQ 的響應因為網路原因沒有收到，所以在面對不確定的結果只好進行回滾；但是 MQ

端又確實是收到了這條訊息的，只是回給客戶端的 響應丟失 了！
 
所以 事務訊息 就是用來保證 本地事務 與 MQ訊息傳送 的原子性！

 

四、RocketMQ事務訊息原理

主要的邏輯分為兩個流程：

• 事務訊息傳送及提交：
  1. 傳送 half訊息
  2. MQ服務端 響應訊息寫入結果
  3. 根據傳送結果執行 本地事務（如果寫入失敗，此時half訊息對業務 不可見，本地邏輯不執行）
  4. 根據本地事務狀態執行 Commit 或者 Rollback（Commit操作生成訊息索引，訊息對消費者 可見）
      
• 回查流程：
  1. 對於長時間沒有 Commit/Rollback 的事務訊息（pending
     狀態的訊息），從服務端發起一次 回查
  2. Producer 收到回查訊息，檢查回查訊息對應的 本地事務狀態
  3. 根據本地事務狀態，重新 Commit 或者 Rollback

 
邏輯時序圖

 

五、非同步架構一致性實現思路

從上面的原理可以發現 事務訊息 僅僅只是保證本地事務和MQ訊息傳送形成整體的 原子性，而投遞到MQ伺服器後，並無法保證消費者一定能消費成功！
 
如果 消費端消費失敗 後的處理方式，建議是記錄異常資訊然後 人工處理，並不建議回滾上游服務的資料(因為兩者是 解耦 的，而且 複雜度 太高)
 
我們可以利用 MQ 的兩個特性 重試 和 死信佇列 來協助消費端處理：

1. 消費失敗後進行一定次數的 重試
2. 重試後也失敗的話該訊息丟進 死信佇列 裡
3. 另外起一個執行緒監聽消費 死信佇列 裡的訊息，記錄日誌並且預警！

因為有 重試 所以消費者需要實現 冪等性

 

六、分散式事務場景樣例

下面就用剛剛提到的場景：生成訂單記錄 -> MQ -> 增加積分；來簡單講一下 Spring Cloud 中應該怎麼做，詳細程式碼請 下載demo 檢視。
PS：怎樣安裝部署RocketMQ可以參考《Apache RocketMQ 訊息佇列部署與視覺化介面安裝》

6.1. 引入依賴

使用 spring-cloud-stream 框架來訪問 RocketMQ

Spring Cloud Stream 是一個構建訊息驅動的框架，通過抽象的定義實現應用與MQ訊息佇列之間的解耦，目前支援 RabbitMQ、kafka 和 RocketMQ

 

6.2. 開啟事務訊息

訊息生產者需要新增 transactional: true 開啟 事務訊息

 

6.3. 訂單服務傳送half訊息

因為開啟了 事務訊息 所以這裡傳送的是 half訊息 對於消費端是 不可見 的

 

6.4. 訂單服務監聽half訊息

使用 @RocketMQTransactionListener 註解監聽 半訊息，並實現 RocketMQLocalTransactionListener 介面，該介面有兩個方法

• executeLocalTransaction：用於提交本地事務
• checkLocalTransaction：用於事務回查

如果提交事務訊息失敗，需等待約1分鐘左右 事務回查 方法才會被呼叫

 

6.5. 積分服務消費訊息

注意：因為有 重試，這裡如果是真實的業務需要自行實現 冪等性

 

6.6. 消費死信佇列預警

監聽並消費死信佇列中的訊息，用於記錄錯誤日誌，並且預警通知運維人員等

 

6.7. 測試用例

demo中提供了3個介面分別測試不同的場景：

• 事務成功
  http://localhost:11002/success
  流程如下：
  1. 訂單建立 成功
  2. 提交事務訊息 成功
  3. 消費訊息增加積分 成功
• 訂單建立成功但提交事務訊息失敗
  http://localhost:11002/produceError
  流程如下：
  1. 訂單建立 成功
  2. 提交事務訊息 失敗
  3. 事務回查(等待1分鐘左右) 成功
  4. 提交事務訊息 成功
  5. 消費訊息增加積分 成功
• 消費訊息失敗
  http://localhost:11002/consumeError
  流程如下：
  1. 訂單建立 成功
  2. 提交事務訊息 成功
  3. 消費訊息增加積分 失敗
  4. 重試消費訊息 失敗
  5. 進入死信佇列 成功
  6. 消費死信佇列的訊息 成功
  7. 記錄日誌併發出預警 成功

 

七、demo下載地址

https://gitee.com/zlt2000/microservices-platform/tree/master/zlt-demo/rocketmq-demo/rocketmq-transactional

 

推薦閱讀

• 日誌排查問題困難？分散式日誌鏈路跟蹤來幫你
• zuul整合Sentinel最新的閘道器流控元件
• 阿里註冊中心Nacos生產部署方案
• Spring Boot自定義配置項在IDE裡面實現自動提示
• Spring Cloud Zuul的動態路由怎樣做？整合Nacos實現很簡單
• Spring Cloud開發人員如何解決服務衝突和例項亂竄？
• Spring Cloud同步場景分散式事務怎樣做？試試Seata

 
掃碼關注有驚喜！

相關推薦