如何處理消費過程中的重復消息瑞客論壇

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

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

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

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

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

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

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

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

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

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

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

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