ApsaraMQ for RabbitMQ:Implement message idempotence

What is message idempotence

An idempotent operation produces the same result whether executed once or multiple times. In messaging, idempotence means repeated consumption of a message yields the same outcome as consuming it once, with no negative impact on the business system.

For example, a consumer processes a deduction message for 100 USD. If network instability causes redelivery, the consumer may process the message multiple times, but the payment of 100 USD is deducted only once. The order shows a single deduction record with no duplicate charges. This is idempotent processing.

Causes of duplicate messages

ApsaraMQ for RabbitMQ may deliver duplicate messages, particularly during network instability. Implement idempotence if duplication affects your business logic. Common causes:

• Duplicate messages during sending

  After a message is persisted on the server, a transient disconnection or client failure may prevent the server from acknowledging the producer. The producer resends the message, and the consumer receives two messages with the same content and ID.

• Duplicate messages during delivery

  A consumer processes a message but a transient disconnection prevents the acknowledgment from reaching the server. To guarantee at-least-once delivery, the ApsaraMQ for RabbitMQ server redelivers the message after the network recovers, resulting in a duplicate.

• Duplicate messages during rebalancing

  Server or client restarts, scale-outs, or scale-ins trigger rebalancing in ApsaraMQ for RabbitMQ, which can cause duplicate message delivery. Network jitter and application restarts can also trigger rebalancing.

Implementation

Use the message ID as the idempotence key:

1. Create a database table with the message ID as a unique key.

2. Set a unique message ID for each message in the producer.

   Sample code:

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

   How to set a message ID.

3. Deduplicate messages by message ID in the consumer.

   Sample code:

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