大家好，我是小富～

這篇文章，我們聊聊如何應(yīng)對(duì) RocketMQ 消息堆積。

1 基礎(chǔ)概念

消費(fèi)者在消費(fèi)的過(guò)程中，消費(fèi)的速度跟不上服務(wù)端的發(fā)送速度，未處理的消息會(huì)越來(lái)越多，消息出現(xiàn)堆積進(jìn)而會(huì)造成消息消費(fèi)延遲。

雖然筆者經(jīng)常講：RocketMQ 、Kafka 具備堆積的能力，但是以下場(chǎng)景需要重點(diǎn)關(guān)注消息堆積和延遲的問(wèn)題：

業(yè)務(wù)系統(tǒng)上下游能力不匹配造成的持續(xù)堆積，且無(wú)法自行恢復(fù)。
業(yè)務(wù)系統(tǒng)對(duì)消息的消費(fèi)實(shí)時(shí)性要求較高，即使是短暫的堆積造成的消息延遲也無(wú)法接受。

2 消費(fèi)原理

客戶端使用 Push 模式 啟動(dòng)后，消費(fèi)消息時(shí)，分為以下兩個(gè)階段：

階段一：拉取消息

客戶端通過(guò)長(zhǎng)輪詢批量拉取的方式從 Broker 服務(wù)端獲取消息，將拉取到的消息緩存到本地緩沖隊(duì)列中。

客戶端批量拉取消息，常見內(nèi)網(wǎng)環(huán)境下都會(huì)有很高的吞吐量，例如：1個(gè)單線程單分區(qū)的低規(guī)格機(jī)器（4C8GB）可以達(dá)到幾萬(wàn) TPS ，如果是多個(gè)分區(qū)可以達(dá)到幾十萬(wàn) TPS 。所以這一階段一般不會(huì)成為消息堆積的瓶頸。
階段二：消費(fèi)消息

提交消費(fèi)線程，客戶端將本地緩存的消息提交到消費(fèi)線程中，使用業(yè)務(wù)消費(fèi)邏輯進(jìn)行處理。

此時(shí)客戶端的消費(fèi)能力就完全依賴于業(yè)務(wù)邏輯的復(fù)雜度（消費(fèi)耗時(shí)）和消費(fèi)邏輯并發(fā)度了。如果業(yè)務(wù)處理邏輯復(fù)雜，處理單條消息耗時(shí)都較長(zhǎng)，則整體的消息吞吐量肯定不會(huì)高，此時(shí)就會(huì)導(dǎo)致客戶端本地緩沖隊(duì)列達(dá)到上限，停止從服務(wù)端拉取消息。

通過(guò)以上客戶端消費(fèi)原理可以看出，消息堆積的主要瓶頸在于本地客戶端的消費(fèi)能力，即消費(fèi)耗時(shí)和消費(fèi)并發(fā)度。

想要避免和解決消息堆積問(wèn)題，必須合理的控制消費(fèi)耗時(shí)和消息并發(fā)度，其中消費(fèi)耗時(shí)的優(yōu)先級(jí)高于消費(fèi)并發(fā)度，必須先保證消費(fèi)耗時(shí)的合理性，再考慮消費(fèi)并發(fā)度問(wèn)題。

3 消費(fèi)瓶頸

3.1 消費(fèi)耗時(shí)

影響消費(fèi)耗時(shí)的消費(fèi)邏輯主要分為 CPU 內(nèi)存計(jì)算和外部 I/O 操作，通常情況下代碼中如果沒(méi)有復(fù)雜的遞歸和循環(huán)的話，內(nèi)部計(jì)算耗時(shí)相對(duì)外部 I/O 操作來(lái)說(shuō)幾乎可以忽略。

外部 I/O 操作通常包括如下業(yè)務(wù)邏輯：

