ApsaraMQ for RocketMQ:Batch consumption

Como funciona

Um consumidor push gerencia o consumo em lote em duas etapas:

1. Coleta e cache -- As threads de pull recuperam mensagens do ApsaraMQ for RocketMQ por meio de long polling e as armazenam em cache localmente.

2. Despacho -- O consumidor push encaminha o lote para processamento quando as mensagens em cache atingem o limite de tamanho ou de tempo de espera.

Casos de uso

O consumo em lote é mais eficaz quando os sistemas downstream se beneficiam de operações em lote. Se o objetivo for apenas aumentar o paralelismo, considere primeiro alternativas mais simples, como adicionar instâncias de consumidor ou ajustar o tamanho do pool de threads.

• Indexação em massa -- Um sistema de pedidos upstream publica mensagens de log indexadas por um cluster Elasticsearch downstream. Cada mensagem aciona uma solicitação RPC (~10 ms). Processar 10 mensagens individualmente leva 100 ms; agrupá-las em uma única chamada de indexação em massa reduz o total para ~10 ms.

• Inserções em massa no banco de dados -- Uma aplicação insere registros em um banco de dados um por vez com alta frequência de atualização, gerando alta carga. Agrupar 10 registros por inserção e executar o flush a cada 5 segundos reduz a sobrecarga de conexão e a amplificação de escrita.

Limitações

• O consumo em lote é compatível apenas via TCP. Use a edição comercial do SDK do cliente TCP para Java, versão 1.8.7.3.Final ou posterior. Para consultar as notas de versão e as instruções de download, consulte Release notes.

• Tamanho máximo do lote: 1.024 mensagens.

• Tempo máximo de espera entre lotes: 450 segundos.

Parâmetros

Dois parâmetros determinam o momento de envio do lote. O envio é acionado quando qualquer uma das condições é atendida, o que ocorrer primeiro.

Código de exemplo

Configure o consumo em lote por meio das Properties passadas para ONSFactory.createBatchConsumer(). O callback BatchMessageListener recebe uma List<Message> contendo até ConsumeMessageBatchMaxSize mensagens.

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();
        }
    }
}

Práticas recomendadas

Referências

• Commercial edition of TCP client SDK for Java