這篇文章主要為大家展示了“如何使用RocketMQ的事務(wù)消息來解決一致性問題”，內(nèi)容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領(lǐng)大家一起研究并學(xué)習(xí)一下“如何使用RocketMQ的事務(wù)消息來解決一致性問題”這篇文章吧。

創(chuàng)新互聯(lián)建站是網(wǎng)站建設(shè)專家,致力于互聯(lián)網(wǎng)品牌建設(shè)與網(wǎng)絡(luò)營銷，專業(yè)領(lǐng)域包括成都網(wǎng)站設(shè)計(jì)、成都網(wǎng)站建設(shè)、電商網(wǎng)站制作開發(fā)、微信小程序定制開發(fā)、微信營銷、系統(tǒng)平臺(tái)開發(fā)，與其他網(wǎng)站設(shè)計(jì)及系統(tǒng)開發(fā)公司不同，我們的整合解決方案結(jié)合了恒基網(wǎng)絡(luò)品牌建設(shè)經(jīng)驗(yàn)和互聯(lián)網(wǎng)整合營銷的理念，并將策略和執(zhí)行緊密結(jié)合，且不斷評(píng)估并優(yōu)化我們的方案，為客戶提供全方位的互聯(lián)網(wǎng)品牌整合方案！

一、背景

在微服務(wù)架構(gòu)中，我們常常使用異步化的手段來提升系統(tǒng)的 吞吐量 和 解耦 上下游，而構(gòu)建異步架構(gòu)最常用的手段就是使用 消息隊(duì)列(MQ)，那異步架構(gòu)怎樣才能實(shí)現(xiàn)數(shù)據(jù)一致性呢？

二、MQ選型

可以看到在 業(yè)務(wù)處理 方面來說 RocketMQ 優(yōu)于其他對(duì)手，而且原生支持 事務(wù)消息

PS：業(yè)務(wù)系統(tǒng)用的是其他 MQ 產(chǎn)品但是又需要 事務(wù)消息 怎么辦？學(xué)習(xí)原理自己開發(fā)實(shí)現(xiàn)！

三、什么是事務(wù)消息

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

我們是應(yīng)該先 創(chuàng)建訂單記錄，還是先 發(fā)送MQ消息 呢？

1. 先發(fā)送MQ消息：這個(gè)明顯是不行的，因?yàn)槿绻l(fā)送成功，而訂單創(chuàng)建失敗的話是沒辦法把消息收回來的

2. 先創(chuàng)建訂單記錄：如果訂單創(chuàng)建成功后MQ消息發(fā)送失敗 拋出異常，因?yàn)閮蓚€(gè)操作都在本地事務(wù)中所以訂單數(shù)據(jù)是可以 回滾 的

上面的 方式二 看似沒問題，但是 網(wǎng)絡(luò)是不可靠的！如果 MQ 的響應(yīng)因?yàn)榫W(wǎng)絡(luò)原因沒有收到，所以在面對(duì)不確定的結(jié)果只好進(jìn)行回滾；但是 MQ 端又確實(shí)是收到了這條消息的，只是回給客戶端的 響應(yīng)丟失 了！
 
所以 事務(wù)消息 就是用來保證 本地事務(wù) 與 MQ消息發(fā)送 的原子性！

四、RocketMQ事務(wù)消息原理

主要的邏輯分為兩個(gè)流程：

• 事務(wù)消息發(fā)送及提交：

1. 發(fā)送 half消息

2. MQ服務(wù)端 響應(yīng)消息寫入結(jié)果

3. 根據(jù)發(fā)送結(jié)果執(zhí)行 本地事務(wù)（如果寫入失敗，此時(shí)half消息對(duì)業(yè)務(wù) 不可見，本地邏輯不執(zhí)行）

4. 根據(jù)本地事務(wù)狀態(tài)執(zhí)行 Commit 或者 Rollback（Commit操作生成消息索引，消息對(duì)消費(fèi)者 可見）
    

• 回查流程：

1. 對(duì)于長時(shí)間沒有 Commit/Rollback 的事務(wù)消息（pending 狀態(tài)的消息），從服務(wù)端發(fā)起一次 回查

2. Producer 收到回查消息，檢查回查消息對(duì)應(yīng)的 本地事務(wù)狀態(tài)

3. 根據(jù)本地事務(wù)狀態(tài)，重新 Commit 或者 Rollback

 
邏輯時(shí)序圖

五、異步架構(gòu)一致性實(shí)現(xiàn)思路

從上面的原理可以發(fā)現(xiàn) 事務(wù)消息 僅僅只是保證本地事務(wù)和MQ消息發(fā)送形成整體的 原子性，而投遞到MQ服務(wù)器后，并無法保證消費(fèi)者一定能消費(fèi)成功！
 
