rocketMQ如何避免消息重復(fù)消費(fèi)

在互聯(lián)網(wǎng)應(yīng)用中，尤其在網(wǎng)絡(luò)不穩(wěn)定的情況下，消息隊(duì)列 RocketMQ 的消息有可能會(huì)出現(xiàn)重復(fù)，造成重復(fù)消費(fèi)。

這個(gè)重復(fù)簡(jiǎn)單可以概括為以下情況：

• produce發(fā)送到Broker時(shí)消息重復(fù)

當(dāng)一條消息已被成功發(fā)送到服務(wù)端并完成持久化，此時(shí)出現(xiàn)了網(wǎng)絡(luò)閃斷或者producer宕機(jī)，導(dǎo)致服務(wù)端Broker 對(duì) producer應(yīng)答失敗。

如果此時(shí)生產(chǎn)者意識(shí)到消息發(fā)送失敗并嘗試再次發(fā)送消息，消費(fèi)者后續(xù)會(huì)收到兩條內(nèi)容相同并且 Message ID 也相同的消息。

• Broker投遞消息到Consumer時(shí)消息重復(fù)

消息消費(fèi)的場(chǎng)景下，消息已投遞到消費(fèi)者并完成業(yè)務(wù)處理，當(dāng)客戶(hù)端給服務(wù)端反饋應(yīng)答的時(shí)候網(wǎng)絡(luò)閃斷。

為了保證消息至少被消費(fèi)一次，消息隊(duì)列 RocketMQ 的服務(wù)端將在網(wǎng)絡(luò)恢復(fù)后再次嘗試投遞之前已被處理過(guò)的消息，消費(fèi)者后續(xù)會(huì)收到兩條內(nèi)容相同并且 Message ID 也相同的消息。

• 負(fù)載均衡時(shí)消息重復(fù)（包括但不限于網(wǎng)絡(luò)抖動(dòng)、Broker 重啟以及訂閱方應(yīng)用重啟）

當(dāng)消息隊(duì)列 RocketMQ 的 Broker或客戶(hù)端重啟、擴(kuò)容或縮容時(shí)，會(huì)觸發(fā)Rebalance重平衡機(jī)制，此時(shí)消費(fèi)者可能會(huì)收到重復(fù)消息。

冪等性解決方案

冪等性：就是用戶(hù)對(duì)于同一操作發(fā)起的一次請(qǐng)求或者多次請(qǐng)求的結(jié)果是一致的，不會(huì)因?yàn)槎啻吸c(diǎn)擊而產(chǎn)生了副作用。

從上面的分析中，我們知道，在RocketMQ中，是無(wú)法保證每個(gè)消息只被投遞一次的，所以要在業(yè)務(wù)上自行來(lái)保證消息消費(fèi)的冪等性。而要處理這個(gè)問(wèn)題，RocketMQ的每條消息都有一個(gè)唯一的MessageId，這個(gè)參數(shù)在多次投遞的過(guò)程中是不會(huì)改變的，所以業(yè)務(wù)上可以用這個(gè)MessageId來(lái)作為判斷冪等的關(guān)鍵依據(jù)。

但是，這個(gè)MessageId是無(wú)法保證全局唯一的，也會(huì)有沖突的情況。所以在一些對(duì)冪等性要求嚴(yán)格的場(chǎng)景，最好是使用業(yè)務(wù)上唯一的一個(gè)標(biāo)識(shí)比較靠譜。這個(gè)id可以使用分布式中間件redis，zookeeper等去生成。例如訂單ID。而這個(gè)業(yè)務(wù)標(biāo)識(shí)可以使用Message的Key來(lái)進(jìn)行傳遞。

同時(shí)，建立一個(gè)消息表，拿到這個(gè)消息做數(shù)據(jù)庫(kù)的insert操作。給這個(gè)消息做一個(gè)唯一主鍵（primary key）或者唯一約束，那么就算出現(xiàn)重復(fù)消費(fèi)的情況，就會(huì)導(dǎo)致主鍵沖突，那么就不再處理這條消息。

消息持久層做消息唯一性的策略（該方案未經(jīng)證實(shí)）

