隨著分布式服務(wù)架構(gòu)的流行與普及，原來在單體應(yīng)用中執(zhí)行的多個邏輯操作，現(xiàn)在被拆分成了多個服務(wù)之間的遠(yuǎn)程調(diào)用。雖然服務(wù)化為我們的系統(tǒng)帶來了水平伸縮的能力，然而隨之而來挑戰(zhàn)就是分布式事務(wù)問題，多個服務(wù)之間使用自己單獨維護(hù)的數(shù)據(jù)庫，它們彼此之間不在同一個事務(wù)中，假如A執(zhí)行成功了，B執(zhí)行卻失敗了，而A的事務(wù)此時已經(jīng)提交，無法回滾，那么最終就會導(dǎo)致兩邊數(shù)據(jù)不一致性的問題；盡管很早之前就有基于兩階段提交的XA分布式事務(wù)，但是這類方案因為需要資源的全局鎖定，導(dǎo)致性能極差；因此后面就逐漸衍生出了消息最終一致性、TCC等柔性事務(wù)的分布式事務(wù)方案，本文主要分析的是基于消息的最終一致性方案。

01 普通消息的處理流程

• 消息生成者發(fā)送消息
• MQ收到消息，將消息進(jìn)行持久化，在存儲中新增一條記錄
• 返回ACK給生產(chǎn)者
• MQ push 消息給對應(yīng)的消費者，然后等待消費者返回ACK
• 如果消息消費者在指定時間內(nèi)成功返回ack，那么MQ認(rèn)為消息消費成功，在存儲中刪除消息，即執(zhí)行第6步；如果MQ在指定時間內(nèi)沒有收到ACK，則認(rèn)為消息消費失敗，會嘗試重新push消息,重復(fù)執(zhí)行4、5、6步驟
• MQ刪除消息

1.1 普通消息處理存在的一致性問題

我們以訂單創(chuàng)建為例，訂單系統(tǒng)先創(chuàng)建訂單(本地事務(wù))，再發(fā)送消息給下游處理；如果訂單創(chuàng)建成功，然而消息沒有發(fā)送出去，那么下游所有系統(tǒng)都無法感知到這個事件，會出現(xiàn)臟數(shù)據(jù)；

public void processOrder() {

// 訂單處理(業(yè)務(wù)操作)

orderService.process();

// 發(fā)送訂單處理成功消息(發(fā)送消息)

sendBizMsg ();

}

如果先發(fā)送訂單消息，再創(chuàng)建訂單；那么就有可能消息發(fā)送成功，但是在訂單創(chuàng)建的時候卻失敗了，此時下游系統(tǒng)卻認(rèn)為這個訂單已經(jīng)創(chuàng)建，也會出現(xiàn)臟數(shù)據(jù)。

