一、先搞懂：消息堆積的核心原因
消息堆積本質是「生產速度 > 消費速度」，常見誘因：
- 消費端：消費線程數不足、業務邏輯耗時久、消費端故障 / 重啓、消費異常重試頻繁；
- 生產端：突發流量（如秒殺）導致消息量暴增；
- 集羣端：Broker 性能瓶頸（磁盤 IO / 網絡帶寬不足）、隊列數配置不合理、消息堆積閾值未監控。
二、應急處理：快速緩解堆積（先止損）
1. 臨時擴容消費端（最快見效）
- 增加消費實例數量：多部署幾台消費服務，分攤消費壓力（需確保消費組配置正確，避免重複消費）；
- 提升單實例消費線程數：調整
consumeThreadMin/consumeThreadMax（默認 10/20），根據 CPU / 內存情況調至 50-200（避免線程過多導致上下文切換）；
java
運行
// 示例：調整消費線程數
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setConsumeThreadMin(50);
consumer.setConsumeThreadMax(200);
- 臨時關閉非核心消費邏輯：註釋掉消費端中耗時的 DB 寫入、第三方接口調用，先把消息消費掉再補處理（適合緊急場景）。
2. 優化消費端業務邏輯（降耗時）
- 同步改異步：把消費後的業務操作（如入庫、通知）改成異步線程池處理，消費端只做 “接收消息 + 落盤”，核心是讓
consumeMessage 方法快速返回；
- 批量消費：開啓批量消費功能，單次拉取多條消息（如 32/64 條）批量處理，減少網絡交互和業務調用次數；
java
運行
// 開啓批量消費，設置單次拉取最大條數
consumer.setConsumeMessageBatchMaxSize(32);
- 優化慢查詢 / 慢接口：排查消費端 DB 慢 SQL、第三方接口超時問題，加緩存、優化索引，把單條消息處理耗時從秒級壓到毫秒級。
3. 臨時調整 Broker 參數（兜底）
- 提升 Broker 拉取限速：默認 Broker 對消費端拉取速度有閾值，臨時調大
pullThresholdForQueue（隊列級拉取閾值，默認 1000），允許消費端一次性拉更多消息；
- 擴容 Broker 節點：若 Broker 磁盤 IO / 網絡瓶頸，臨時增加 Broker 節點，拆分隊列到新節點，分攤存儲和轉發壓力。
三、長期優化：從根源避免堆積（治本）
1. 架構層面：削峯填谷 + 分流
- 接入限流 / 削峯：生產端前置 Redis / 消息隊列做流量控制，避免突發流量直接打滿 RocketMQ（如秒殺場景用令牌桶限流）；
- 消息分級處理：核心消息（如訂單支付）和非核心消息（如日誌、通知）分不同 Topic / 隊列，核心隊列配置更多消費資源，非核心隊列可降級；
- 死信隊列兜底：配置死信隊列（DLQ），把消費失敗多次的消息轉移到死信隊列，避免無效重試佔用消費資源。
2. 配置層面：合理規劃隊列與資源
- 增加 Topic 隊列數：RocketMQ 的消費並行度上限等於隊列數，若 Topic 隊列數過少（如默認 4 個），再多消費線程也沒用，按業務峯值規劃隊列數（如秒殺場景配 32/64 個隊列）；
- 消費端資源預留：按業務峯值的 1.5-2 倍配置消費端機器數、線程數，避免資源不足；
- 開啓消息回溯與監控：配置消息堆積監控告警（如隊列堆積數 > 10000 觸發告警），支持消息回溯重放，堆積時可定向重放未消費消息。
3. 監控與預案（早發現）
- 核心監控指標：監控「隊列堆積數」「消費延遲時間」「單條消息處理耗時」「消費失敗率」，設置閾值告警（如堆積 > 5000、延遲 > 5 分鐘告警）；
- 制定應急預案：提前梳理 “秒殺 / 大促” 等高峯場景的堆積預案，包括臨時擴容腳本、批量消費開關、非核心業務降級開關，做到一鍵執行。
四、避坑提醒
❌ 只擴容消費線程不優化業務：線程數過多會導致 CPU 上下文切換頻繁，反而降低消費效率；❌ 批量消費條數設置過大：單次拉取太多消息會導致 OOM，建議按消息大小調整（小消息 32/64 條，大消息 8/16 條）；✅ 核心原則：先通過 “擴容消費端 + 批量消費” 快速止損，再通過 “優化業務 + 調整配置” 長期解決，最後靠 “監控 + 預案” 提前預防。
