Rocketmq重複消費問題

一.背景：

• 背景：

  測試環境Rocketmq的server端有2台伺服器(X.X.X.55和X.X.X.49)，在測試同僚回報推送有問題(重複消費)的前一天早上，發現49這台伺服器上的mq程序沒有了，于是手動重新開機了49的mq服務(也正是這次重新開機，誤導了問題排查的方向)

• 業務場景：

  業務端通過Rocketmq生産和消費消息，其中消費的消息會調用極光接口，推送至app端，正常情況下，每生産一條消息，消費者隻消費一次，app端隻會收到一條推送通知

• 回報問題：

  同一條消息，app端收到多條相同的推送通知

• 猜測原因：

  Rocketmq在消費消息時産生了異常，MessageListenerConcurrently接口傳回了RECONSUME_LATER标志

二.如何在控制台檢視是否有消息積壓或消費失敗

• RocketMQ Console檢視：

  

• 參數：

名稱	作用
#Topic	topic名稱
%RETRY%messageConsumerGroupTest	重試隊列：當消息發送失敗，會将發送失敗的消息發送到topic為%RETRY%groupname 的隊列中
#Broker Offset	topic在該broker上存儲(生産)的消息偏移量
#Consumer Offset	topic在broker上消息的消息偏移量
#Diff	topic在該broker上的消息積壓量，#Diff = (#Broker Offset) - (#Consumer Offset )
Diff Total	該Consumer Group下積壓的消息總數
%DLQ%+consumergroup	死信隊列：消費失敗到一定次數(預設16次)，就會被發送到死信隊列，上圖未出現

• 上圖分析：

1）

以%RETRY%為字首的topic的#Diff的值不為0

，說明對應的topic在消費端出現了

消費失敗後的消息積壓

2）

字首不為%RETRY%的topic的#Diff的值不為0

，說明該topic出現了

消息積壓

三.重複消費排查過程

• 排查步驟1：

  是否有消息積壓

• 排查步驟：

1）打開RocketMQ Console——>Consumer——>ConsumerProgress

2）在 consumer group name輸入框輸入 messageConsumerGroupTest

3）檢視topic消費資訊

• 結果：

當我們拿着測試的手機，手機上在不斷的收到重複推送通知，但檢視控制台，顯示沒有任何的消息積壓

• 分析：

當時很奇怪，收到多條相同的推送，肯定是重複消費了，但控制台卻沒有顯示消息積壓

(原因就詳見本文末尾的問題5解釋)

，隻能将信将疑的去排查Consumer用戶端是否出了問題

• 排查2：

  業務代碼消費日志是有否異常
• 步驟

1）登入業務代碼所在的伺服器，檢視消息消費日志

• 結果

1）同一條推送通知，有被不斷的觸發的日志，但未見Exception和Error相關的可能造成消費失敗的日志

• 分析：

突然又回想起前一天49 伺服器上的mq程序突然沒了，于是就懷疑問題出在Rocketmq的server端

• 排查3：

  Rocketmq server端是否有異常

• 排查步驟：

1）檢視mq的程序是否正常(namesrv和broker)

2）檢視伺服器的記憶體和硬碟空間是否不足

3）檢視/rocketmqlogs目錄下的各個日志是否有異常

4）檢視防火牆是不是打開，或者mq相關服務的端口号無法通路

• 結果：

1） rocketmq的nameserver和broker的程序服務正常

2）服務所在的伺服器記憶體和磁盤空間正常

3）伺服器端各個日志檔案中，也未發現有error日志

4）防火牆是關閉的

mq相關程序：

ps -ef|grep rocketmq

記憶體和磁盤空間，

free -h 、df -h

防火牆狀态，

systemctl status firewalld

server端相關日志，

${user.home}/logs/rocketmqlogs目錄下

• 分析：

在Rocketmq的伺服器端，各種可能情況，檢查了一遍未發現任何異常，隻能一臉懵逼的去Debug單步調式業務代碼，看能不能發現問題

• 排查4：

  Consumer 用戶端是有否異常
• 步驟

1）Idea中啟動相關服務

debug單步調試

2）PostMan模拟使用者請求本地服務，觸發Rocketmq生産者生産消息

3）本地服務的消費者消費消息，調用極光接口發送推送通知，處理相關業務

4）觀察app端推送通知及控制台日志輸出情況

• 結果

1）app端在不斷的接收到重複的推動通知

2）同一條消息，被不斷的觸發消費

3）debug時，在Rocketmq的源碼中捕捉到了異常

Consumer用戶端的核心是實作MessageListenerConcurrently接口，

消費成功傳回CONSUME_SUCCESS

，

消費失敗傳回RECONSUME_LATER

，注意

業務代碼(非源碼)

此處的異常捕捉級别為

Exception

，正是這個異常級别，導緻異常日志沒有被列印出來

1）當debug單步調試到業務代碼的某一處時，代碼突然跳到Rocketmq的源碼的

