ApsaraMQ for RocketMQ:Manejo de la acumulación de mensajes

Posibles causas

En Message Queue for RocketMQ, cuando los mensajes llegan al broker, un cliente configurado con un Group ID extrae dichos mensajes según el desplazamiento actual del consumidor y los procesa localmente. La acumulación de mensajes no suele ocurrir durante la extracción desde el broker. Por lo general, este fenómeno se debe a una capacidad de procesamiento insuficiente en el cliente, provocada por factores como tiempos de consumo prolongados o baja concurrencia en los consumidores. Para obtener más información sobre el mecanismo de consumo y las causas de la acumulación de mensajes, consulte Message accumulation and latency issues.

Solución

Si se presenta una acumulación de mensajes, siga estos pasos para resolver el problema.

1. Determine si la acumulación de mensajes ocurre en el servidor de Message Queue for RocketMQ o en el cliente.

   Revise el archivo de registro local del cliente ons.log y busque el siguiente mensaje:

   the cached message count exceeds the threshold

   • Si aparece este mensaje en el registro, la cola de búfer local del cliente está llena y los mensajes se están acumulando en el cliente. En este caso, continúe con Step 2.

   • Si no encuentra este mensaje en el registro, la acumulación no se origina en el cliente. Ante esta situación excepcional, póngase en contacto con el soporte técnico de Alibaba Cloud.

2. Compruebe si el tiempo de consumo de mensajes es razonable.

   • Si el tiempo de consumo es excesivo, revise el seguimiento de pila del cliente para depurar su lógica de negocio. En este escenario, continúe con Step 3.

   • Si el tiempo de consumo es normal, la acumulación podría deberse a una concurrencia insuficiente en los consumidores. Para solucionar esto, aumente gradualmente el número de hilos de consumo o escale horizontalmente sus nodos consumidores.

   Puede verificar el tiempo de consumo de mensajes mediante los siguientes métodos:

   • Inicie sesión en la consola de Message Queue for RocketMQ para consultar el rastreo de mensajes. En la sección Consumer, encontrará el consumption time de cada mensaje individual. Para más detalles, consulte Query message traces. Para obtener el tiempo de consumo de un mensaje específico, consulte el rastreo correspondiente y revise el Message Processing Time dentro de los resultados de entrega al consumidor en la página de detalles del rastreo.

   • Inicie sesión en la consola de Message Queue for RocketMQ para revisar el estado del consumidor. En la información de conexión del cliente, verifique el Response Time para conocer el tiempo promedio de consumo. Para más detalles, consulte View consumer status. En la sección Consumption Statistics de la ventana emergente Java Client Real-time Data, podrá consultar métricas por tema, tales como Response Time (ms/message), Successful Messages (messages/s), Failed Messages (messages/s) y Message Accumulation. Si el valor de Response Time es demasiado alto (por ejemplo, 5003.94 ms/message) y la acumulación de mensajes aumenta constantemente, existe un problema de latencia en el consumo.

   • Utilice otros productos de monitoreo, como Application Real-Time Monitoring Service (ARMS), para instrumentar su aplicación y recopilar datos sobre el tiempo de consumo de mensajes.

3. Consulte el seguimiento de pila del cliente. Enfóquese únicamente en los hilos denominados ConsumeMessageThread, ya que contienen la lógica de negocio para el consumo de mensajes. Consulte la documentación oficial de Java para determinar el estado del hilo y ajustar su lógica de negocio en consecuencia.

   Puede obtener el seguimiento de pila del cliente mediante los siguientes métodos:

   • Inicie sesión en la consola de Message Queue for RocketMQ y consulte el estado del consumidor. En la información de conexión del cliente, localice la opción View Stack Information. Para más detalles, consulte View consumer status.

   • Utilice la herramienta jstack para imprimir el seguimiento de pila.

     1. Consulte View consumer status para encontrar la dirección IP del host donde se ejecuta la instancia consumidora que presenta acumulación de mensajes. A continuación, inicie sesión en dicho host.

     2. Ejecute uno de los siguientes comandos para consultar y registrar el identificador de proceso (PID) del proceso Java.

        ps -ef 
        |grep javajps -lm

     3. Ejecute el siguiente comando para consultar el seguimiento de pila.

        jstack -l pid > /tmp/pid.jstack

     4. Ejecute el siguiente comando para consultar información sobre los hilos ConsumeMessageThread.

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

   Los siguientes ejemplos muestran seguimientos de pila anómalos comunes:

   • Ejemplo 1: Seguimiento de pila en estado inactivo sin acumulación.

     Cuando un consumidor está inactivo, sus hilos de consumo se encuentran en estado WAITING y esperan recuperar mensajes de la cola de tareas de consumo.

     Por ejemplo, el seguimiento de pila del hilo ConsumeMessageThread_7 muestra TID: 53 STATE: WAITING, y la cadena de llamadas es ThreadPoolExecutor.runWorker → ThreadPoolExecutor.getTask → LinkedBlockingQueue.take → LockSupport.park → sun.misc.Unsafe.park.

   • Ejemplo 2: La lógica de consumo implica contención de bloqueos o estados de espera.

     El hilo del consumidor se bloquea debido a una operación interna de espera o suspensión, lo que ralentiza el consumo.

     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)

   • Ejemplo 3: La lógica de consumo se detiene en una operación con un sistema de almacenamiento externo, como una base de datos.

     El hilo del consumidor se bloquea debido a una llamada HTTP externa, lo que ralentiza el consumo.

     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. Si la acumulación de mensajes afecta sus servicios y los mensajes pueden omitirse de forma segura, utilice la función Reset Consumer Offsets. Esta función le permite saltar los mensajes acumulados y reanudar el consumo desde el desplazamiento más reciente para restaurar rápidamente sus servicios. Para más información, consulte Reset Consumer Offsets.