如何使用RocketMQ的事務(wù)消息來(lái)解決一致性問(wèn)題
這篇文章主要為大家展示了“如何使用RocketMQ的事務(wù)消息來(lái)解決一致性問(wèn)題”，內(nèi)容簡(jiǎn)而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領(lǐng)大家一起研究并學(xué)習(xí)一下“如何使用RocketMQ的事務(wù)消息來(lái)解決一致性問(wèn)題”這篇文章吧。
創(chuàng)新互聯(lián)建站是網(wǎng)站建設(shè)專家,致力于互聯(lián)網(wǎng)品牌建設(shè)與網(wǎng)絡(luò)營(yíng)銷，專業(yè)領(lǐng)域包括成都網(wǎng)站設(shè)計(jì)、成都網(wǎng)站建設(shè)、電商網(wǎng)站制作開(kāi)發(fā)、微信小程序定制開(kāi)發(fā)、微信營(yíng)銷、系統(tǒng)平臺(tái)開(kāi)發(fā)，與其他網(wǎng)站設(shè)計(jì)及系統(tǒng)開(kāi)發(fā)公司不同，我們的整合解決方案結(jié)合了恒基網(wǎng)絡(luò)品牌建設(shè)經(jīng)驗(yàn)和互聯(lián)網(wǎng)整合營(yíng)銷的理念，并將策略和執(zhí)行緊密結(jié)合，且不斷評(píng)估并優(yōu)化我們的方案，為客戶提供全方位的互聯(lián)網(wǎng)品牌整合方案！
一、背景
在微服務(wù)架構(gòu)中，我們常常使用異步化的手段來(lái)提升系統(tǒng)的 吞吐量 和 解耦 上下游，而構(gòu)建異步架構(gòu)最常用的手段就是使用 消息隊(duì)列(MQ)，那異步架構(gòu)怎樣才能實(shí)現(xiàn)數(shù)據(jù)一致性呢？
二、MQ選型
可以看到在 業(yè)務(wù)處理 方面來(lái)說(shuō) RocketMQ 優(yōu)于其他對(duì)手，而且原生支持 事務(wù)消息
PS：業(yè)務(wù)系統(tǒng)用的是其他 MQ 產(chǎn)品但是又需要 事務(wù)消息 怎么辦？學(xué)習(xí)原理自己開(kāi)發(fā)實(shí)現(xiàn)！
三、什么是事務(wù)消息
例如下圖的場(chǎng)景：生成訂單記錄 -> MQ -> 增加積分
我們是應(yīng)該先 創(chuàng)建訂單記錄，還是先 發(fā)送MQ消息 呢？
先發(fā)送MQ消息：這個(gè)明顯是不行的，因?yàn)槿绻l(fā)送成功，而訂單創(chuàng)建失敗的話是沒(méi)辦法把消息收回來(lái)的
先創(chuàng)建訂單記錄：如果訂單創(chuàng)建成功后MQ消息發(fā)送失敗 拋出異常，因?yàn)閮蓚€(gè)操作都在本地事務(wù)中所以訂單數(shù)據(jù)是可以 回滾 的
上面的 方式二 看似沒(méi)問(wèn)題，但是 網(wǎng)絡(luò)是不可靠的！如果 MQ 的響應(yīng)因?yàn)榫W(wǎng)絡(luò)原因沒(méi)有收到，所以在面對(duì)不確定的結(jié)果只好進(jìn)行回滾；但是 MQ 端又確實(shí)是收到了這條消息的，只是回給客戶端的 響應(yīng)丟失 了！
所以 事務(wù)消息 就是用來(lái)保證 本地事務(wù) 與 MQ消息發(fā)送 的原子性！
四、RocketMQ事務(wù)消息原理
主要的邏輯分為兩個(gè)流程：
事務(wù)消息發(fā)送及提交：
發(fā)送
half消息MQ服務(wù)端響應(yīng)消息寫入結(jié)果根據(jù)發(fā)送結(jié)果執(zhí)行
本地事務(wù)（如果寫入失敗，此時(shí)half消息對(duì)業(yè)務(wù) 不可見(jiàn)，本地邏輯不執(zhí)行）根據(jù)本地事務(wù)狀態(tài)執(zhí)行
Commit或者Rollback（Commit操作生成消息索引，消息對(duì)消費(fèi)者 可見(jiàn)）
回查流程：
對(duì)于長(zhǎng)時(shí)間沒(méi)有
Commit/Rollback的事務(wù)消息（pending狀態(tài)的消息），從服務(wù)端發(fā)起一次 回查Producer收到回查消息，檢查回查消息對(duì)應(yīng)的本地事務(wù)狀態(tài)根據(jù)本地事務(wù)狀態(tài)，重新
Commit或者Rollback
邏輯時(shí)序圖
五、異步架構(gòu)一致性實(shí)現(xiàn)思路
從上面的原理可以發(fā)現(xiàn) 事務(wù)消息 僅僅只是保證本地事務(wù)和MQ消息發(fā)送形成整體的 原子性，而投遞到MQ服務(wù)器后，并無(wú)法保證消費(fèi)者一定能消費(fèi)成功！
如果 消費(fèi)端消費(fèi)失敗 后的處理方式，建議是記錄異常信息然后 人工處理，并不建議回滾上游服務(wù)的數(shù)據(jù)(因?yàn)閮烧呤?nbsp;解耦 的，而且 復(fù)雜度 太高)
我們可以利用 MQ 的兩個(gè)特性 重試 和 死信隊(duì)列 來(lái)協(xié)助消費(fèi)端處理：
消費(fèi)失敗后進(jìn)行一定次數(shù)的
重試重試后也失敗的話該消息丟進(jìn)
死信隊(duì)列里另外起一個(gè)線程監(jiān)聽(tīng)消費(fèi)
死信隊(duì)列里的消息，記錄日志并且預(yù)警！
因?yàn)橛?nbsp;重試 所以消費(fèi)者需要實(shí)現(xiàn)冪等性
六、分布式事務(wù)場(chǎng)景樣例
下面就用剛剛提到的場(chǎng)景：生成訂單記錄 -> MQ -> 增加積分；來(lái)簡(jiǎn)單講一下 Spring Cloud 中應(yīng)該怎么做，詳細(xì)代碼請(qǐng) 下載demo 查看。
PS：怎樣安裝部署RocketMQ可以參考《Apache RocketMQ 消息隊(duì)列部署與可視化界面安裝》
6.1. 引入依賴
使用 spring-cloud-stream 框架來(lái)訪問(wèn) RocketMQ
Spring Cloud Stream 是一個(gè)構(gòu)建消息驅(qū)動(dòng)的框架，通過(guò)抽象的定義實(shí)現(xiàn)應(yīng)用與MQ消息隊(duì)列之間的解耦，目前支持
RabbitMQ、kafka和RocketMQ
6.2. 開(kāi)啟事務(wù)消息
消息生產(chǎn)者需要添加 transactional: true 開(kāi)啟 事務(wù)消息
6.3. 訂單服務(wù)發(fā)送half消息
因?yàn)殚_(kāi)啟了
事務(wù)消息所以這里發(fā)送的是half消息對(duì)于消費(fèi)端是不可見(jiàn)的
6.4. 訂單服務(wù)監(jiān)聽(tīng)half消息
使用 @RocketMQTransactionListener 注解監(jiān)聽(tīng) 半消息，并實(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)聽(tīng)并消費(fèi)死信隊(duì)列中的消息，用于記錄錯(cuò)誤日志，并且預(yù)警通知運(yùn)維人員等
6.7. 測(cè)試用例
demo中提供了3個(gè)接口分別測(cè)試不同的場(chǎng)景：
事務(wù)成功
http://localhost:11002/success
流程如下：訂單創(chuàng)建 成功
提交事務(wù)消息 成功
消費(fèi)消息增加積分 成功
訂單創(chuàng)建成功但提交事務(wù)消息失敗
http://localhost:11002/produceError
流程如下：訂單創(chuàng)建 成功
提交事務(wù)消息 失敗
事務(wù)回查(等待1分鐘左右) 成功
提交事務(wù)消息 成功
消費(fèi)消息增加積分 成功
消費(fèi)消息失敗
http://localhost:11002/consumeError
流程如下：訂單創(chuàng)建 成功
提交事務(wù)消息 成功
消費(fèi)消息增加積分 失敗
重試消費(fèi)消息 失敗
進(jìn)入死信隊(duì)列 成功
消費(fèi)死信隊(duì)列的消息 成功
記錄日志并發(fā)出預(yù)警 成功
以上是“如何使用RocketMQ的事務(wù)消息來(lái)解決一致性問(wèn)題”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對(duì)大家有所幫助，如果還想學(xué)習(xí)更多知識(shí)，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！
分享文章：如何使用RocketMQ的事務(wù)消息來(lái)解決一致性問(wèn)題
本文路徑：http://m.jiaotiyi.com/article/igpjoc.html