ApsaraMQ for RocketMQ:Idempotência de consumo

Nova tentativa do produtor

Um produtor envia uma mensagem e o broker do ApsaraMQ for RocketMQ a persiste. No entanto, o broker pode falhar ao confirmar o recebimento para o produtor devido a um problema transitório de rede ou a uma falha no próprio produtor. O produtor interpreta isso como um envio com falha e tenta novamente. Consequentemente, o consumidor recebe duas mensagens com o mesmo conteúdo, mas IDs de mensagem diferentes.

Reentrega pelo broker

Um consumidor recebe e processa uma mensagem, mas a confirmação de volta para o broker falha devido a um problema transitório de rede. Como o broker não consegue confirmar se a mensagem foi consumida, ele a reentrega após a recuperação da rede para honrar a garantia de entrega pelo menos uma vez. Nesse caso, o consumidor recebe duas mensagens com o mesmo conteúdo e o mesmo ID de mensagem.

Balanceamento de carga

Eventos como instabilidades de rede, reinicializações de brokers ou reinicializações da aplicação consumidora acionam o balanceamento de carga. Durante o rebalanceamento, um consumidor pode receber mensagens já entregues anteriormente.

Etapa 1: Defina a chave da mensagem no produtor

Anexe um identificador único de negócio como chave da mensagem ao enviá-la.

Message message = new Message();
message.setKey("ORDERID_100");
SendResult sendResult = producer.send(message);

Substitua ORDERID_100 pelo identificador único de negócio real, como um ID de pedido ou ID de transação.

Etapa 2: Recupere a chave da mensagem no consumidor

Recupere a chave da mensagem no callback do consumidor e utilize-a para impor o processamento idempotente.

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey()
        // Perform idempotent processing based on the message key that uniquely identifies your business.
    }
});

Etapa 3: Imponha a idempotência com um armazenamento de deduplicação

A chave da mensagem sozinha não impede o processamento duplicado — sua aplicação deve verificar se a chave já foi processada. Um padrão comum utiliza um banco de dados relacional com uma restrição de unicidade:

1. Antes de processar, insira a chave da mensagem em uma tabela de deduplicação com uma restrição de unicidade na coluna da chave.

2. Se a inserção for bem-sucedida, processe a mensagem.

3. Se a inserção falhar devido a uma violação de chave primária ou restrição de unicidade, a mensagem já foi processada. Ignore-a.

Exemplo de tabela de deduplicação (MySQL):

CREATE TABLE message_dedup (
    message_key VARCHAR(255) NOT NULL,
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (message_key)
);

Exemplo de lógica de consumidor idempotente:

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey();

        try {
            // Attempt to insert the message key. Fails if already processed.
            insertMessageKey(key);
        } catch (DuplicateKeyException e) {
            // Already processed. Skip.
            return Action.CommitMessage;
        }

        // Process the business logic.
        processOrder(key);

        return Action.CommitMessage;
    }
});

Substitua insertMessageKey e processOrder pelos seus métodos reais de banco de dados e lógica de negócio.

Para cenários de alto throughput em que um banco de dados relacional pode se tornar um gargalo, utilize uma abordagem baseada em Redis com SETNX (SET if Not eXists).