ApsaraMQ for RabbitMQ:Implement message idempotence

Qu'est-ce que l'idempotence des messages

Une opération idempotente produit le même résultat, qu'elle soit exécutée une ou plusieurs fois. Dans le contexte de la messagerie, l'idempotence signifie que la consommation répétée d'un message aboutit au même résultat que sa consommation unique, sans impact négatif sur le système métier.

Par exemple, un consommateur traite un message de débit de 100 USD. Si une instabilité du réseau provoque une redistribution du message, le consommateur peut le traiter plusieurs fois, mais le débit de 100 USD n'est appliqué qu'une seule fois. La commande affiche un seul enregistrement de débit, sans double facturation. Il s'agit d'un traitement idempotent.

Causes des messages en double

ApsaraMQ for RabbitMQ peut distribuer des messages en double, notamment en cas d'instabilité du réseau. Implémentez l'idempotence si cette duplication affecte votre logique métier. Voici les causes courantes :

• Messages en double lors de l'envoi

  Après la persistance d'un message sur le serveur, une déconnexion temporaire ou une défaillance du client peut empêcher le serveur d'accuser réception auprès du producteur. Ce dernier renvoie alors le message et le consommateur reçoit deux messages identiques, avec le même contenu et le même ID.

• Messages en double lors de la distribution

  Un consommateur traite un message, mais une déconnexion temporaire empêche l'accusé de réception d'atteindre le serveur. Pour garantir une livraison « au moins une fois », le serveur ApsaraMQ for RabbitMQ redistribue le message après le rétablissement du réseau, ce qui entraîne un doublon.

• Messages en double lors du rééquilibrage

  Les redémarrages du serveur ou du client, ainsi que les opérations de mise à l'échelle horizontale (ajout ou suppression de nœuds), déclenchent un rééquilibrage dans ApsaraMQ for RabbitMQ, susceptible de provoquer la distribution de messages en double. Les fluctuations du réseau et les redémarrages d'applications peuvent également déclencher un rééquilibrage.

Mise en œuvre

Utilisez l'ID du message comme clé d'idempotence :

1. Créez une table de base de données dont la clé unique est l'ID du message.

2. Définissez un ID de message unique pour chaque message côté producteur.

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

   Comment définir un ID de message.

3. Dédupliquez les messages par ID côté consommateur.

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