全部產品
Search

搜尋本產品

    ApsaraMQ for RocketMQ：如何處理訊息堆積

    文件中心

    ApsaraMQ for RocketMQ：如何處理訊息堆積

    更新時間：Jun 05, 2026

    問題現象

    在使用雲訊息佇列 RocketMQ 版執行個體時收到訊息堆積警示，登入雲訊息佇列 RocketMQ 版控制台後發現了下列現象：

    • 在Group 详情頁面，看到Group ID的实时消息堆积量的值高於預期。

    • 導覽列中選擇訊息軌跡，單擊创建查询任务，選擇按 Message ID 查询，輸入對應的資訊，發現部分訊息已發送至Broker節點，但未投遞給下遊消費者。

    可能原因

    雲訊息佇列 RocketMQ 版的訊息發送至Broker節點後，配置了Group ID的用戶端根據當前的消費位點，從Broker節點拉取部分訊息到本地進行消費。一般情況下，用戶端從Broker節點拉取訊息的過程不會導致訊息堆積，主要是用戶端本地消費過程中，由於消費耗時過長或消費並發度較小等原因，導致用戶端消費能力不足，出現訊息堆積的問題。具體的消費原理和訊息堆積原因請參見訊息堆積和延遲問題。

    解決方案

    若出現訊息堆積，可參考以下措施進行定位和處理。

    1. 判斷訊息堆積在雲訊息佇列 RocketMQ 版服務端還是用戶端。

      查看用戶端本地記錄檔ons.log，搜尋是否出現如下資訊：

      the cached message count exceeds the threshold

      • 出現相關日誌資訊，說明用戶端本地緩衝隊列已滿，訊息堆積在用戶端，請執行步驟2。

      • 若未出現相關日誌，說明訊息堆積不在用戶端，若出現這種特殊情況，請直接聯絡阿里雲支援人員。

    2. 確認訊息的消費耗時是否合理。

      • 若查看到消費耗時較長，則需要查看用戶端堆棧資訊排查具體商務邏輯，請執行步驟3。

      • 若查看到消費耗時正常，則有可能是因為消費並發度不夠導致訊息堆積，需要逐步調大消費線程或擴容節點來解決。

      訊息的消費耗時可以通過以下方式查看：

      • 登入雲訊息佇列 RocketMQ 版控制台查看訊息的消費軌跡，在消费者地區中可以看到單條訊息的消費耗時。具體操作，請參見查詢訊息軌跡。登入雲訊息佇列 RocketMQ 版控制台，查詢訊息軌跡，在軌跡詳情頁面的消費者投遞結果中查看訊息處理耗時，擷取單條訊息的消費耗時。

      • 登入雲訊息佇列 RocketMQ 版控制台查看消費者狀態，在用戶端串連資訊中查看業務處理時間，擷取消費耗時的平均值。具體操作，請參見查看消費者狀態。在Java 用戶端即時資料彈窗的消費統計地區，可查看各 Topic 的業務處理時間（毫秒/條）、訊息成功（條/秒）、訊息失敗（條/秒）及訊息堆積量等指標。若業務處理時間值過大（例如 5003.94 毫秒/條），且訊息堆積量持續增長，則說明存在消費延遲問題。

      • 使用阿里雲ARMS等其他監控產品做業務埋點採集訊息的消費耗時。

    3. 查看用戶端堆棧資訊。只需要關注線程名為ConsumeMessageThread的線程，這些都是業務消費訊息的邏輯。可參見Java官方文檔判斷線程的狀態並根據具體問題修改商務邏輯。

      用戶端堆棧資訊可以通過以下方式擷取：

      • 登入雲訊息佇列 RocketMQ 版控制台查看消費者狀態，在用戶端串連資訊中查看Java用戶端堆棧資訊。具體操作，請參見查看消費者狀態。

      • 使用Jstack工具列印堆棧資訊。

        1. 請參見查看消費者狀態擷取訊息堆積的消費者執行個體所對應的宿主機IP地址，並登入該宿主機。

        2. 執行以下任意命令，查看並記錄Java進程的PID。

          ps -ef 
|grep javajps -lm

        3. 執行以下命令，查看堆棧資訊。

          jstack -l pid > /tmp/pid.jstack

        4. 執行以下命令，查看ConsumeMessageThread的資訊。

          cat /tmp/pid.jstack|grep ConsumeMessageThread -A 10 --color

      常見的異常堆棧資訊如下：

      • 樣本一：空閑無堆積的堆棧。

        消費空閑情況下消費線程都會處於WAITING狀態等待從消費任務隊中擷取訊息。

        例如線程 ConsumeMessageThread_7 的堆棧顯示 TID: 53 STATE: WAITING，調用鏈為 ThreadPoolExecutor.runWorker → ThreadPoolExecutor.getTask → LinkedBlockingQueue.take → LockSupport.park → sun.misc.Unsafe.park。

      • 樣本二：消費邏輯有搶鎖休眠等待等情況。

        消費線程阻塞在內部的一個睡眠等待上，導致消費緩慢。

        Thread: ConsumeMessageThread_16
Stack:
TID: 51 STATE: TIMED_WAITING
java.lang.Thread.sleep(Native Method)
mqtest.DelayTest$1.consume(DelayTest.java:51)
com.aliyun.openservices.ons.api.impl.rocketmq.ConsumerImpl$MessageListenerImpl.consumeMessage(ConsumerImpl.java:101)
com.aliyun.openservices.shade.com.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:415)
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
java.util.concurrent.FutureTask.run(FutureTask.java:266)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)

      • 樣本三：消費邏輯操作資料庫等外部儲存卡住。

        消費線程阻塞在外部的HTTP調用上，導致消費緩慢。

        ConsumeMessageThread_3
TID: 54 STATE: RUNNABLE
java.lang.ClassLoader.loadClass(ClassLoader.java:404)
sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
java.lang.ClassLoader.loadClass(ClassLoader.java:357)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:174)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:158)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:149)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:125)
refactor.base.Tools.getHttpsClient(Tools.java:138)
refactor.base.Tools.httpsPost(Tools.java:257)
mqtest.DelayTest$1.consume(DelayTest.java:58)
com.aliyun.openservices.ons.api.impl.rocketmq.ConsumerImpl$MessageListenerImpl.consumeMessage(ConsumerImpl.java:101)
com.aliyun.openservices.shade.com.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:415)
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
java.util.concurrent.FutureTask.run(FutureTask.java:266)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)

    4. 針對某些特殊業務情境，如果訊息堆積已經影響到業務運行，且堆積的訊息本身可以跳過不消費，您可以通過重設消費位點跳過這些堆積的訊息從最新位點開始消費，快速恢複業務。具體操作，請參見重設消費位點。