1.持久化過(guò)程中業(yè)務(wù)唯一標(biāo)識(shí)驗(yàn)證，每個(gè)消息具有業(yè)務(wù)唯一標(biāo)識(shí)，在消息最終持久化之前通過(guò)驗(yàn)證唯一性標(biāo)識(shí)保證消息的唯一性。消息持久化位置如果出現(xiàn)同樣的消息，系統(tǒng)就不做處理，期間無(wú)任何傳遞過(guò)程，保證消息的唯一性。

2.使用過(guò)程中業(yè)務(wù)唯一標(biāo)識(shí)驗(yàn)證，使用過(guò)程中如果出現(xiàn)同樣的消息，系統(tǒng)進(jìn)行相應(yīng)的異常處理。

rocketMQ消息重復(fù)消費(fèi)場(chǎng)景

答案

將去重操作直接放在了消費(fèi)端，消費(fèi)端處理消息的業(yè)務(wù)邏輯保持冪等性。消費(fèi)者收到消息后，從消息中獲取消息標(biāo)識(shí)寫(xiě)入到Redis（分布式鎖）或數(shù)據(jù)庫(kù)（標(biāo)識(shí)作為表唯一索引插入一條記錄），當(dāng)再次收到該消息時(shí)就不作處理。

在broker端對(duì)Queue加鎖（synchronized），Consumer監(jiān)聽(tīng)的Queue存在已投遞但未收到ack且未超時(shí)的消息，不允許獲取鎖，直到該Queue投遞的消息全部ack或者消費(fèi)超時(shí)，才允許新的Consumer獲取鎖，拉取消息。

思考問(wèn)題

• 1、為什么不在生產(chǎn)者去重？
• 2、為什么在消費(fèi)者做去重？

一、網(wǎng)絡(luò)波動(dòng)導(dǎo)致系統(tǒng)A消息發(fā)送到RocketMQ后沒(méi)有收到消息發(fā)送超時(shí)，系統(tǒng)A重試導(dǎo)致消息重復(fù)  

1、RocketMQ支持消息查詢(xún)的功能，消息發(fā)送前去RocketMQ查詢(xún)一下是否已經(jīng)發(fā)送過(guò)該條消息，存在則不發(fā)送，不存在發(fā)送到RocketMQ。在高并發(fā)的場(chǎng)景下，每條消息在發(fā)送到RocketMQ時(shí)都去查詢(xún)一下，會(huì)影響接口的性能。

2、redis分布式鎖，在發(fā)送消息到RocketMQ成功之后，向redis中插入一條數(shù)據(jù)，如果發(fā)生重試，則先去redis中查詢(xún)是否存在，存在的話(huà)不再發(fā)送消息。redis集群此時(shí)宕機(jī)，再次查詢(xún)r(jià)edis判斷消息是否已經(jīng)發(fā)送過(guò)，無(wú)法得到正確結(jié)果的。

以上兩種方式只是保證只發(fā)送了一次，不能保證消費(fèi)只一次的情況。

二、MQ要保證消息投遞的可靠性，對(duì)未ack的消息，會(huì)重復(fù)投遞。

• 場(chǎng)景一：broker發(fā)送Consumer超時(shí)后重新發(fā)送

消費(fèi)者端要保證消費(fèi)的冪等性，從消息中獲取消息標(biāo)識(shí)寫(xiě)入到Redis或數(shù)據(jù)庫(kù)，當(dāng)再次收到該消息時(shí)不作處理。

• 場(chǎng)景二：負(fù)載均衡階段，前一個(gè)監(jiān)聽(tīng)Queue的消費(fèi)實(shí)例拉取的消息未全部ack，新的消費(fèi)實(shí)例監(jiān)聽(tīng)到這個(gè)Queue重新拉取消息。

在負(fù)載均衡結(jié)果變化過(guò)程增加了一個(gè)過(guò)渡態(tài)，在過(guò)渡態(tài)的時(shí)候，Consumer會(huì)繼續(xù)保留上一次負(fù)載均衡的結(jié)果，直到原消費(fèi)者拉取的消息全部ack，才釋放老的結(jié)果。

在broker端對(duì)Queue加鎖（synchronized），Consumer監(jiān)聽(tīng)的Queue存在已投遞但未收到ack且未超時(shí)的消息，不允許獲取鎖，直到該Queue投遞的消息全部ack或者消費(fèi)超時(shí)，才允許新的Consumer獲取鎖，拉取消息。

總結(jié)

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評(píng)論