讀寫外部數(shù)據(jù)庫(kù)，例如 MySQL 數(shù)據(jù)庫(kù)讀寫。
讀寫外部緩存等系統(tǒng)，例如 Redis 讀寫。
下游系統(tǒng)調(diào)用，例如 Dubbo 調(diào)用或者下游 HTTP 接口調(diào)用。

這類外部調(diào)用的邏輯和系統(tǒng)容量需要提前梳理，掌握每個(gè)調(diào)用操作預(yù)期的耗時(shí)，這樣才能判斷消費(fèi)邏輯中I/O操作的耗時(shí)是否合理。

通常消費(fèi)堆積都是由于這些下游系統(tǒng)出現(xiàn)了服務(wù)異常、容量限制導(dǎo)致的消費(fèi)耗時(shí)增加。

例如：某業(yè)務(wù)消費(fèi)邏輯中需要調(diào)用下游 Dubbo 接口，單次消費(fèi)耗時(shí)為 20 ms，平時(shí)消息量小未出現(xiàn)異常。業(yè)務(wù)側(cè)進(jìn)行大促活動(dòng)時(shí)，下游 Dubbo 服務(wù)未進(jìn)行優(yōu)化，消費(fèi)單條消息的耗時(shí)增加到 200 ms，業(yè)務(wù)側(cè)可以明顯感受到消費(fèi)速度大幅下跌。此時(shí)，通過(guò)提升消費(fèi)并行度并不能解決問(wèn)題，需要大幅提高下游 Dubbo 服務(wù)性能才行。

3.2 消費(fèi)并發(fā)度

絕大部分消息消費(fèi)行為都屬于 IO 密集型，即可能是操作數(shù)據(jù)庫(kù)，或者調(diào)用 RPC，這類消費(fèi)行為的消費(fèi)速度在于后端數(shù)據(jù)庫(kù)或者外系統(tǒng)的吞吐量，通過(guò)增加消費(fèi)并行度，可以提高總的消費(fèi)吞吐量，但是并行度增加到一定程度，反而會(huì)下降。

所以，應(yīng)用必須要設(shè)置合理的并行度。如下有幾種修改消費(fèi)并行度的方法：

同一個(gè) ConsumerGroup 下，通過(guò)增加 Consumer 實(shí)例數(shù)量來(lái)提高并行度（需要注意的是超過(guò)訂閱隊(duì)列數(shù)的 Consumer 實(shí)例無(wú)效）?？梢酝ㄟ^(guò)加機(jī)器，或者在已有機(jī)器啟動(dòng)多個(gè)進(jìn)程的方式。
提高單個(gè) Consumer 實(shí)例的消費(fèi)并行線程，通過(guò)修改參數(shù) consumeThreadMin、consumeThreadMax 實(shí)現(xiàn)。

4 解決策略

當(dāng)面對(duì)消息堆積問(wèn)題時(shí)，我們需要明確到底哪個(gè)環(huán)節(jié)出現(xiàn)問(wèn)題了，不要慌張，也不要貿(mào)然動(dòng)手。

4.1 確認(rèn)消息的消費(fèi)耗時(shí)是否合理

首先，我們需要查看消費(fèi)耗時(shí)，確認(rèn)消息的消費(fèi)耗時(shí)是否合理。查看消費(fèi)耗時(shí)一般來(lái)講有兩種方式：

1、打印日志

public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
     try {
        for (MessageExt messageExt : msgs) {
           long start = System.currentTimeMillis();
           // TODO 業(yè)務(wù)邏輯
          logger.info("MessageId:" + messageExt.getMsgId() + " costTime:" + (System.currentTimeMillis() - start));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
     } catch (Exception e) {
        logger.error("consumeMessage error:", e);
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
     }
}

2、查看消息軌跡

當(dāng)確定好消費(fèi)耗時(shí)后，可以根據(jù)耗時(shí)大小，采取不同的措施。

