メッセージの蓄積への対処法 - ApsaraMQ for RocketMQ - Alibaba Cloud ドキュメントセンター

現象

Message Queue for RocketMQ インスタンスでのメッセージの蓄積に関するアラートは、潜在的な問題を示しています。Message Queue for RocketMQ コンソールにログインすると、次の現象が見られる場合があります：

グループ詳細 ページで、グループ ID の リアルタイムメッセージ積み上げ数 の値が想定より高くなっています。
ナビゲーションペインでメッセージトレースを選択し、クエリタスクの作成をクリックしてから、Message ID から検索を選択します。必要な情報を入力すると、一部のメッセージはブローカーに送信されますが、下流のコンシューマーには配信されないことがわかります。

考えられる原因

Message Queue for RocketMQ では、メッセージがブローカーに送信された後、Group ID で設定されたクライアントが現在のコンシューマーオフセットに基づいてブローカーからメッセージをプルし、ローカルでメッセージを消費します。通常、クライアントがブローカーからメッセージをプルする際にメッセージの蓄積は発生しません。メッセージの蓄積は通常、長い消費時間や低いコンシューマーの同時実行性といった要因によるクライアントの処理能力不足が原因で発生します。消費メカニズムとメッセージの蓄積の原因に関する詳細については、「メッセージの蓄積とレイテンシーの問題」をご参照ください。

ソリューション

メッセージの蓄積が発生した場合は、次の手順に従って問題を解決してください。

Message Queue for RocketMQ サーバーとクライアントのどちらでメッセージの蓄積が発生しているかを判断します。
ローカルクライアントのログファイル ons.log を確認し、次のメッセージを検索します：
```
the cached message count exceeds the threshold
```
- このログメッセージが表示された場合、クライアントのローカルバッファキューがいっぱいで、クライアントでメッセージが蓄積されています。この場合は、手順 2 に進んでください。
- このログメッセージが見つからない場合、メッセージの蓄積はクライアントで発生していません。この例外的なケースでは、Alibaba Cloud テクニカルサポートにお問い合わせください。
メッセージの消費時間が妥当かどうかを確認します。
- 消費時間が長すぎる場合は、クライアントスタックトレースを確認してビジネスロジックのトラブルシューティングを行ってください。この場合は、手順 3 に進んでください。
- 消費時間が正常な場合、メッセージの蓄積はコンシューマーの同時実行性の不足が原因である可能性があります。この問題を解決するには、コンシューマースレッドの数を徐々に増やすか、コンシューマーノードをスケールアウトしてください。
メッセージの消費時間は、次の方法で確認できます：
- Message Queue for RocketMQ コンソールにログオンして、メッセージトレースを表示します。コンシューマー セクションで、単一メッセージの [消費時間] を表示できます。詳細については、「メッセージトレースのクエリ」をご参照ください。単一メッセージの消費時間を取得するには、メッセージトレースをクエリし、トレース詳細ページのコンシューマー配信結果で [メッセージ処理時間] を表示します。
- Message Queue for RocketMQ コンソールにログオンして、コンシューマーのステータスを表示します。クライアント接続情報で、[レスポンス時間] を確認して平均消費時間を取得します。詳細については、「コンシューマーのステータスを表示」をご参照ください。[Java クライアントのリアルタイムデータ] ポップアップウィンドウの [消費統計] セクションでは、[レスポンス時間 (ms/メッセージ)]、[成功したメッセージ (メッセージ/s)]、[失敗したメッセージ (メッセージ/s)]、[メッセージの蓄積] などの各トピックのメトリックを表示できます。[レスポンス時間] の値が高すぎる (例: 5003.94 ms/message) かつメッセージの蓄積が継続的に増加する場合、消費の遅延に関する問題が発生しています。
- Application Real-Time Monitoring Service (ARMS) などの他の監視製品を使用して、アプリケーションをインストルメント化し、メッセージの消費時間に関するデータを収集できます。
クライアントスタックトレースを表示します。ConsumeMessageThread という名前のスレッドにのみ注目してください。これらのスレッドには、メッセージ消費のビジネスロジックが含まれています。Java の公式ドキュメントを参照して、スレッドの状態を判断し、それに応じてビジネスロジックを修正できます。
クライアントスタックトレースは、次の方法で取得できます：
- Message Queue for RocketMQ コンソールにログオンし、コンシューマーステータスを表示します。クライアント接続情報で、[スタック情報の表示] オプションを探します。詳細については、「コンシューマーステータスの表示」をご参照ください。
- jstack ツールを使用してスタックトレースを出力します。
  1. 「コンシューマーのステータス表示」をご参照の上、メッセージが蓄積しているコンシューマーインスタンスを実行しているホストの IP アドレスを見つけます。その後、ホストにログインします。
  2. 次のいずれかのコマンドを実行して、Java プロセスのプロセス ID (PID) を表示および記録します。
```
ps -ef | grep java
jps -lm
```
  3. 次のコマンドを実行して、スタックトレースを表示します。
```
jstack -l pid > /tmp/pid.jstack
```
  4. 次のコマンドを実行して、ConsumeMessageThread スレッドに関する情報を表示します。
```
cat /tmp/pid.jstack|grep ConsumeMessageThread -A 10 --color
```
次の例は、一般的な異常なスタックトレースを示しています：
- 例 1：蓄積のないアイドル状態のスタックトレース。
  
  コンシューマーがアイドル状態の場合、そのコンシューマースレッドは WAITING 状態にあり、消費タスクキューからのメッセージの取得を待機しています。
  
  たとえば、ConsumeMessageThread_7 スレッドのスタックトレースは TID: 53 STATE: WAITING を示し、コールチェーンは ThreadPoolExecutor.runWorker → ThreadPoolExecutor.getTask → LinkedBlockingQueue.take → LockSupport.park → sun.misc.Unsafe.park です。
- 例 2：消費ロジックにロック競合またはスリープ状態が含まれている。
  コンシューマースレッドが内部のスリープまたは待機操作によってブロックされ、消費が遅くなっています。
```
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)
```
- 例 3：消費ロジックが、データベースなどの外部ストレージシステムとの操作でスタックしている。
  コンシューマースレッドが外部 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)
```
メッセージの蓄積がサービスに影響を与え、メッセージを安全にスキップできる場合は、コンシューマーオフセットのリセット機能を使用してください。この機能を使用すると、蓄積されたメッセージをスキップし、最新のオフセットから消費を再開して、サービスを迅速に復旧できます。詳細については、「コンシューマーオフセットのリセット」をご参照ください。