如何處理消費(fèi)過(guò)程中的重復(fù)消息瑞客論壇

1、06-如何處理消費(fèi)過(guò)程中的重復(fù)消息？你好，我是。上節(jié)課講了如何確保消息不會(huì)丟失，課后我給你留了個(gè)思考題，如果消息重復(fù)了怎么辦？這節(jié)課，就來(lái)聊聊如何處理重復(fù)消息。在消息傳遞過(guò)程中，如果出現(xiàn)傳遞失敗的情況，發(fā)送方會(huì)執(zhí)行重試，重試的過(guò)程中就有可能會(huì)產(chǎn)生重復(fù)的消息。對(duì)使用消息隊(duì)列的業(yè)務(wù)系統(tǒng)來(lái)說(shuō)，如果沒有對(duì)重復(fù)消息進(jìn)行處理，就有可能會(huì)導(dǎo)致系統(tǒng)的數(shù)據(jù)出現(xiàn)錯(cuò)誤。比如說(shuō)，個(gè)消費(fèi)訂單消息，統(tǒng)計(jì)下單金額的微服務(wù)，如果沒有正確處理重復(fù)消息，那就會(huì)出現(xiàn)重復(fù)統(tǒng)計(jì)，導(dǎo)致統(tǒng)計(jì)結(jié)果錯(cuò)誤。你可能會(huì)問(wèn)，如果消息隊(duì)列本身能保證消息不重復(fù)，那應(yīng)用程序的實(shí)現(xiàn)不就簡(jiǎn)單了？那有沒有消息隊(duì)列能保證消息不重復(fù)呢？消息重復(fù)的情況必然存在在MQT

2、T協(xié)議中，給出了三種傳遞消息時(shí)能夠提供的服務(wù)質(zhì)量標(biāo)準(zhǔn)，這三種服務(wù)質(zhì)量從低到高依次是：At most once: 至多次。消息在傳遞時(shí)，最多會(huì)被送達(dá)次。換個(gè)說(shuō)法就是，沒什么消息可靠性保證，允許丟消息。般都是些對(duì)消息可靠性要求不太高的數(shù)據(jù)，可以接受數(shù)據(jù)少量丟失。場(chǎng)景使用，比如每分鐘上報(bào)次機(jī)房溫度east once: 至少次。消息在傳遞時(shí)，至少會(huì)被送達(dá)次。也就是說(shuō)，不允許丟消息，但是允許有少量重復(fù)消息出現(xiàn)。Exactly once：恰好次。消息在傳遞時(shí)，只會(huì)被送達(dá)次，不允許丟失也不允許重復(fù)，這個(gè)是最高的等級(jí)。這個(gè)服務(wù)質(zhì)量標(biāo)準(zhǔn)不僅適用于MQTT，對(duì)所有的消息隊(duì)列都是適用的?，F(xiàn)在常用的絕大部分消息隊(duì)列提

3、供的服務(wù)質(zhì)量都是保證消息不重復(fù)。east once，包括RocketMQ、RabbitMQ和Kafka 都是這樣。也就是說(shuō)，消息隊(duì)列很難說(shuō)到這兒我知道肯定有的同學(xué)會(huì)反駁我：“你說(shuō)的不對(duì)，我看過(guò)Kafka的文檔，Kafka是支持Exactly once的?！蔽以谶@里跟這些同學(xué)解釋下，你說(shuō)的沒錯(cuò)，Kafka的確是支持Exactly once，但是我講的也沒有問(wèn)題，為什么呢？Kafka支持的“Exactly once”和剛剛提到的消息傳遞的服務(wù)質(zhì)量標(biāo)準(zhǔn)“Exactly once”是不樣的，它是Kafka提供的另外個(gè)特性，Kafka中支持的事務(wù)也和通常意義理解的事務(wù)有定的差異。在Kafka中，事務(wù)和E

