ApsaraMQ for RocketMQ:Consumer types

Como funciona

O SDK utiliza internamente um modelo de thread Reactor:

1. Uma thread interna de long-polling obtém mensagens do servidor de forma assíncrona.

2. As mensagens são armazenadas em uma fila de cache interna.

3. O SDK despacha mensagens para as threads de consumo, que invocam seu listener.

Código de exemplo:

// Sample consumption: Use a PushConsumer to consume normal messages.
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 ConsumerGroup")
                // Set the endpoint.
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                // Set the pre-bound subscription relationship.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Set the message listener.
                .setMessageListener(new MessageListener() {
                    @Override
                    public ConsumeResult consume(MessageView messageView) {
                        // Consume the message and return the processing result.
                        return ConsumeResult.SUCCESS;
                    }
                })
                .build();
                

Resultados do listener

O listener de mensagens deve retornar um dos seguintes resultados:

Restrições de confiabilidade

O Push consumer determina sucesso ou falha estritamente pelo valor de retorno do listener. Para preservar essa garantia:

1. Processe sincronamente. Conclua todo o processamento antes de retornar o resultado.

2. Não redistribua mensagens. Não transfira mensagens para outras threads e retorne um resultado antes que essas threads terminem.

Se o listener retornar sucesso antes da conclusão do processamento e este falhar posteriormente, o servidor considerará a mensagem como consumida e não tentará novamente.

Entrega ordenada de mensagens

Quando um grupo de consumidores usa o modo de consumo ordenado, o Push consumer invoca o listener na ordem estrita das mensagens, sem necessidade de configuração adicional. Para mais informações, consulte Mensagens ordenadas.

Quando usar Push consumers

Os Push consumers são mais adequados quando:

• O tempo de processamento é previsível. Durações imprevisíveis acionam timeouts frequentes, causando mensagens duplicadas devido às novas tentativas.

• O consumo padrão é suficiente. O SDK controla o modelo de threads e entrega mensagens com throughput máximo. Isso simplifica o desenvolvimento, mas não suporta processamento assíncrono ou controle de taxa personalizado.

Classes do SDK

O ApsaraMQ for RocketMQ fornece duas classes de SDK para Push consumers:

• **PushConsumer** -- Consome mensagens de tópicos padrão (não Lite).

• **LitePushConsumer** -- Consome mensagens de tópicos do tipo Lite, com controle de consumo na granularidade do tópico 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>");

Como funciona

1. Chame ReceiveMessage para obter um lote de mensagens do servidor.

2. Distribua as mensagens para suas threads de negócios para processamento.

3. Chame AckMessage para cada mensagem processada com sucesso.

Se o processamento falhar, não envie uma confirmação. A mensagem fica disponível novamente após a expiração da duração de invisibilidade, acionando uma nova tentativa. Para mais informações, consulte Política de nova tentativa do SimpleConsumer.

Operações de API

Tratamento de falhas

A tabela a seguir descreve como diferentes cenários de falha afetam a entrega de mensagens:

Entrega ordenada de mensagens

Um Simple consumer processa mensagens ordenadas na ordem de armazenamento. Para um grupo de mensagens que devem permanecer em sequência, não é possível recuperar a próxima mensagem até que a anterior seja processada.

Quando usar Simple consumers

Os Simple consumers são recomendados quando:

• O tempo de processamento é imprevisível. Especifique uma duração inicial de invisibilidade da mensagem ao chamar ReceiveMessage e estenda-a com ChangeInvisibleDuration se necessário.

• Fluxos de trabalho personalizados são necessários. O SDK não impõe nenhum modelo de threads -- implemente distribuição assíncrona, consumo em lote ou qualquer padrão personalizado.

• O controle de taxa é importante. Seu código decide quando e com que frequência chamar ReceiveMessage, fornecendo controle direto sobre o throughput.

Classe do SDK

O ApsaraMQ for RocketMQ fornece uma classe de SDK para Simple consumers: SimpleConsumer. Esta classe não consome mensagens de tópicos 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();
}

Controle o tempo de processamento para Push consumers

Mantenha o processamento de mensagens dentro do limiar de timeout. Timeouts frequentes causam novas tentativas desnecessárias e mensagens duplicadas. Se sua aplicação lida regularmente com tarefas de longa duração, mude para um Simple consumer e defina uma duração apropriada de invisibilidade da mensagem.