public void processOrder() { // 發(fā)送訂單處理成功消息(發(fā)送消息) sendBizMsg (); // 訂單處理(業(yè)務(wù)操作) orderService.process();}

1.2 一個錯誤的想法

此時可能有同學(xué)會想，我們可否將消息發(fā)送和業(yè)務(wù)處理放在同一個本地事務(wù)中來進(jìn)行處理，如果業(yè)務(wù)消息發(fā)送失敗，那么本地事務(wù)就回滾，這樣是不是就能解決消息發(fā)送的一致性問題呢?

@Transactionnalpublic void processOrder() { try{ // 訂單處理(業(yè)務(wù)操作) orderService.process(); // 發(fā)送訂單處理成功消息(發(fā)送消息) sendBizMsg (); }catch(Exception e){ 事務(wù)回滾; }}

1.3 消息發(fā)送的異常情況分析

可能的情況 一致性 訂單處理成功，然后突然宕機，事務(wù)未提交，消息沒有發(fā)送出去 一致 訂單處理成功，由于網(wǎng)絡(luò)原因或者M(jìn)Q宕機，消息沒有發(fā)送出去，事務(wù)回滾 一致 訂單處理成功，消息發(fā)送成功，但是MQ由于其他原因，導(dǎo)致消息存儲失敗，事務(wù)回滾 一致 訂單處理成功，消息存儲成功，但是MQ處理超時，從而ACK確認(rèn)失敗，導(dǎo)致發(fā)送方本地事務(wù)回滾 不一致 從上面的情況分析，我們可以看到，使用普通的處理方式，無論如何，都無法保證業(yè)務(wù)處理與消息發(fā)送兩邊的一致性，其根本的原因就在于：遠(yuǎn)程調(diào)用，結(jié)果最終可能為成功、失敗、超時；而對于超時的情況，處理方最終的結(jié)果可能是成功，也可能是失敗，調(diào)用方是無法知曉的。 筆者就曾經(jīng)在項目中出現(xiàn)類似的情況，調(diào)用方先在本地寫數(shù)據(jù)，然后發(fā)起RPC服務(wù)調(diào)用，但是處理方由于DB數(shù)據(jù)量比較大，導(dǎo)致處理超時，調(diào)用方在出現(xiàn)超時異常后，直接回滾本地事務(wù)，從而導(dǎo)致調(diào)用方這邊沒數(shù)據(jù)，而處理方那邊數(shù)據(jù)卻已經(jīng)寫入了，最終導(dǎo)致兩邊業(yè)務(wù)數(shù)據(jù)的不一致。為了保證兩邊數(shù)據(jù)的一致性，我們只能從其他地方尋找新的突破口。

02 事務(wù)消息

由于傳統(tǒng)的處理方式無法解決消息生成者本地事務(wù)處理成功與消息發(fā)送成功兩者的一致性問題，因此事務(wù)消息就誕生了，它實現(xiàn)了消息生成者本地事務(wù)與消息發(fā)送的原子性，保證了消息生成者本地事務(wù)處理成功與消息發(fā)送成功的最終一致性問題。

03 事務(wù)消息處理的流程

• 事務(wù)消息與普通消息的區(qū)別就在于消息生產(chǎn)環(huán)節(jié)，生產(chǎn)者首先預(yù)發(fā)送一條消息到MQ(這也被稱為發(fā)送half消息)
• MQ接受到消息后，先進(jìn)行持久化，則存儲中會新增一條狀態(tài)為待發(fā)送的消息
• 然后返回ACK給消息生產(chǎn)者，此時MQ不會觸發(fā)消息推送事件
• 生產(chǎn)者預(yù)發(fā)送消息成功后，執(zhí)行本地事務(wù)
• 執(zhí)行本地事務(wù)，執(zhí)行完成后，發(fā)送執(zhí)行結(jié)果給MQ
• MQ會根據(jù)結(jié)果刪除或者更新消息狀態(tài)為可發(fā)送
• 如果消息狀態(tài)更新為可發(fā)送，則MQ會push消息給消費者，后面消息的消費和普通消息是一樣的

注意點：由于MQ通常都會保證消息能夠投遞成功，因此，如果業(yè)務(wù)沒有及時返回ACK結(jié)果，那么就有可能造成MQ的重復(fù)消息投遞問題。因此，對于消息最終一致性的方案，消息的消費者必須要對消息的消費支持冪等，不能造成同一條消息的重復(fù)消費的情況。

3.1 事務(wù)消息異常情況分析

異常情況 一致性 處理異常方法 消息未存儲，業(yè)務(wù)操作未執(zhí)行 一致 無 存儲待發(fā)送消息成功，但是ACK失敗，導(dǎo)致業(yè)務(wù)未執(zhí)行(可能是MQ處理超時、網(wǎng)絡(luò)抖動等原因) 不一致 MQ確認(rèn)業(yè)務(wù)操作結(jié)果，處理消息(刪除消息) 存儲待發(fā)送消息成功，ACK成功，業(yè)務(wù)執(zhí)行(可能成功也可能失敗)，但是MQ沒有收到生產(chǎn)者業(yè)務(wù)處理的最終結(jié)果 不一致 MQ確認(rèn)業(yè)務(wù)操作結(jié)果，處理消息(根據(jù)就業(yè)務(wù)處理結(jié)果，更新消息狀態(tài)，如果業(yè)務(wù)執(zhí)行成功，則投遞消息，失敗則刪除消息) 業(yè)務(wù)處理成功，并且發(fā)送結(jié)果給MQ，但是MQ更新消息失敗，導(dǎo)致消息狀態(tài)依舊為待發(fā)送 不一致 同上

04 支持事務(wù)消息的MQ

現(xiàn)在目前較為主流的MQ，比如ActiveMQ、RabbitMQ、Kafka、RocketMQ等，只有RocketMQ支持事務(wù)消息。據(jù)筆者了解，早年阿里對MQ增加事務(wù)消息也是因為支付寶那邊因為業(yè)務(wù)上的需求而產(chǎn)生的。因此，如果我們希望強依賴一個MQ的事務(wù)消息來做到消息最終一致性的話，在目前的情況下，技術(shù)選型上只能去選擇RocketMQ來解決。上面我們也分析了事務(wù)消息所存在的異常情況，即MQ存儲了待發(fā)送的消息，但是MQ無法感知到上游處理的最終結(jié)果。對于RocketMQ而言，它的解決方案非常的簡單，就是其內(nèi)部實現(xiàn)會有一個定時任務(wù)，去輪訓(xùn)狀態(tài)為待發(fā)送的消息，然后給producer發(fā)送check請求，而producer必須實現(xiàn)一個check監(jiān)聽器，監(jiān)聽器的內(nèi)容通常就是去檢查與之對應(yīng)的本地事務(wù)是否成功(一般就是查詢DB)，如果成功了，則MQ會將消息設(shè)置為可發(fā)送，否則就刪除消息。

05 常見的問題

(1)問：如果預(yù)發(fā)送消息失敗，是不是業(yè)務(wù)就不執(zhí)行了？

答：是的，對于基于消息最終一致性的方案，一般都會強依賴這步，如果這個步驟無法得到保證，那么最終也 就不可能做到最終一致性了。

(2)問：為什么要增加一個消息預(yù)發(fā)送機制，增加兩次發(fā)布出去消息的重試機制，為什么不在業(yè)務(wù)成功之后，發(fā)送失敗的話使用一次重試機制？

答：如果業(yè)務(wù)執(zhí)行成功，再去發(fā)消息，此時如果還沒來得及發(fā)消息，業(yè)務(wù)系統(tǒng)就已經(jīng)宕機了，系統(tǒng)重啟后，根本沒有記錄之前是否發(fā)送過消息，這樣就會導(dǎo)致業(yè)務(wù)執(zhí)行成功，消息最終沒發(fā)出去的情況。

(3)問：如果consumer消費失敗，是否需要producer做回滾呢？

答：這里的事務(wù)消息，producer不會因為consumer消費失敗而做回滾，采用事務(wù)消息的應(yīng)用，其所追求的是高可用和最終一致性，消息消費失敗的話，MQ自己會負(fù)責(zé)重推消息，直到消費成功。因此，事務(wù)消息是針對生產(chǎn)端而言的，而消費端，消費端的一致性是通過MQ的重試機制來完成的。

(4)問：如果consumer端因為業(yè)務(wù)異常而導(dǎo)致回滾，那么豈不是兩邊最終無法保證一致性?

答：基于消息的最終一致性方案必須保證消費端在業(yè)務(wù)上的操作沒障礙，它只允許系統(tǒng)異常的失敗，不允許業(yè)務(wù)上的失敗，比如在你業(yè)務(wù)上拋出個NPE之類的問題，導(dǎo)致你消費端執(zhí)行事務(wù)失敗，那就很難做到一致了。

由于并非所有的MQ都支持事務(wù)消息，假如我們不選擇RocketMQ來作為系統(tǒng)的MQ，是否能夠做到消息的最終一致性呢？答案是可以的。

06 基于本地消息的最終一致性

基于本地消息的最終一致性方案的最核心做法就是在執(zhí)行業(yè)務(wù)操作的時候，記錄一條消息數(shù)據(jù)到DB，并且消息數(shù)據(jù)的記錄與業(yè)務(wù)數(shù)據(jù)的記錄必須在同一個事務(wù)內(nèi)完成，這是該方案的前提核心保障。在記錄完成后消息數(shù)據(jù)后，后面我們就可以通過一個定時任務(wù)到DB中去輪訓(xùn)狀態(tài)為待發(fā)送的消息，然后將消息投遞給MQ。這個過程中可能存在消息投遞失敗的可能，此時就依靠重試機制來保證，直到成功收到MQ的ACK確認(rèn)之后，再將消息狀態(tài)更新或者消息清除；而后面消息的消費失敗的話，則依賴MQ本身的重試來完成，其最后做到兩邊系統(tǒng)數(shù)據(jù)的最終一致性?；诒镜叵⒎?wù)的方案雖然可以做到消息的最終一致性，但是它有一個比較嚴(yán)重的弊端，每個業(yè)務(wù)系統(tǒng)在使用該方案時，都需要在對應(yīng)的業(yè)務(wù)庫創(chuàng)建一張消息表來存儲消息。針對這個問題，我們可以將該功能單獨提取出來，做成一個消息服務(wù)來統(tǒng)一處理，因而就衍生出了我們下面將要討論的方案。

07 獨立消息服務(wù)的最終一致性

獨立消息服務(wù)最終一致性與本地消息服務(wù)最終一致性最大的差異就在于將消息的存儲單獨地做成了一個RPC的服務(wù)，這個過程其實就是模擬了事務(wù)消息的消息預(yù)發(fā)送過程，如果預(yù)發(fā)送消息失敗，那么生產(chǎn)者業(yè)務(wù)就不會去執(zhí)行，因此對于生產(chǎn)者的業(yè)務(wù)而言，它是強依賴于該消息服務(wù)的。不過好在獨立消息服務(wù)支持水平擴(kuò)容，因此只要部署多臺，做成HA的集群模式，就能夠保證其可靠性。在消息服務(wù)中，還有一個單獨地定時任務(wù)，它會定期輪訓(xùn)長時間處于待發(fā)送狀態(tài)的消息，通過一個check補償機制來確認(rèn)該消息對應(yīng)的業(yè)務(wù)是否成功，如果對應(yīng)的業(yè)務(wù)處理成功，則將消息修改為可發(fā)送，然后將其投遞給MQ；如果業(yè)務(wù)處理失敗，則將對應(yīng)的消息更新或者刪除即可。因此在使用該方案時，消息生產(chǎn)者必須同時實現(xiàn)一個check服務(wù)，來供消息服務(wù)做消息的確認(rèn)。對于消息的消費，該方案與上面的處理是一樣，都是通過MQ自身的重發(fā)機制來保證消息被消費。

總結(jié)：上游事務(wù)提交之后，在基于MQ的場景下就不考慮回滾了。失敗的可能是由于網(wǎng)絡(luò)、服務(wù)宕機所導(dǎo)致，文章中提到說業(yè)務(wù)上執(zhí)行是無障礙的。如果下游服務(wù)長時間沒有恢復(fù)，那么就應(yīng)該設(shè)置告警，在這里有幾種機制來解決一些牛皮癬類型的問題，假如上游消息始終發(fā)送失敗(這種可能性基本不存在除非代碼是假的)這種情況我們可以設(shè)置報警機制比如發(fā)生異常時可以打印日志，發(fā)送短信，發(fā)送郵件，將異常訂單保存到數(shù)據(jù)庫，這些措施可以同時用于下游一些異常訂單，同時也可以在發(fā)生異常的時候新建一個異常Topic的消息提示，讓人工來介入數(shù)據(jù)訂正。

原文鏈接：https://juejin.im/post/5d8882bdf265da03c9273821

總結(jié)

以上是生活随笔為你收集整理的rocket mq 监听端口_MQ消息最终一致性解决方案的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。