ApsaraMQ for RabbitMQ:Implement message idempotence

O que é idempotência de mensagens

Uma operação idempotente produz o mesmo resultado independentemente de ser executada uma ou várias vezes. No contexto de mensageria, a idempotência significa que o consumo repetido de uma mensagem gera o mesmo resultado do consumo único, sem impactos negativos no sistema de negócios.

Por exemplo, um consumidor processa uma mensagem de débito no valor de 100 USD. Se uma instabilidade na rede provocar reentrega, o consumidor poderá processar a mensagem múltiplas vezes, mas o débito de 100 USD ocorrerá apenas uma vez. O pedido exibirá um único registro de débito, sem cobranças duplicadas. Esse cenário caracteriza o processamento idempotente.

Causas de mensagens duplicadas

O ApsaraMQ for RabbitMQ pode entregar mensagens duplicadas, especialmente durante instabilidades de rede. Implemente idempotência caso a duplicação afete sua lógica de negócios. As causas mais comuns incluem:

• Duplicação durante o envio

  Após a persistência da mensagem no servidor, uma desconexão transitória ou falha no cliente pode impedir o envio do reconhecimento ao produtor. Consequentemente, o produtor reenvia a mensagem e o consumidor recebe duas mensagens com o mesmo conteúdo e ID.

• Duplicação durante a entrega

  O consumidor processa a mensagem, mas uma desconexão temporária impede que o reconhecimento chegue ao servidor. Para garantir a entrega pelo menos uma vez (at-least-once), o servidor do ApsaraMQ for RabbitMQ reentrega a mensagem após a recuperação da rede, resultando em duplicidade.

• Duplicação durante o rebalanceamento

  Reinicializações de servidor ou cliente, expansões ou reduções de escala acionam o rebalanceamento no ApsaraMQ for RabbitMQ, o que pode causar entrega duplicada de mensagens. Oscilações de rede e reinícios de aplicações também podem desencadear esse processo.

Implementação

Use o ID da mensagem como chave de idempotência:

1. Crie uma tabela no banco de dados com o ID da mensagem como chave única.

2. Defina um ID exclusivo para cada mensagem no produtor.

   Exemplo de código:

   AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(UUID.randomUUID().toString()).build();
   channel.basicPublish("${ExchangeName}", "RoutingKey", true, props, ("Message Body" + i).getBytes(StandardCharsets.UTF_8));

   Como definir um ID de mensagem.

3. Deduplique as mensagens por ID no consumidor.

   Exemplo de código:

   channel.basicConsume(Producer.QueueName, false, new DefaultConsumer(channel) {
       @Override public void handleDelivery(String consumerTag, Envelope envelope,
                   AMQP.BasicProperties properties, byte[] body) throws IOException {
           // 1. Obtain the data for the unique business index.
           try{
               String messageId = properties.getMessageId();
               // The message ID or other information that can be used as a unique key.
               // 2. Start a database transaction.
               idempTable.insert(messageId);
               // 3. Process the business logic for the received message.
               // 4. Commit or roll back the transaction. Send an acknowledgement (ACK) only if the processing is successful.
               channel.basicAck(envelope.getDeliveryTag(), false);
           }
           catch (DatabasePrimaryKeyConflictException e){
               // This is a duplicate message. Directly acknowledge it.
               channel.basicAck(envelope.getDeliveryTag(), false);
           }
       }
   }
   );