4、xcactly once主要是為了配合流計(jì)算使用的特性，的節(jié)課來(lái)講Kafka的事務(wù)和它支持的Exactly once特性。在專欄“進(jìn)階篇”這個(gè)模塊中，會(huì)有專門稍微說(shuō)些題外話，Kafka的團(tuán)隊(duì)是個(gè)非常包裝和的團(tuán)隊(duì)，你看他們很巧妙地用了兩個(gè)所有人都非常熟悉的概念“事務(wù)”和“Exactly once”來(lái)包裝它的新的特性，實(shí)際上它實(shí)現(xiàn)的這個(gè)事務(wù)和Exactly once并不是雖然和通常理解的那兩個(gè)特性，但是你深入了解Kafka的事務(wù)和Exactly once后，會(huì)發(fā)現(xiàn)其實(shí)它這個(gè)特性通常的理解不樣，但確實(shí)和事務(wù)、Exactly once有定關(guān)系。這點(diǎn)上，都要學(xué)習(xí)Kafka團(tuán)隊(duì)。個(gè)優(yōu)秀的開發(fā)團(tuán)隊(duì)，不僅要能

5、寫代碼，更要能寫文檔，能寫Slide（PPT），還要能講，會(huì)。對(duì)于每個(gè)程序員來(lái)說(shuō)，也是樣的。把話題收回來(lái)，繼續(xù)來(lái)說(shuō)重復(fù)消息。既然消息隊(duì)列無(wú)法保證消息不重復(fù)，就需要的消費(fèi)代碼能夠接受“消息是可能會(huì)重復(fù)的”這現(xiàn)狀，然后，通過(guò)些方法來(lái)消除重復(fù)消息對(duì)業(yè)務(wù)的影響。用冪等性解決重復(fù)消息問(wèn)題般解決重復(fù)消息的辦法是，在消費(fèi)端，讓消費(fèi)消息的操作具備冪等性。冪等（Idempotence） 本來(lái)是個(gè)數(shù)學(xué)上的概念，它是這樣定義的：如果個(gè)函數(shù)f(x)滿足：f(f(x) = f(x)，則函數(shù)f(x)滿足冪等性。這個(gè)概念被拓展到計(jì)算機(jī)領(lǐng)域，被執(zhí)行所產(chǎn)生的影響均與次執(zhí)行的影響相同。個(gè)操作、方法或者服務(wù)。個(gè)冪等操作的特點(diǎn)是，其

6、任意多次個(gè)冪等的方法，使用同樣的參數(shù)，對(duì)它進(jìn)行多次調(diào)用和次調(diào)用，對(duì)系統(tǒng)產(chǎn)生的影響是樣的。所以，對(duì)于冪等的方法，不用擔(dān)心重復(fù)執(zhí)行會(huì)對(duì)系統(tǒng)造成任何改變。舉個(gè)例子來(lái)說(shuō)明下。在不考慮并發(fā)的情況下，“將賬戶X的余額設(shè)置為100元”，執(zhí)行次后對(duì)系統(tǒng)的影響是，賬戶X的余額變成了100元。只要提供的參數(shù)100元不變，那即使再執(zhí)行多少次，賬戶X的余額始終都是100元，不會(huì)變化，這個(gè)操作就是個(gè)冪等的操作。再舉個(gè)例子，“將賬戶X的余額加100元”，這個(gè)操作它就不是冪等的，每執(zhí)行次，賬戶余額就會(huì)增加100元，執(zhí)行多次和執(zhí)行次對(duì)系統(tǒng)的影響（也就是賬戶的余額）是不樣的。如果系統(tǒng)消費(fèi)消息的業(yè)務(wù)邏輯具備冪等性，那就不用擔(dān)心消

7、息重復(fù)了，因?yàn)橥瑮l消息，消費(fèi)次和消費(fèi)多次對(duì)系統(tǒng)的影響是完全樣的。也就可以認(rèn)為，消費(fèi)多次等于消費(fèi)次。從對(duì)系統(tǒng)的影響結(jié)果來(lái)說(shuō)：east once + 冪等消費(fèi) = Exactly once。那么如何實(shí)現(xiàn)冪等操作呢？最好的方式就是，從業(yè)務(wù)邏輯設(shè)計(jì)上入手，將消費(fèi)的業(yè)務(wù)邏輯設(shè)計(jì)成具備冪等性的操作。但是，不是所有的業(yè)務(wù)都能設(shè)計(jì)成天然冪等的，這里就需要些方法和技巧來(lái)實(shí)現(xiàn)冪等。下面我給你介紹幾種常用的設(shè)計(jì)冪等操作的方法：1. 利用數(shù)據(jù)庫(kù)的唯約束實(shí)現(xiàn)冪等例如剛剛提到的那個(gè)不具備冪等特性的轉(zhuǎn)賬的例子：將賬戶X的余額加100元。在這個(gè)例子中，可以通過(guò)改造業(yè)務(wù)邏輯，讓它具備冪等性。首先，可以限定，對(duì)于每個(gè)轉(zhuǎn)賬單每個(gè)賬

