ApsaraMQ for RocketMQ:Consumption idempotence

Nouvelle tentative du producteur

Le producteur envoie un message et le courtier ApsaraMQ for RocketMQ le persiste. Toutefois, le courtier peut ne pas parvenir à envoyer l'accusé de réception au producteur en raison d'un problème réseau temporaire ou d'une panne de ce dernier. Le producteur considère alors l'envoi comme échoué et réessaie. Le consommateur reçoit deux messages contenant le même contenu mais des ID de message différents.

Redistribution par le courtier

Le consommateur reçoit et traite un message, mais l'accusé de réception envoyé au courtier échoue en raison d'un problème réseau temporaire. Ne pouvant confirmer la consommation, le courtier redistribue le message après le rétablissement du réseau afin de respecter la garantie de livraison « au moins une fois ». Le consommateur reçoit alors deux messages avec le même contenu et le même ID de message.

Équilibrage de charge

Des instabilités réseau, des redémarrages de courtiers ou d'applications consommatrices déclenchent un équilibrage de charge. Durant le rééquilibrage, un consommateur peut recevoir des messages déjà livrés.

Étape 1 : Définissez la clé de message côté producteur

Joignez un identifiant métier unique comme clé de message lors de l'envoi.

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

Remplacez ORDERID_100 par l'identifiant métier unique réel, tel qu'un ID de commande ou de transaction.

Étape 2 : Récupérez la clé de message côté consommateur

Récupérez la clé de message dans le rappel du consommateur et utilisez-la pour garantir l'idempotence du traitement.

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

Étape 3 : Garantissez l'idempotence via un magasin de déduplication

La clé de message seule n'empêche pas le traitement en double : votre application doit vérifier si la clé a déjà été traitée. Un modèle courant utilise une base de données relationnelle avec une contrainte d'unicité :

1. Avant le traitement, insérez la clé de message dans une table de déduplication dotée d'une contrainte d'unicité sur la colonne de clé.

2. Si l'insertion réussit, traitez le message.

3. Si l'insertion échoue en raison d'une violation de clé primaire ou de contrainte d'unicité, le message a déjà été traité. Ignorez-le.

Exemple de table de déduplication (MySQL) :

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

Exemple de logique de consommateur idempotent :

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

Remplacez insertMessageKey et processOrder par vos méthodes réelles d'accès à la base de données et de logique métier.

Pour les scénarios à haut débit où une base de données relationnelle deviendrait un goulot d'étranglement, privilégiez une approche basée sur Redis avec SETNX (SET if Not eXists).