ApsaraMQ for RocketMQ:Consumer types

Funcionamiento

El SDK utiliza internamente un modelo de hilos Reactor:

1. Un hilo integrado de long-polling obtiene mensajes del servidor de manera asíncrona.

2. Los mensajes se colocan en una cola de caché interna.

3. El SDK distribuye los mensajes a los hilos de consumo, los cuales invocan su listener.

Resultados del listener

El message listener debe devolver uno de los siguientes resultados:

Restricciones de confiabilidad

El Push consumer determina el éxito o el fallo estrictamente a partir del valor de retorno del listener. Para preservar esta garantía:

1. Procese de forma síncrona. Complete todo el procesamiento antes de devolver el resultado.

2. No redistribuya mensajes. No transfiera mensajes a otros hilos ni devuelva un resultado antes de que dichos hilos finalicen.

Si el listener devuelve éxito antes de que finalice el procesamiento y este falla posteriormente, el servidor considera el mensaje como consumido y no lo reintenta.

Entrega ordenada de mensajes

Cuando un grupo de consumidores utiliza el modo de consumo ordenado, el Push consumer invoca el listener respetando estrictamente el orden de los mensajes, sin necesidad de configuración adicional. Para más información, consulte Ordered messages.

Cuándo utilizar Push consumers

Los Push consumers son más adecuados cuando:

• El tiempo de procesamiento es predecible. Las duraciones impredecibles provocan tiempos de espera agotados frecuentes, lo que genera mensajes duplicados debido a los reintentos.

• El consumo estándar es suficiente. El SDK controla el modelo de hilos y entrega mensajes con el máximo rendimiento. Esto simplifica el desarrollo, pero no admite procesamiento asíncrono ni control personalizado de la tasa de transferencia.

Clases del SDK

ApsaraMQ for RocketMQ proporciona dos clases del SDK para Push consumers:

• **PushConsumer**: consume mensajes de temas estándar (no Lite).

• **LitePushConsumer**: consume mensajes de temas de tipo Lite, con control de consumo a nivel de granularidad del tema Lite.

// Consume normal messages with PushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind the subscription
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // Register the message listener
        .setMessageListener(new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                // Process the message and return the result
                return ConsumeResult.SUCCESS;
            }
        })
        .build();

// Consume normal messages with LitePushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();

LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Set the topic
        .bindTopic("<your-topic>")
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Register the message listener
        .setMessageListener(messageView -> {
            // Process the message and return the result
            return ConsumeResult.SUCCESS;
        })
        .build();

// Subscribe to Lite topics
litePushConsumer.subscribeLite("<your-lite-topic-1>");
litePushConsumer.subscribeLite("<your-lite-topic-2>");

Funcionamiento

1. Invoque ReceiveMessage para obtener un lote de mensajes del servidor.

2. Distribuya los mensajes a sus hilos de negocio para su procesamiento.

3. Invoque AckMessage por cada mensaje procesado correctamente.

Si el procesamiento falla, no envíe una confirmación. El mensaje vuelve a estar disponible después de que expire la duración de invisibilidad del mensaje, lo que desencadena un reintento. Para más información, consulte SimpleConsumer retry policy.

Operaciones de API

Gestión de fallos

La siguiente tabla describe cómo afectan los distintos escenarios de fallo a la entrega de mensajes:

Entrega ordenada de mensajes

Un Simple consumer procesa ordered messages en el orden de almacenamiento. Para un grupo de mensajes que deben mantener su secuencia, no se puede obtener el siguiente mensaje hasta que se haya procesado el anterior.

Cuándo utilizar Simple consumers

Los Simple consumers son más adecuados cuando:

• El tiempo de procesamiento es impredecible. Especifique una duración inicial de invisibilidad del mensaje al invocar ReceiveMessage y extiéndala con ChangeInvisibleDuration si es necesario.

• Se requieren flujos de trabajo personalizados. El SDK no impone ningún modelo de hilos; implemente distribución asíncrona, consumo por lotes o cualquier patrón personalizado.

• El control de la tasa de transferencia es importante. Su código decide cuándo y con qué frecuencia invocar ReceiveMessage, lo que proporciona control directo sobre el rendimiento.

Clase del SDK

ApsaraMQ for RocketMQ proporciona una clase del SDK para Simple consumers: SimpleConsumer. Esta clase no puede consumir mensajes de temas Lite.

// Consume normal messages with SimpleConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind the subscription
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        .build();

try {
    // Pull up to 10 messages, wait up to 30 seconds
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        // Acknowledge each message after successful processing
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    // Handle failures such as throttling, then retry the receive call
    e.printStackTrace();
}

Control del tiempo de procesamiento para Push consumers

Mantenga el procesamiento de mensajes dentro del umbral de tiempo de espera. Los tiempos de espera agotados frecuentes causan reintentos innecesarios y mensajes duplicados. Si su aplicación maneja habitualmente tareas de larga duración, cambie a un Simple consumer y establezca una duración adecuada de invisibilidad del mensaje.

Migración de LitePullConsumer a SimpleConsumer

LitePullConsumer era un tipo de consumidor en el SDK Remoting de Apache RocketMQ 4.x. Se eliminó en el SDK gRPC 5.x, y los SDK 5.x de Go, Java y Python no incluyen este tipo.

Si anteriormente utilizaba LitePullConsumer para consumo basado en extracción, migre a **SimpleConsumer** después de actualizar al SDK 5.x. SimpleConsumer proporciona las siguientes operaciones:

• ReceiveMessage: obtiene mensajes del servidor bajo demanda.

• AckMessage: confirma el consumo exitoso.

• ChangeInvisibleDuration: modifica la duración de invisibilidad del mensaje para controlar el intervalo de reintentos.

En el SDK de Go, los métodos correspondientes son Receive y Ack. En el SDK de Java, los métodos correspondientes son receive y ack.