8、戶只可以執(zhí)行次變更操作，在分布式系統(tǒng)中，這個(gè)限制實(shí)現(xiàn)的方法非常多，最簡(jiǎn)單的是在數(shù)據(jù)庫(kù)中建張轉(zhuǎn)賬流水表，這個(gè)表有三個(gè)字段：轉(zhuǎn)賬單ID、賬戶ID和變更金額，然后給轉(zhuǎn)賬單ID和賬戶ID這兩個(gè)字段創(chuàng)建個(gè)唯約束，這樣對(duì)于相同的轉(zhuǎn)賬單ID和賬戶ID，表里至多只能存在條。這樣，消費(fèi)消息的邏輯可以變?yōu)椋骸霸谵D(zhuǎn)賬流水表中增加條轉(zhuǎn)賬，然后再根據(jù)轉(zhuǎn)賬，異步操作更新用戶余額即可?！痹谵D(zhuǎn)賬流水表增加條轉(zhuǎn)賬這個(gè)操作中，由于在這個(gè)表中預(yù)先定義了“賬戶ID轉(zhuǎn)賬單ID”的唯約束，對(duì)于同個(gè)轉(zhuǎn)賬單同個(gè)賬戶只能條，后續(xù)重復(fù)的操作都會(huì)失敗，這樣就實(shí)現(xiàn)了個(gè)冪等的操作。只要寫個(gè)SQL，正確地實(shí)現(xiàn)它就可以了。基于這個(gè)思路，不光是可以使用關(guān)系

9、型數(shù)據(jù)庫(kù)，只要是支持類似“INSERT IF NOT EXIST”語(yǔ)義的類系統(tǒng)都可以用于實(shí)現(xiàn)冪等，比如，你可以用Redis的SETNX命令來(lái)替代數(shù)據(jù)庫(kù)中的唯約束，來(lái)實(shí)現(xiàn)冪等消費(fèi)。2. 為更新的數(shù)據(jù)設(shè)置前置條件另外種實(shí)現(xiàn)冪等的思路是，給數(shù)據(jù)變更設(shè)置個(gè)前置條件，如果滿足條件就更新數(shù)據(jù)，否則更新數(shù)據(jù)，在更新數(shù)據(jù)的時(shí)候，同時(shí)變更前置條件中需要判斷的數(shù)據(jù)。這樣，重復(fù)執(zhí)行這個(gè)操作時(shí)，由于第次更新數(shù)據(jù)的時(shí)候已經(jīng)變更了前置條件中需要判斷的數(shù)據(jù)，不滿足前置條件，則不會(huì)重復(fù)執(zhí)行更新數(shù)據(jù)操作。比如，剛剛，“將賬戶X的余額增加100元”這個(gè)操作并不滿足冪等性，可以把這個(gè)操作加上個(gè)前置條件，變?yōu)椋骸叭绻~戶X當(dāng)前的余額

10、為500元，將余額加100元”，這個(gè)操作就具備了冪等性。對(duì)應(yīng)到消息隊(duì)列中的使用時(shí)，可以在發(fā)消息時(shí)在消息體中帶上當(dāng)前的余額，在消費(fèi)的時(shí)候進(jìn)行判斷數(shù)據(jù)庫(kù)中，當(dāng)前余額是否與消息中的余額相等，只有相等才執(zhí)行變更操作。但是，如果要更新的數(shù)據(jù)不是數(shù)值，或者要做個(gè)比較復(fù)雜的更新操作怎么辦？用什么作為前置判斷條件呢？更加通用的方法是，給你的數(shù)據(jù)增加個(gè)版本號(hào)屬性，每次更數(shù)據(jù)前，比較當(dāng)前數(shù)據(jù)的版本號(hào)是否和消息中的版本號(hào)致，如果不致就更新。更新數(shù)據(jù)，更新數(shù)據(jù)的同時(shí)將版本號(hào)+1，樣可以實(shí)現(xiàn)冪等3.并檢查操作如果上面提到的兩種實(shí)現(xiàn)冪等方法都不能適用于你的場(chǎng)景，還有種通用性最強(qiáng)，適用范圍最廣的實(shí)現(xiàn)冪等性方法：并檢查操作，

