在眾多關(guān)于MQ的面試八股文中有這么一道題，“如何保證MQ消息消費的冪等性”。
為什么需要保證冪等性呢？是因為消息會重復消費。
為什么消息會重復消費？
明明已經(jīng)消費了，為什么消息會被再次被消費呢？
不同的MQ產(chǎn)生的原因可能不一樣
本文就以RocketMQ為例，來扒一扒RocketMQ中會導致消息重復消息的原因，最終你會發(fā)現(xiàn)，其實消息重復消費算是RocketMQ無奈的“bug”。
消息發(fā)送異常時重復發(fā)送
首先，我們來瞅瞅RocketMQ發(fā)送消息和消費消息的基本原理。
如圖，簡單說一下上圖中的概念：
- Broker，就是RocketMQ的服務(wù)端，如上圖就有兩個服務(wù)實例
- Topic就是一類消息集合的名字
- Queue就是Topic的對應(yīng)的隊列，消息都存在Queue上，每個Topic都會有自己的幾個Queue
所以，整個消息發(fā)送和消費過程大致如下：
- 生產(chǎn)者在發(fā)送消息之前根據(jù)負載均衡策略(默認是輪詢)選擇一個Queue，然后跟這個Queue所在的機器建立連接，把消息發(fā)送到這個Queue上
- 消費者只要消費這個Queue，那么就能消費到消息
在正常情況下，生產(chǎn)者的確是按照這個方式來發(fā)送消息的
但是當出現(xiàn)了異常時，這種異常包括消息發(fā)送超時、響應(yīng)超時等等，RocketMQ為了保證消息成功發(fā)送，會進行消息發(fā)送的重試操作，默認情況下會最多會重試兩次
重試操作比較簡單，就是選擇另一臺機器的Queue來發(fā)送。
雖然重試操作可以很大程度保證消息能夠發(fā)送成功，但是同時也會帶來消息重復發(fā)送的問題。
舉個例子，假設(shè)生產(chǎn)者向A機器發(fā)送消息，發(fā)生了異常，響應(yīng)超時了，但是就一定代表消息沒發(fā)成功么？
不一定，有可能會出現(xiàn)服務(wù)端的確接收到并處理了消息，但是由于網(wǎng)絡(luò)波動等等，導致生產(chǎn)者接收不到服務(wù)端響應(yīng)的情況，此時消息處理成功了，但是生成者還是以為發(fā)生了異常
此時如果發(fā)生重試操作，那么勢必會導致消息被發(fā)送了兩次甚至更多次，導致服務(wù)端存了多條相同的消息，那么就一定會導致消費者重復消費消息。
消費消息拋出異常
在RocketMQ的并發(fā)消費消息的模式下，需要用戶實現(xiàn)MessageListenerConcurrently接口來處理消息
當消費者獲取到消息之后會調(diào)用MessageListenerConcurrently的實現(xiàn)，傳入需要消費的消息集合msgs，這里提到的msgs很重要
如上代碼，當消息消費出現(xiàn)異常的時候，status就會為null，后面就會將status設(shè)置成為RECONSUME_LATER。
RECONSUME_LATER翻譯成功中文就是稍后重新消費的意思
所以從這可以看出，一旦拋出異常，那么消息之后就可以被重復消息。
到這其實可能有小伙伴覺得消息消費失敗重新消費很正常，保證消息盡可能消費成功。
對，這句話不錯，的確可以在一定程度上保證消費異常的消息可以消費成功。
但是坑不在這，而是前面提到的消費時傳入的整個集合中的消息都需要被重新消費。
具體的原因我們接著往下看
當消息處理之后，不論是成功還是異常，都需要對結(jié)果進行處理，代碼如下
當處理結(jié)果為RECONSUME_LATER的時候（異常會設(shè)置為RECONSUME_LATER），此時ackIndex會設(shè)置成-1，后面循環(huán)遍歷的時候就會遍歷到所有這次消費的消息，然后調(diào)用sendMessageBack方法，sendMessageBack方式是用來實現(xiàn)消息重新消費的邏輯，這里就不展開說了。
所以，一旦被消費的一批消息中出現(xiàn)一個消費異常的情況，那么就會導致整批消息被重新消費，從而會導致在出現(xiàn)異常之前的成功處理的消息都會被重復消費，非常坑。
不過好在消費時傳入的消息集合中的消息數(shù)量是可以設(shè)置的，并且默認就是1
也就說默認情況下那個集合中就一條消息，所以默認情況下不會出現(xiàn)消費成功的消息被重復消費的情況。
所以這個參數(shù)不要輕易設(shè)置，一旦設(shè)置大了，就可能導致消息被重新消費。
除了并發(fā)消費消息的模式以外，RocketMQ還支持順序消費消息的模式，也會造成重復消費，邏輯其實差不多，但是在實現(xiàn)消息重新消費的邏輯不一樣。
消費者提交offset失敗
首先來講一講什么是offset。
前面說過，消息在發(fā)送的時候需要指定發(fā)送到，消息最后會被放到Queue中，其實真正的消息不是在Queue中，Queue存的是每個消息的位置，但是你可以理解為Queue存的是消息。
而消息在Queue中是有序號的，這個序號就被稱為offset，從0開始，單調(diào)遞增1。
比如說，如上圖，消息1的offset就是0，消息2的offset就是1，依次類推。
這個offset的一個作用就是用來管理消費者的消費進度。
當消費者在成功消費消息之后，需要將所消費的消息的offset提交給RocketMQ服務(wù)端，告訴RocketMQ，這個Queue的消息我已經(jīng)消費到了這個位置了。
提交offset的代碼就在上述第二節(jié)提到的處理結(jié)果的后面
這樣有一個好處，那么一旦消費者重啟了或者其它啥的要從這個Queue拉取消息的時候，此時他只需要問問RocketMQ服務(wù)端上次這個Queue消息消費到哪個位置了，之后消費者只需要從這個位置開始消費消息就行了，這樣就解決了接著消費的問題。
但是RocketMQ在設(shè)計的時候，當消費完消息的時候并不是同步告訴RocketMQ服務(wù)端offset，而是定時發(fā)送。
如圖，當消費者消費完消息的時候，會將offset保存到內(nèi)存中的一個Map數(shù)據(jù)結(jié)構(gòu)中，所以上面截圖的那段代碼其實是更新內(nèi)存中的offset
而在消費者啟動的時候會開啟一個定時任務(wù)，默認是5s一次，會通過網(wǎng)絡(luò)請求將內(nèi)存中的每個Queue的消費進度offset發(fā)送給RocketMQ服務(wù)端。
由于是定時任務(wù)，所以就可能出現(xiàn)服務(wù)器一旦宕機，導致最新消費的offset沒有成功告訴RocketMQ服務(wù)端的情況
此時，消費進度offset就丟了，那么消費者重啟的時候只能從RocketMQ中獲取到上一次提交的offset，從這里開始消費，而不是最新的offset，出現(xiàn)明明消費到了第8個消息，RocketMQ卻告訴他只消費到了第5個消息的情況，此時必然會導致消息又出現(xiàn)重復消費的情況。
服務(wù)端持久化offset失敗
上一節(jié)說到，消費者會有一個每隔5s鐘的定時任務(wù)將每個隊列的消費進度offset提交到RocketMQ服務(wù)端
當RocketMQ服務(wù)端接收到提交請求之后，會將這個消費進度offset保存到內(nèi)存中
同時為了保證RocketMQ服務(wù)端重啟消費進度不會丟失，也會開啟一個定時任務(wù)，默認也是5s一次，將內(nèi)存中的消費進度持久化到磁盤文件中
所以，整個消費進度offset的數(shù)據(jù)流轉(zhuǎn)過程如下
當RocketMQ服務(wù)端重啟之后，會從磁盤中讀取文件的數(shù)據(jù)加載到內(nèi)存中。
跟消費者產(chǎn)生的問題一樣，一旦RocketMQ發(fā)生宕機，那么offset就有可能丟失5s鐘的數(shù)據(jù)，RocketMQ服務(wù)端一旦重啟，消費者從RocketMQ服務(wù)端獲取到的消息消費進度就比實際消費的進度低，同樣也會導致消息重復消費。
主從同步offset失敗
在RocketMQ的高可用模式中，有一種名叫主從同步的模式，當主節(jié)點掛了之后，從節(jié)點可以手動升級為主節(jié)點對外提供訪問，保證高可用。
在主從同步模式下，從節(jié)點默認每隔10s會向主節(jié)點發(fā)送請求，同步一些元數(shù)據(jù)，這些元數(shù)據(jù)就包括消費進度
當從節(jié)點獲取到主節(jié)點的消費進度之后，會將主節(jié)點的消費進度設(shè)置到自己的內(nèi)存中，同時也會持久化到磁盤。
所以整個消費進度offset的數(shù)據(jù)的流轉(zhuǎn)過程就會變成如下
同樣，由于也是定時任務(wù)，那么一旦主節(jié)點掛了，從節(jié)點就會丟10s鐘的消費進度，此時如果從節(jié)點升級為主節(jié)點對外提供訪問，就會出現(xiàn)跟上面提到的一樣的情況，消費者從這個新的主節(jié)點中拿到的消費進度比實際的低，自然而然就會重復消費消息。
所以，總的來說，在消費進度數(shù)據(jù)流轉(zhuǎn)的過程中，只要某個環(huán)節(jié)出現(xiàn)了問題，都有很有可能會導致消息重復消費。
重平衡
先來講一講什么是重平衡，其實重平衡很好理解，我說一下你就明白了。
前面說到，消費者是從隊列中獲取消息的
在RocketMQ中，有個消費者組的概念，一個消費者組中可以有多個消費者，不同消費者組之間消費消息是互不干擾的，所以前面提到的消費者其實都在消費組下
在同一個消費者組中，消息消費有兩種模式：
- 集群消費模式
- 廣播消費模式
由于RocketMQ默認是集群消費模式，并且絕大多數(shù)業(yè)務(wù)場景都是使用集群消費模式，所以這里就不討論廣播消費模式了，感興趣的同學可以看看RocketMQ消息短暫而又精彩的一生 這篇文章。
集群消費模式是指同一條消息只能被這個消費者組消費一次，這就叫集群消費。
并且前面提到提交消費進度給RocketMQ服務(wù)端的情況只會集群消費模式下才會有，在廣播消費模式不會提給到RocketMQ服務(wù)端，僅僅持久化到本地磁盤
同時前面說的消費者提交消費進度真正提交的是消費者組對于這個Queue的消費進度，而不是指具體的某個消費者對于Queue消費進度。
雖然說這里將前面提到的一些含義更深一步，但是并不妨礙前面的理解。
集群消費的實現(xiàn)就是將隊列按照一定的算法分配給消費者，默認是按照平均分配的。
如圖所示，假設(shè)某個topic有4個Queue，有個消費者組訂閱了這個topic，這個消費者組有兩個消費者1和消費者2，此時每個消費者就可以被分配兩個隊列，這樣就能保證消息正常情況下只會被消費一次。如果只有一個消費者，那么這個消費者就會消費所有隊列，很好理解。
接著后面又啟動了一個消費者3，此時為了保證剛上線的消費者3能夠消費消息，就要進行重平衡操作，重新分配每個消費者消費的隊列。
在重平衡之后就可能會出現(xiàn)下面這種情況
如上圖，原本被消費者2消費的Queue4被分配給消費者3，此時消費者3就能消費到消息了，這就是重平衡。
除了新增消費者會導致重平衡之外，消費者數(shù)量減少，隊列的數(shù)量增加或者減少都會觸發(fā)重平衡。
在了解了重平衡概念之后，接下來分析一下為什么重平衡會導致消息的重復消費。
假設(shè)在進行重平衡時，還未重平衡完之前，消費者2此時還是會按照上面第二節(jié)提到的消費消息的邏輯來消費Queue4的消息
當消費者2已經(jīng)重平衡完成了，發(fā)現(xiàn)Queue4自己已經(jīng)不能消費了，那么此時就會把這個Queue4設(shè)置為dropped，就是丟棄的意思
但是由于重平衡進行時消費者2仍然在消費Queue4的消息，但是當消費完之后，發(fā)現(xiàn)隊列被設(shè)置成dropped，那么此時被消費者2消費消息的offset就不會被提交，原因如下代碼
這段代碼前面已經(jīng)出現(xiàn)過，一旦dropped被設(shè)置成true，這個if條件就通不過，消費進度就不會被提交。
成功消費消息了，但是卻不提交消費進度，這就非常坑了。。
于是當消費者3開始消費Queue4的消息的時候，他就會問問RocketMQ服務(wù)端，我消費者3所在的消費者組對于Queue4這個隊列消費到哪了，我接著消費就行了。
此時由于沒有提交消費進度，RocketMQ服務(wù)端告訴消費者3的消費進度就會比實際的低，這就造成了消息重復消費的情況。
清理長時間消費的消息
在RocketMQ中有這么一個機制，會定時清理長時間正在消費的消息。
如圖，假設(shè)有5條消息現(xiàn)在正在被消費者處理，這5條消息會被存在一個集合中，并且是按照offset的大小排序，消息1的offset最小，消息5的offset最大。
RocketMQ消費者啟動時會開啟一個默認15分鐘執(zhí)行一次的定時任務(wù)
這個定時任務(wù)會去檢查正在處理的消息的第一條消息，也就是圖中的消息1，一旦發(fā)現(xiàn)消息1已經(jīng)處理了超過15分鐘了，那么此時就會將消息1從集合中移除，之后會隔一定時間再次消費消息1。
這也會有坑，雖然消息1從集合中被移除了，但是消息1并沒有消失，仍然被消費者繼續(xù)處理，但是消息1隔一定時間就會再次被消費，就會出現(xiàn)消息1被重復消費的情況。
這就是清理長時間消費的消息導致重復消費的原因。
但此時又會引出一個新的疑問，為什么要移除這個處理超過15分鐘的消息呢？
這就又跟前面提到的消費進度提交有關(guān)！
前面說過消息被消費完成之后會提交消費進度，提交的消費進度實際會有兩種情況：
第一種就是某個線程消費了所有的消息，當把所有的消息都消費完成之后，就會把消息從集合中全部移除，此時提交的消費進度offset就是圖中消息5的offset+1
加1的操作是為了保證如果發(fā)生重啟，那么消費者下次消費的起始位置就是消息5后面的消息，保證消息5不被重復消費
第二種情況就不太一樣了
假設(shè)現(xiàn)在有兩個線程來處理這5條消息，線程1處理前2條，線程2處理后3條，如圖
現(xiàn)在線程1出現(xiàn)了長時間處理消息的情況。
此時線程2處理完消息之后，移除后面三條消息，準備提交offset的時候發(fā)現(xiàn)集合中還有元素，就是線程1正在處理的前兩條消息，此時線程2提交的offset并不是消息5對應(yīng)的offset，而是消息1的offset，代碼如下
這么做的主要原因就是保證消息1和消息2至少被消費一次。
因為一旦提交了消息5對應(yīng)的offset，如果消費者重啟了，下次消費就會接著從消息5的后面開始消費，而對于消息1和消息2來說，并不知道有沒有被消費成功，就有可能出現(xiàn)消息丟失的情況。
所以，一旦集合中最前面的消息長時間處理，那么就會導致后面被消費的消息進度無法提交，那么重啟之后就會導致大量消息被重復消費。
為了解決這個問題，RocketMQ引入了定時清理的機制，定時清理長時間消費的消息，這樣消費進度就可以提交了。
最后
總得來說，RocketMQ中還是存在很多種導致消息重讀消費的情況，并且官方也說了，只是在大多數(shù)情況下消息不會重復
所以如果你的業(yè)務(wù)場景中需要保證消息不能重復消費，那么就需要根據(jù)業(yè)務(wù)場景合理的設(shè)計冪等技術(shù)方案。
原文：https://mp.weixin.qq.com/s/XtIZbObkDcDzcwttSDslZg 作者：三友
如果感覺本文對你有幫助，點贊關(guān)注支持一下