如果 消費(fèi)端消費(fèi)失敗 后的處理方式，建議是記錄異常信息然后 人工處理，并不建議回滾上游服務(wù)的數(shù)據(jù)(因?yàn)閮烧呤?nbsp;解耦 的，而且 復(fù)雜度 太高)
 
我們可以利用 MQ 的兩個(gè)特性 重試 和 死信隊(duì)列 來協(xié)助消費(fèi)端處理：

1. 消費(fèi)失敗后進(jìn)行一定次數(shù)的 重試

2. 重試后也失敗的話該消息丟進(jìn) 死信隊(duì)列 里

3. 另外起一個(gè)線程監(jiān)聽消費(fèi) 死信隊(duì)列 里的消息，記錄日志并且預(yù)警！

因?yàn)橛?nbsp;重試 所以消費(fèi)者需要實(shí)現(xiàn)冪等性

六、分布式事務(wù)場景樣例

下面就用剛剛提到的場景：生成訂單記錄 -> MQ -> 增加積分；來簡單講一下 Spring Cloud 中應(yīng)該怎么做，詳細(xì)代碼請(qǐng) 下載demo 查看。
PS：怎樣安裝部署RocketMQ可以參考《Apache RocketMQ 消息隊(duì)列部署與可視化界面安裝》

6.1. 引入依賴

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

Spring Cloud Stream 是一個(gè)構(gòu)建消息驅(qū)動(dòng)的框架，通過抽象的定義實(shí)現(xiàn)應(yīng)用與MQ消息隊(duì)列之間的解耦，目前支持 RabbitMQ、kafka 和 RocketMQ

6.2. 開啟事務(wù)消息

消息生產(chǎn)者需要添加 transactional: true 開啟 事務(wù)消息

6.3. 訂單服務(wù)發(fā)送half消息

因?yàn)殚_啟了 事務(wù)消息 所以這里發(fā)送的是 half消息 對(duì)于消費(fèi)端是 不可見 的

6.4. 訂單服務(wù)監(jiān)聽half消息

使用 @RocketMQTransactionListener 注解監(jiān)聽 半消息，并實(shí)現(xiàn) RocketMQLocalTransactionListener 接口，該接口有兩個(gè)方法

• executeLocalTransaction：用于提交本地事務(wù)

• checkLocalTransaction：用于事務(wù)回查

如果提交事務(wù)消息失敗，需等待約1分鐘左右 事務(wù)回查 方法才會(huì)被調(diào)用

6.5. 積分服務(wù)消費(fèi)消息

注意：因?yàn)橛?重試 這里如果是真實(shí)的業(yè)務(wù)需要自行實(shí)現(xiàn) 冪等性

6.6. 消費(fèi)死信隊(duì)列預(yù)警

監(jiān)聽并消費(fèi)死信隊(duì)列中的消息，用于記錄錯(cuò)誤日志，并且預(yù)警通知運(yùn)維人員等

6.7. 測試用例

demo中提供了3個(gè)接口分別測試不同的場景：

• 事務(wù)成功
  http://localhost:11002/success
  流程如下：

1. 訂單創(chuàng)建 成功

2. 提交事務(wù)消息 成功

3. 消費(fèi)消息增加積分 成功

• 訂單創(chuàng)建成功但提交事務(wù)消息失敗
  http://localhost:11002/produceError
  流程如下：

1. 訂單創(chuàng)建 成功

2. 提交事務(wù)消息 失敗

3. 事務(wù)回查(等待1分鐘左右) 成功

4. 提交事務(wù)消息 成功

5. 消費(fèi)消息增加積分 成功

• 消費(fèi)消息失敗
  http://localhost:11002/consumeError
  流程如下：

1. 訂單創(chuàng)建 成功

2. 提交事務(wù)消息 成功

3. 消費(fèi)消息增加積分 失敗

4. 重試消費(fèi)消息 失敗

5. 進(jìn)入死信隊(duì)列 成功

6. 消費(fèi)死信隊(duì)列的消息 成功

7. 記錄日志并發(fā)出預(yù)警 成功

以上是“如何使用RocketMQ的事務(wù)消息來解決一致性問題”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對(duì)大家有所幫助，如果還想學(xué)習(xí)更多知識(shí)，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！

網(wǎng)站欄目：如何使用RocketMQ的事務(wù)消息來解決一致性問題
網(wǎng)頁URL：http://www.chinadenli.net/article2/igpjoc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供小程序開發(fā)、網(wǎng)站制作、網(wǎng)頁設(shè)計(jì)公司、移動(dòng)網(wǎng)站建設(shè)、網(wǎng)站導(dǎo)航、電子商務(wù)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)