ConsumeMessageConcurrentlyService類的run()方法中，在status = listener.consumeMessage(Collections.unmodifiableList(this.msgs), context)代碼處捕獲了一個異常

，該方法的傳回值為

消息消費狀态

。

2）如下圖的源碼，可以看到，當發生異常時listener.consumeMessage方法的傳回值為null，

當status為null時，status會被指派為ConsumeConcurrentlyStatus.RECONSUME_LATER狀态(即消費失敗)

3）如下圖，造成業務代碼異常的原因是：

org.mybatis.spring.transaction.SpringManagedTransaction.getTimeout()Ljava/lang/Integer

，這是mybatis抛出的一個異常，正常情況下，這個異常應該在業務代碼處就被捕捉到(業務代碼處确實有捕捉異常方法，隻是捕捉的級别是Exception)，但奇怪的是這個異常并沒有被業務代碼捕捉到，而被Rocketmq的消費監聽方法捕捉到，造成了消費失敗

4）為什麼這個mybatis的異常沒有業務代碼捕捉到，而是被Rocketmq捕捉到，仔細觀察兩者異常捕捉的代碼，發現

業務代碼處的異常捕捉級别是Excption，Rocketmq源碼的異常捕捉級别是Throwable，而Mybatis抛出的正是Error級别的日志

，是以Mybatis抛出的異常沒有被業務代碼捕捉到，

當我們把業務代碼處的異常捕捉級别改為Throwable時，就能在日志中檢視到異常日志資訊

。

當把業務代碼的異常捕捉，改為Throwable級别，就可以捕捉到異常資訊，并在日志中列印出來

• 由上面debug調式帶來的疑問

1）為什麼mybatis的異常沒有業務代碼捕捉到，而是被Rocketmq捕捉到？

原因：業務代碼處的異常捕捉級别是Excption，mq的異常捕捉級别是Throwable，而當mybatis抛出的Error級别的日志，而Error級别的日志不能被Exception捕捉到。

Error類和Exception類的父類都是throwable類，他們的差別是：

1）

Error類一般是指與虛拟機相關的問題

，如系統崩潰，虛拟機錯誤，記憶體空間不足，方法調用棧溢等。對于這類錯誤的導緻的應用程式中斷，僅靠程式本身無法恢複和和預防，遇到這樣的錯誤，建議讓程式終止。

2）

Exception類表示程式可以處理的異常

，可以捕獲且可能恢複。遇到這類異常，應該盡可能處理異常，使程式恢複運作，而不應該随意終止異常。

2）是什麼原因造成的mybatis抛出異常？

原因：

org.mybatis.spring.transaction.SpringManagedTransaction.getTimeout()Ljava/lang/Integer

，出現這個錯誤的原因

主要是spring-mybatis和mybatis版本不比對

，參考部落格連結:https://blog.csdn.net/AirMario/article/details/53446228，本項目中将mybatis的依賴版本由1.4.1改為1.3.1之後，就可以了，版本号改完之後整個mq的消費異常問題也解決了。

3）異常被rocketmq的client端捕捉到了，那client端的mq日志輸出到那去了？

檢視mq源碼中的日志配置檔案logback_rocketmq_client.xml，發現日志被輸出到client服務的

${user.home}/logs/rocketmqlogs

目錄下

登入rocketmq client端伺服器檢視日志

4）mq的消息消費失敗後，會重複消費，重複消費的次數？

在mq的server端啟動時會列印關于消費失敗的重試機制，關鍵字為：

messageDelayLevel

5）為什麼明明是重複消費，但在排查1中RocketMQ的控制台中看到的#Diff都為0？

原因：

1）#Diff表示某個topic的生産和消費的內插補點。在排查1中，雖然我們的消息是消費失敗了，但生産的消息都是被消費了(即使是消費失敗了，但也觸發ConsumerListener消費了，也屬于被消費了)。

2）當我們把mq的Consumer端的程序kill掉，再過一段時間，就會發現%RETRY%messageConsumerGroupTest和其它的topic的#Diff的值不為0，如标題三中的示例圖。

• 最終解決方法：

• 将mq消息消費子產品的pom.xml檔案中mybatis的版本号由1.4.1改為1.3.1，并将相關業務代碼的異常捕捉級别改為Throwable

• 反思：

整個MQ異常問題的處理過程持續了1天半，其中大半的時間浪費在了MQ server端的排查。

從項目第一次引入Rocketmq作為消息中間件到現在也有一年多的時間了，期間雖然一直在寫相關的業務代碼，但僅"知其然，不知其是以然"，一直沒有深入了解整個MQ的架構、原理、特性。

直到這次出現問題，才發現自己對mq的底層實作知之甚少。

是以要關注源碼，關注底層實作，多寫筆記。

最後，借用《黑客與畫家》裡的一句話：如果自己就是潮水的一部分，怎麼能看見潮流的方向呢？