一翎朱、背景
在微服務(wù)架構(gòu)中，我們常常使用異步化的手段來提升系統(tǒng)的 吞吐量 和 解耦 上下游尺铣，而構(gòu)建異步架構(gòu)最常用的手段就是使用 消息隊列(MQ)拴曲，那異步架構(gòu)怎樣才能實現(xiàn)數(shù)據(jù)一致性呢？本文主要介紹如何使用RocketMQ的事務(wù)消息來解決一致性問題凛忿。
RocketMQ 是阿里巴巴開源的分布式消息中間件澈灼，目前已成為 Apache 的頂級項目。歷經(jīng)多次天貓雙十一海量消息考驗店溢，具有高性能叁熔、低延時和高可靠等特性
PS：同步場景怎樣保證一致性？請看文章《Spring Cloud同步場景分布式事務(wù)怎樣做床牧？試試Seata》
?
二荣回、MQ選型
可以看到在 業(yè)務(wù)處理 方面來說 RocketMQ 優(yōu)于其他對手，而且原生支持 事務(wù)消息
PS：業(yè)務(wù)系統(tǒng)用的是其他 MQ 產(chǎn)品但是又需要 事務(wù)消息 怎么辦戈咳？學習原理自己開發(fā)實現(xiàn)心软！
?
三、什么是事務(wù)消息
例如下圖的場景：生成訂單記錄 -> MQ -> 增加積分
我們是應(yīng)該先 創(chuàng)建訂單記錄著蛙，還是先 發(fā)送MQ消息 呢删铃？
先發(fā)送MQ消息：這個明顯是不行的，因為如果消息發(fā)送成功册踩，而訂單創(chuàng)建失敗的話是沒辦法把消息收回來的
先創(chuàng)建訂單記錄：如果訂單創(chuàng)建成功后MQ消息發(fā)送失敗 拋出異常泳姐，因為兩個操作都在本地事務(wù)中所以訂單數(shù)據(jù)是可以 回滾 的
上面的 方式二 看似沒問題，但是 網(wǎng)絡(luò)是不可靠的暂吉！如果 MQ 的響應(yīng)因為網(wǎng)絡(luò)原因沒有收到胖秒，所以在面對不確定的結(jié)果只好進行回滾缎患；但是 MQ 端又確實是收到了這條消息的，只是回給客戶端的 響應(yīng)丟失 了阎肝！
?
所以 事務(wù)消息 就是用來保證 本地事務(wù) 與 MQ消息發(fā)送 的原子性挤渔！
?
四、RocketMQ事務(wù)消息原理
主要的邏輯分為兩個流程：
-
事務(wù)消息發(fā)送及提交：
- 發(fā)送
half消息 -
MQ服務(wù)端響應(yīng)消息寫入結(jié)果 - 根據(jù)發(fā)送結(jié)果執(zhí)行
本地事務(wù)（如果寫入失敗风题，此時half消息對業(yè)務(wù) 不可見判导，本地邏輯不執(zhí)行） - 根據(jù)本地事務(wù)狀態(tài)執(zhí)行
Commit或者Rollback（Commit操作生成消息索引，消息對消費者 可見）
?
- 發(fā)送
-
回查流程：
- 對于長時間沒有
Commit/Rollback的事務(wù)消息（pending狀態(tài)的消息）沛硅，從服務(wù)端發(fā)起一次 回查 -
Producer收到回查消息眼刃，檢查回查消息對應(yīng)的本地事務(wù)狀態(tài) - 根據(jù)本地事務(wù)狀態(tài)，重新
Commit或者Rollback
- 對于長時間沒有
?
邏輯時序圖
?
五摇肌、異步架構(gòu)一致性實現(xiàn)思路
從上面的原理可以發(fā)現(xiàn) 事務(wù)消息 僅僅只是保證本地事務(wù)和MQ消息發(fā)送形成整體的 原子性擂红，而投遞到MQ服務(wù)器后，并無法保證消費者一定能消費成功围小！
?
如果 消費端消費失敗 后的處理方式昵骤，建議是記錄異常信息然后 人工處理，并不建議回滾上游服務(wù)的數(shù)據(jù)(因為兩者是 解耦 的肯适，而且 復(fù)雜度 太高)
?
我們可以利用 MQ 的兩個特性 重試 和 死信隊列 來協(xié)助消費端處理：
- 消費失敗后進行一定次數(shù)的
重試 - 重試后也失敗的話該消息丟進
死信隊列里 - 另外起一個線程監(jiān)聽消費
死信隊列里的消息变秦，記錄日志并且預(yù)警！
因為有 重試 所以消費者需要實現(xiàn) 冪等性
?
六框舔、分布式事務(wù)場景樣例
下面就用剛剛提到的場景：生成訂單記錄 -> MQ -> 增加積分蹦玫；來簡單講一下 Spring Cloud 中應(yīng)該怎么做，詳細代碼請 下載demo 查看雨饺。
PS：怎樣安裝部署RocketMQ可以參考《Apache RocketMQ 消息隊列部署與可視化界面安裝》
6.1. 引入依賴
使用 spring-cloud-stream 框架來訪問 RocketMQ
Spring Cloud Stream 是一個構(gòu)建消息驅(qū)動的框架钳垮，通過抽象的定義實現(xiàn)應(yīng)用與MQ消息隊列之間的解耦，目前支持
RabbitMQ额港、kafka和RocketMQ
mark
?
6.2. 開啟事務(wù)消息
消息生產(chǎn)者需要添加 transactional: true 開啟 事務(wù)消息
?
6.3. 訂單服務(wù)發(fā)送half消息
因為開啟了
事務(wù)消息所以這里發(fā)送的是half消息對于消費端是不可見的
?
6.4. 訂單服務(wù)監(jiān)聽half消息
使用 @RocketMQTransactionListener 注解監(jiān)聽 半消息，并實現(xiàn) RocketMQLocalTransactionListener 接口歧焦，該接口有兩個方法
- executeLocalTransaction：用于提交本地事務(wù)
- checkLocalTransaction：用于事務(wù)回查
如果提交事務(wù)消息失敗移斩，需等待約1分鐘左右 事務(wù)回查 方法才會被調(diào)用
?
6.5. 積分服務(wù)消費消息
注意：因為有
重試，這里如果是真實的業(yè)務(wù)需要自行實現(xiàn)冪等性
?
6.6. 消費死信隊列預(yù)警
監(jiān)聽并消費死信隊列中的消息绢馍，用于記錄錯誤日志向瓷，并且預(yù)警通知運維人員等
?
6.7. 測試用例
demo中提供了3個接口分別測試不同的場景：
-
事務(wù)成功
http://localhost:11002/success
流程如下：- 訂單創(chuàng)建 成功
- 提交事務(wù)消息 成功
- 消費消息增加積分 成功
-
訂單創(chuàng)建成功但提交事務(wù)消息失敗
http://localhost:11002/produceError
流程如下：- 訂單創(chuàng)建 成功
- 提交事務(wù)消息 失敗
- 事務(wù)回查(等待1分鐘左右) 成功
- 提交事務(wù)消息 成功
- 消費消息增加積分 成功
-
消費消息失敗
http://localhost:11002/consumeError
流程如下：- 訂單創(chuàng)建 成功
- 提交事務(wù)消息 成功
- 消費消息增加積分 失敗
- 重試消費消息 失敗
- 進入死信隊列 成功
- 消費死信隊列的消息 成功
- 記錄日志并發(fā)出預(yù)警 成功
?
七、demo下載地址
?
推薦閱讀
- 日志排查問題困難舰涌？分布式日志鏈路跟蹤來幫你
- zuul集成Sentinel最新的網(wǎng)關(guān)流控組件
- 阿里注冊中心Nacos生產(chǎn)部署方案
- Spring Boot自定義配置項在IDE里面實現(xiàn)自動提示
- Spring Cloud Zuul的動態(tài)路由怎樣做猖任？集成Nacos實現(xiàn)很簡單
- Spring Cloud開發(fā)人員如何解決服務(wù)沖突和實例亂竄？
- Spring Cloud同步場景分布式事務(wù)怎樣做瓷耙？試試Seata
?
掃碼關(guān)注有驚喜朱躺！