11、也稱為“Token機(jī)制或者GUID（全局唯ID）機(jī)制”，實(shí)現(xiàn)的思路特別簡(jiǎn)單：在執(zhí)行數(shù)據(jù)更新操作之前，先檢查下是否執(zhí)行過(guò)這個(gè)更新操作。具體的實(shí)現(xiàn)方法是，在發(fā)送消息時(shí)，給每條消息指定個(gè)全局唯的ID，消費(fèi)時(shí)，先根據(jù)這個(gè)ID檢查這條消息是否有被消費(fèi)過(guò)，如果沒有消費(fèi)過(guò)，才更新數(shù)據(jù)，然后將消費(fèi)狀態(tài)置為已消費(fèi)。原理是不是很簡(jiǎn)單？其實(shí)點(diǎn)兒都不簡(jiǎn)單，在分布式系統(tǒng)中，這個(gè)方法其實(shí)是非常難實(shí)現(xiàn)的。首先，給每個(gè)消息指定個(gè)全局唯的ID就是件不那么簡(jiǎn)單的事兒，方法有很多，但都不太好同時(shí)滿足簡(jiǎn)單、高可用和高性能，或多或少都要有些犧牲。更加麻煩的是，在“檢查消費(fèi)狀態(tài)，然后更新數(shù)據(jù)并且設(shè)置消費(fèi)狀態(tài)”中，三個(gè)操作必須作為組操作保

12、證原子性，才能真正實(shí)現(xiàn)冪等，否則就會(huì)出現(xiàn)Bug。比如說(shuō)，對(duì)于同條消息：“全局ID為8，操作為：給ID為666賬戶增加100元”，有可能出現(xiàn)這樣的情況：t0時(shí)刻：Consumer A 收到條消息，檢查消息執(zhí)行狀態(tài)，發(fā)現(xiàn)消息未處理過(guò)，開始執(zhí)行“賬戶增加100元”；t1時(shí)刻：Consumer B 收到條消息，檢查消息執(zhí)行狀態(tài)，發(fā)現(xiàn)消息未處理過(guò)，因?yàn)檫@個(gè)時(shí)刻，ConsumerA還未來(lái)得及更新消息執(zhí)行狀態(tài)。這樣就會(huì)導(dǎo)致賬戶被錯(cuò)誤地增加了兩次100元，這是個(gè)在分布式系統(tǒng)中非常容易犯的錯(cuò)誤，定要引以為戒。對(duì)于這個(gè)問(wèn)題，當(dāng)然可以用事務(wù)來(lái)實(shí)現(xiàn)，也可以用鎖來(lái)實(shí)現(xiàn)，但是在分布式系統(tǒng)中，無(wú)論是分布式事務(wù)還是分布式鎖都是比較難解決問(wèn)題。小結(jié)這節(jié)課主要介紹了通過(guò)冪等消費(fèi)來(lái)解決消息重復(fù)，然后我重點(diǎn)講了幾種實(shí)現(xiàn)冪等操作的方法，你可以利用數(shù)據(jù)庫(kù)的約束來(lái)防止重復(fù)更新數(shù)據(jù)，也可以為數(shù)據(jù)更新設(shè)置次性的前置條件，來(lái)防止重復(fù)消息，如果這兩種方法都不適用于你的場(chǎng)景，還可以用“最廣，但是實(shí)現(xiàn)難度和復(fù)雜度也比較高，般不并檢查操作”的方式來(lái)保證冪等，這種方法適用范圍使用。這些實(shí)現(xiàn)冪等的方法，不僅可以用于解決重復(fù)消息，也同樣適用于，在其他場(chǎng)景中來(lái)解決重復(fù)請(qǐng)求或等的，解決前端