消息無(wú)序產(chǎn)生的原因

消息隊(duì)列，既然是隊(duì)列就能保證消息在進(jìn)入隊(duì)列，以及出隊(duì)列的時(shí)候保證消息的有序性，顯然這是在消息的生產(chǎn)端(Producer)，但是往往在生產(chǎn)環(huán)境中有多個(gè)消息的消費(fèi)端(Consumer)，盡管消費(fèi)端在拉取消息時(shí)是有序的，但各個(gè)消息由于網(wǎng)絡(luò)等方面原因無(wú)法保證在各個(gè)消費(fèi)端中處理時(shí)有序。

創(chuàng)新互聯(lián)公司是一家專業(yè)提供伽師企業(yè)網(wǎng)站建設(shè),專注與網(wǎng)站建設(shè)、成都網(wǎng)站制作、HTML5、小程序制作等業(yè)務(wù)。10年已為伽師眾多企業(yè)、政府機(jī)構(gòu)等服務(wù)。創(chuàng)新互聯(lián)專業(yè)網(wǎng)站制作公司優(yōu)惠進(jìn)行中。

場(chǎng)景分析

先后兩次修改了商品信息，消息A和消息B先后同步寫入MySQL，接著異步寫入消息隊(duì)列中發(fā)送消息，此時(shí)消息隊(duì)列生產(chǎn)端（Producer）按時(shí)序先后發(fā)出了A和B兩條消息(消息A先發(fā)出，消息B后發(fā)出)。按業(yè)務(wù)邏輯，商品信息的最終狀態(tài)需要以消息A和消息B綜合為準(zhǔn)。

看似一個(gè)比較常見的同步寫數(shù)據(jù)庫(kù)，異步發(fā)送消息的場(chǎng)景，但實(shí)際上需要保證消息的有序消費(fèi)。

• 假設(shè)1：消息A只包含修改的商品名稱，消息B只包含修改的商品重量，此時(shí)消息隊(duì)列的消費(fèi)端實(shí)際上不需要關(guān)注消息時(shí)序，消息隊(duì)列消費(fèi)端（Consumer）只管消費(fèi)即可。
• 假設(shè)2：消息A包含修改的商品名稱、重量，消息B包含修改的商品名稱，此時(shí)消費(fèi)端首先接收到消息B，后接收到消息A，那么消息B的修改就會(huì)被覆蓋。此時(shí)消息隊(duì)列的消費(fèi)端實(shí)際上又需要關(guān)注消息時(shí)序。

可見，你無(wú)法保證消息中包含什么信息，此時(shí)必須保證消息的有序消費(fèi)。

業(yè)務(wù)角度如何保證消息有序消費(fèi)

• 生產(chǎn)端在發(fā)送消息時(shí)，始終保證消息是全量信息。
• 消費(fèi)端在接收消息時(shí)，通過(guò)緩存時(shí)間戳的方式，消費(fèi)消息時(shí)判斷消息產(chǎn)生的時(shí)間是否最新，如果不是則丟棄，如果是則執(zhí)行下一步。

下面通過(guò)偽代碼的方式描述：

生產(chǎn)端偽代碼

insertWare(ware); #插入數(shù)據(jù)到數(shù)據(jù)庫(kù)，通常在插入數(shù)據(jù)庫(kù)時(shí)我們只會(huì)update修改的字段，而不會(huì)全量插入

ware = selectWareById(ware.getId); #獲取商品的全量信息（此時(shí)是最新的），用于將它放入到消息隊(duì)列中

syncMq(ware); #異步發(fā)送mq消息A

消費(fèi)端偽代碼

ware = fetchWare(); #獲取消息

if (isLasted(ware)) #通過(guò)商品的修改時(shí)間戳判斷是否是最新的修改

? TODO #執(zhí)行下一步業(yè)務(wù)邏輯

else

? return #丟棄該消息

重點(diǎn)在于消費(fèi)端如何判斷該消息是否是最新的修改也就是isLasted方法。

isLasted方法

Long modified = getCacheById(ware.getId); #獲取緩存中該條商品的最新修改時(shí)間

If (ware.getModified > modified) { #如果消息中商品修改時(shí)間大于緩存中的時(shí)間，說(shuō)明是最新操作

? setCacheById(ware); #將該條消息的商品修改時(shí)間戳寫入到緩存中

? return true;
} else #如果消息中的商品修改時(shí)間小于緩存中的時(shí)間，說(shuō)明該條消息屬于“歷史操作”，不對(duì)其更新

? return false;

以上就是通過(guò)偽代碼的方式，描述如何通過(guò)業(yè)務(wù)手段保證消息有序消費(fèi)，重點(diǎn)在于全量發(fā)送信息和緩存時(shí)間戳。在其中還有一些技術(shù)實(shí)現(xiàn)細(xì)節(jié)。

例如：消費(fèi)端消費(fèi)消息B，執(zhí)行到獲取時(shí)間戳緩存之后，并在重新設(shè)置新的緩存之前，此時(shí)另一個(gè)消費(fèi)端恰好也正在消費(fèi)B它也正執(zhí)行到獲取時(shí)間戳緩存，由于消息A此時(shí)并沒有更新緩存，消息A拿到的緩存仍然是舊的緩存，這時(shí)就會(huì)存在兩個(gè)消費(fèi)端都認(rèn)為自己所消費(fèi)的消息時(shí)最新的，造成該丟棄的消息沒丟。

顯然，這是分布式線程安全問題，分布式鎖通常使用Redis或者ZooKeeper，加鎖后的執(zhí)行時(shí)序如下圖所示。

這是從業(yè)務(wù)角度保證消息在消費(fèi)端有序消費(fèi)。通過(guò)在消息發(fā)送端全量發(fā)送消息以及在消息消費(fèi)端緩存時(shí)間戳就可以保證消息的有序消費(fèi)。

在上述場(chǎng)景中是先同步寫入MySQL，再獲取商品全量數(shù)據(jù)，接著再異步發(fā)送消息。這一系列的步驟可以通過(guò)接MySQL的binlog實(shí)現(xiàn)，在同步寫入MySQL后，MySQL發(fā)送binlog變更，通過(guò)阿里巴巴Canal中間件接收MySQL的binlog變更再發(fā)送消息到消息隊(duì)列。

創(chuàng)新互聯(lián)www.cdcxhl.cn，專業(yè)提供香港、美國(guó)云服務(wù)器，動(dòng)態(tài)BGP最優(yōu)骨干路由自動(dòng)選擇，持續(xù)穩(wěn)定高效的網(wǎng)絡(luò)助力業(yè)務(wù)部署。公司持有工信部辦法的idc、isp許可證， 機(jī)房獨(dú)有T級(jí)流量清洗系統(tǒng)配攻擊溯源，準(zhǔn)確進(jìn)行流量調(diào)度，確保服務(wù)器高可用性。佳節(jié)活動(dòng)現(xiàn)已開啟，新人活動(dòng)云服務(wù)器買多久送多久。

網(wǎng)站名稱：消費(fèi)端如何保證消息隊(duì)列MQ的有序消費(fèi)-創(chuàng)新互聯(lián)
分享地址：http://jinyejixie.com/article6/ccesig.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供定制開發(fā)、電子商務(wù)、微信小程序、動(dòng)態(tài)網(wǎng)站、軟件開發(fā)、外貿(mào)網(wǎng)站建設(shè)

廣告

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