如何應(yīng)對(duì) RocketMQ 消息堆積
這篇文章，我們聊聊如何應(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ì)列中。客戶端批量拉取消息，常見(jiàn)內(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常見(jiàn)的異常堆棧信息如下：
- 示例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)解決。
參考文檔：
阿里云官方文檔：
