Cara menangani akumulasi pesan - ApsaraMQ for RocketMQ - Alibaba Cloud Documentation Center

Gejala

Peringatan akumulasi pesan pada instans Message Queue for RocketMQ mengindikasikan potensi masalah. Setelah Anda masuk ke Message Queue for RocketMQ console, Anda mungkin melihat gejala berikut:

Pada halaman Group Details, nilai Real-time Accumulated Messages untuk Group ID lebih tinggi dari yang diharapkan.
Pada panel navigasi, pilih Message Tracing, klik Create Query Task, lalu pilih Query by Message ID. Setelah memasukkan informasi yang diperlukan, Anda menemukan bahwa beberapa pesan telah dikirim ke broker tetapi belum dikonsumsi oleh konsumen downstream.

Kemungkinan penyebab

Di Message Queue for RocketMQ, setelah pesan dikirim ke broker, client yang dikonfigurasi dengan Group ID menarik pesan dari broker berdasarkan offset konsumen saat ini dan mengonsumsi pesan tersebut secara lokal. Akumulasi pesan biasanya tidak terjadi selama proses penarikan pesan dari broker. Akumulasi umumnya disebabkan oleh kapasitas pemrosesan client yang tidak mencukupi, seperti waktu konsumsi yang lama atau konkurensi konsumen yang rendah. Untuk informasi lebih lanjut mengenai mekanisme konsumsi dan penyebab akumulasi pesan, lihat Masalah akumulasi dan latensi pesan.

Solusi

Jika terjadi akumulasi pesan, ikuti langkah-langkah berikut untuk menyelesaikan masalah tersebut.

Tentukan apakah akumulasi pesan terjadi di server Message Queue for RocketMQ atau di client.
Periksa file log client lokal ons.log dan cari pesan berikut:
```
the cached message count exceeds the threshold
```
- Jika pesan log ini muncul, antrian buffer lokal client penuh dan pesan terakumulasi di sisi client. Dalam kasus ini, lanjutkan ke Langkah 2.
- Jika pesan log ini tidak ditemukan, akumulasi pesan tidak terjadi di client. Dalam kasus luar biasa ini, hubungi dukungan teknis Alibaba Cloud.
Periksa apakah waktu konsumsi pesan masuk akal.
- Jika waktu konsumsi terlalu lama, periksa jejak stack client untuk mendiagnosis logika bisnis Anda. Dalam kasus ini, lanjutkan ke Langkah 3.
- Jika waktu konsumsi normal, akumulasi pesan kemungkinan disebabkan oleh konkurensi konsumen yang tidak mencukupi. Untuk mengatasi masalah ini, tingkatkan jumlah thread konsumen secara bertahap atau lakukan scale out pada node konsumen Anda.
Anda dapat memeriksa waktu konsumsi pesan dengan cara berikut:
- Masuk ke Message Queue for RocketMQ console untuk melihat jejak pesan. Pada bagian Consumer, Anda dapat melihat consumption time dari satu pesan. Untuk informasi lebih lanjut, lihat Kueri jejak pesan. Untuk mendapatkan waktu konsumsi satu pesan, kueri jejak pesan tersebut dan lihat Message Processing Time pada hasil pengiriman konsumen di halaman detail jejak.
- Masuk ke Message Queue for RocketMQ console untuk melihat status konsumen. Pada informasi koneksi client, periksa Response Time untuk mendapatkan waktu konsumsi rata-rata. Untuk informasi lebih lanjut, lihat Lihat status konsumen. Pada bagian Consumption Statistics di jendela pop-up Java Client Real-time Data, Anda dapat melihat metrik untuk setiap topik, seperti Response Time (ms/message), Successful Messages (messages/s), Failed Messages (messages/s), dan Message Accumulation. Jika nilai Response Time terlalu tinggi (misalnya, 5.003,94 ms/message) dan akumulasi pesan terus meningkat, berarti terdapat masalah latensi konsumsi.
- Gunakan produk pemantauan lain, seperti Application Real-Time Monitoring Service (ARMS), untuk menginstrumentasi aplikasi Anda dan mengumpulkan data mengenai waktu konsumsi pesan.

Lihat jejak stack client. Fokus hanya pada thread yang bernama ConsumeMessageThread. Thread-thread ini berisi logika bisnis untuk konsumsi pesan. Anda dapat merujuk ke dokumentasi resmi Java untuk menentukan status thread dan memodifikasi logika bisnis Anda sesuai kebutuhan.

Anda dapat memperoleh jejak stack client dengan cara berikut:

Masuk ke Message Queue for RocketMQ console dan lihat status konsumen. Pada informasi koneksi client, temukan opsi View Stack Information. Untuk informasi lebih lanjut, lihat Lihat status konsumen.
Gunakan tool jstack untuk mencetak jejak stack.
1. Lihat Lihat status konsumen untuk menemukan alamat IP host yang menjalankan instans konsumen dengan akumulasi pesan. Kemudian, masuk ke host tersebut.
2. Jalankan salah satu perintah berikut untuk melihat dan mencatat process ID (PID) proses Java.
```
ps -ef 
|grep javajps -lm
```
3. Jalankan perintah berikut untuk melihat jejak stack.
```
jstack -l pid > /tmp/pid.jstack
```
4. Jalankan perintah berikut untuk melihat informasi tentang thread ConsumeMessageThread.
```
cat /tmp/pid.jstack|grep ConsumeMessageThread -A 10 --color
```

Berikut ini contoh-contoh jejak stack abnormal yang umum:

Contoh 1: Jejak stack idle tanpa akumulasi.

Ketika konsumen dalam keadaan idle, thread konsumennya berada dalam status WAITING dan menunggu untuk mengambil pesan dari antrian tugas konsumsi.

Misalnya, jejak stack untuk thread ConsumeMessageThread_7 menunjukkan TID: 53 STATE: WAITING, dan rantai panggilannya adalah ThreadPoolExecutor.runWorker → ThreadPoolExecutor.getTask → LinkedBlockingQueue.take → LockSupport.park → sun.misc.Unsafe.park.

Contoh 2: Logika konsumsi melibatkan kontensi lock atau status sleep.

Thread konsumen diblokir oleh operasi sleep atau wait internal, yang memperlambat konsumsi.

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)

Contoh 3: Logika konsumsi tersangkut pada operasi sistem penyimpanan eksternal, seperti database.

Thread konsumen diblokir oleh panggilan HTTP eksternal, yang memperlambat konsumsi.

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)

Jika akumulasi pesan memengaruhi layanan Anda dan pesan tersebut dapat dilewati dengan aman, gunakan fitur Reset Consumer Offsets. Fitur ini memungkinkan Anda melewati pesan yang terakumulasi dan melanjutkan konsumsi dari offset terbaru untuk segera memulihkan layanan Anda. Untuk informasi lebih lanjut, lihat Reset Consumer Offsets.