若查看到消費(fèi)耗時(shí)較長(zhǎng)，則需要查看客戶端 JVM 堆棧信息排查具體業(yè)務(wù)邏輯，并優(yōu)化消費(fèi)邏輯。
若查看到消費(fèi)耗時(shí)正常，則有可能是因?yàn)橄M(fèi)并發(fā)度不夠?qū)е孪⒍逊e，需要逐步調(diào)大消費(fèi)線程或擴(kuò)容節(jié)點(diǎn)來(lái)解決。

4.2 查看客戶端 JVM 的堆棧

假如消費(fèi)耗時(shí)非常高，需要查看 Consumer 實(shí)例 JVM 的堆棧。

通過(guò) jps -m 或者 ps -ef | grep java 命令獲取當(dāng)前正在運(yùn)行的 Java 程序，通過(guò)啟動(dòng)主類即可獲得應(yīng)用的進(jìn)程 pid ;
通過(guò) jstack pid > stack.log 命令獲取線程的堆棧。
執(zhí)行以下命令，查看 ConsumeMessageThread 的信息。

cat stack.log | grep ConsumeMessageThread -A 10 --color

常見的異常堆棧信息如下：

示例1：空閑無(wú)堆積的堆棧 。

消費(fèi)空閑情況下消費(fèi)線程都會(huì)處于 WAITING 狀態(tài)等待從消費(fèi)任務(wù)隊(duì)里中獲取消息。

示例2：消費(fèi)邏輯有搶鎖休眠等待等情況 。

消費(fèi)線程阻塞在內(nèi)部的一個(gè)睡眠等待上，導(dǎo)致消費(fèi)緩慢。

示例3：消費(fèi)邏輯操作數(shù)據(jù)庫(kù)等外部存儲(chǔ)卡住 。

消費(fèi)線程阻塞在外部的 HTTP 調(diào)用上，導(dǎo)致消費(fèi)緩慢。

5 總結(jié)

客戶端使用 Push模式 啟動(dòng)后，消費(fèi)消息時(shí)，分為以下兩個(gè)階段：拉取消息和消費(fèi)消息。

客戶端消費(fèi)原理可以看出，消息堆積的主要瓶頸在于本地客戶端的消費(fèi)能力，即消費(fèi)耗時(shí)和消費(fèi)并發(fā)度。

首先分析消費(fèi)耗時(shí)，然后根據(jù)耗時(shí)大小，采取不同的措施。

若查看到消費(fèi)耗時(shí)較長(zhǎng)，則查看客戶端堆棧信息排查具體業(yè)務(wù)邏輯，并優(yōu)化消費(fèi)邏輯。
若查看到消費(fèi)耗時(shí)正常，則有可能是因?yàn)橄M(fèi)并發(fā)度不夠?qū)е孪⒍逊e，需要逐步調(diào)大消費(fèi)線程或擴(kuò)容節(jié)點(diǎn)來(lái)解決。

參考文檔：

阿里云官方文檔：

https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/use-cases/message-accumulation-and-latency#concept-2004064

如果我的文章對(duì)你有所幫助，還請(qǐng)幫忙點(diǎn)贊、在看、轉(zhuǎn)發(fā)一下，你的支持會(huì)激勵(lì)我輸出更高質(zhì)量的文章，非常感謝！

我是小富～下期見！

·········· END ··············

        
         在看、點(diǎn)贊、轉(zhuǎn)發(fā)，是對(duì)我最大的鼓勵(lì)。
        
        
         

        
        
         《ShardingSphere5.x分庫(kù)分表原理與實(shí)戰(zhàn)》PDF公眾號(hào)內(nèi)回復(fù)[ 分庫(kù)分表 ] Get。
        

技術(shù)書籍公眾號(hào)內(nèi)回復(fù)[ pdf ] Get。

面試筆記、springcloud進(jìn)階實(shí)戰(zhàn)PDF，公眾號(hào)內(nèi)回復(fù)[ 1222 ] Get。

應(yīng)對(duì) RocketMQ 百萬(wàn)消息堆積，我出手了