Consumo por lotes - ApsaraMQ for RocketMQ - Centro de documentación de Alibaba Cloud

El consumo por lotes entrega varios mensajes a un hilo consumidor en un único envío, en lugar de hacerlo uno por uno. Esto reduce la sobrecarga de las llamadas a procedimientos remotos (RPC) hacia los sistemas posteriores y aumenta el rendimiento de los mensajes.

Cómo funciona

Un push consumer gestiona el consumo por lotes en dos etapas:

Extracción y almacenamiento en caché: los hilos de extracción de mensajes obtienen los mensajes de ApsaraMQ for RocketMQ mediante long polling y los almacenan localmente en la caché.
Envío: cuando los mensajes almacenados en la caché alcanzan el umbral de tamaño del lote o el umbral de tiempo de espera (lo que ocurra primero), el push consumer envía el lote a un hilo consumidor para su procesamiento.

batch_consume

Nota

ApsaraMQ for RocketMQ es compatible con push consumers y pull consumers. El consumo por lotes solo aplica a los push consumers. Para obtener más información, consulte

Terms

Casos de uso

El consumo por lotes resulta más eficaz cuando los sistemas posteriores se benefician de las operaciones agrupadas. Si su objetivo es únicamente aumentar el paralelismo, considere primero alternativas más sencillas, como agregar instancias consumidoras o ajustar los tamaños de los pools de hilos.

Indexación masiva: un sistema de pedidos anterior publica mensajes de registro que un clúster de Elasticsearch posterior indexa. Cada mensaje genera una solicitud RPC (~10 ms). Procesar 10 mensajes de forma individual toma 100 ms; agruparlos en una sola llamada de indexación masiva reduce el total a ~10 ms.
Inserciones masivas en bases de datos: una aplicación inserta registros en una base de datos uno por uno con una alta frecuencia de actualización, lo que genera una carga pesada. Agrupar 10 registros por inserción y vaciar cada 5 segundos reduce la sobrecarga de conexión y la amplificación de escritura.

Limitaciones

El consumo por lotes es compatible solo con TCP. Use la edición comercial del SDK de cliente TCP para Java, versión 1.8.7.3.Final o posterior. Para consultar las notas de la versión y las instrucciones de descarga, consulte Release notes.
Tamaño máximo del lote: 1,024 mensajes.
Tiempo máximo de espera entre lotes: 450 segundos.

Parámetros

Dos parámetros controlan cuándo se envía un lote. El envío ocurre cuando se cumple cualquiera de las condiciones, lo que suceda primero.

Parámetro	Tipo	Valor predeterminado	Rango válido	Descripción
`ConsumeMessageBatchMaxSize`	String	32	1--1,024	Número máximo de mensajes por lote. Cuando la cantidad de mensajes en la caché alcanza este valor, el SDK envía el lote a un hilo consumidor de inmediato.
`BatchConsumeMaxAwaitDurationInSeconds`	String	0	0--450	Tiempo máximo de espera en segundos. Al transcurrir este intervalo, el SDK envía los mensajes acumulados hasta ese momento, incluso si no se alcanzó el umbral de tamaño del lote.

Código de ejemplo

Configure el consumo por lotes mediante Properties que se pasan a ONSFactory.createBatchConsumer(). La devolución de llamada BatchMessageListener recibe un objeto List<Message> que contiene hasta ConsumeMessageBatchMaxSize mensajes.

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import java.util.List;
import java.util.Properties;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.tcp.example.MqConfig;

public class SimpleBatchConsumer {

    public static void main(String[] args) {
        Properties consumerProperties = new Properties();
        consumerProperties.setProperty(PropertyKeyConst.GROUP_ID, MqConfig.GROUP_ID);
        consumerProperties.setProperty(PropertyKeyConst.AccessKey, MqConfig.ACCESS_KEY);
        consumerProperties.setProperty(PropertyKeyConst.SecretKey, MqConfig.SECRET_KEY);
        consumerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, MqConfig.NAMESRV_ADDR);

        // Set the maximum number of messages per batch.
        // Default: 32. Valid values: 1 to 1024.
        consumerProperties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(128));
        // Set the maximum wait time between batches, in seconds.
        // Default: 0. Valid values: 0 to 450.
        consumerProperties.setProperty(PropertyKeyConst.BatchConsumeMaxAwaitDurationInSeconds, String.valueOf(10));

        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        batchConsumer.subscribe(MqConfig.TOPIC, MqConfig.TAG, new BatchMessageListener() {

             @Override
            public Action consume(final List<Message> messages, ConsumeContext context) {
                System.out.printf("Batch-size: %d\n", messages.size());
                // Process messages in batches.
                return Action.CommitMessage;
            }
        });
        // Start BatchConsumer.
        batchConsumer.start();
        System.out.println("Consumer start success.") ;

        // Wait for a fixed period to prevent the process from exiting.
        try {
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Nota

Para consultar el código fuente completo, consulte la biblioteca de código en GitHub.
Para consultar la referencia completa de parámetros, consulte Methods and parameters.

Mejores prácticas

Ajuste el tamaño del lote y el tiempo de espera de forma conjunta

El envío se activa cuando se alcanza ya sea el umbral de tamaño del lote o el de tiempo de espera. Establezca ambos parámetros de acuerdo con su carga de trabajo:

Escenarios de alto rendimiento: establezca ConsumeMessageBatchMaxSize en un valor alto (por ejemplo, 128 o 256) y BatchConsumeMaxAwaitDurationInSeconds en un intervalo corto (por ejemplo, de 1 a 5 segundos). De esta manera, los lotes se envían con frecuencia sin esperar a que se llene el lote.
Escenarios de bajo rendimiento: configure un tamaño de lote moderado (por ejemplo, 32) con un tiempo de espera más largo (por ejemplo, de 10 a 30 segundos) para evitar enviar lotes muy pequeños.

Ejemplo: si establece ConsumeMessageBatchMaxSize en 128 y BatchConsumeMaxAwaitDurationInSeconds en 1, el lote se envía después de 1 segundo, incluso si se acumularon menos de 128 mensajes. En este caso, messages.size() en la devolución de llamada retorna un valor menor a 128.

Implemente la idempotencia en el consumo

Para lograr un mejor consumo por lotes, implemente la idempotencia de mensajes en su cliente consumidor. Así se asegura de que cada mensaje se procese solo una vez. Para obtener más información, consulte Consumption idempotence.

Referencias

Commercial edition of TCP client SDK for Java