Envoyer et recevoir des messages ordonnés - ApsaraMQ for RocketMQ

ApsaraMQ for RocketMQ distribue les messages ordonnés selon une séquence stricte de type premier entré, premier sorti (FIFO). Utilisez les messages ordonnés lorsque votre application exige un traitement des messages exactement dans l'ordre d'envoi, par exemple pour traiter les commandes commerciales selon leur heure de soumission ou synchroniser les modifications de base de données de manière séquentielle.

Cette rubrique fournit des exemples de code pour envoyer et recevoir des messages ordonnés à l'aide du SDK client HTTP pour Java.

Fonctionnement des messages ordonnés

Il existe deux catégories de messages ordonnés :

Messages globalement ordonnés : Tous les messages d'un topic suivent une seule séquence FIFO. Cette approche offre la garantie d'ordre la plus forte, mais limite le débit à une seule partition.
Messages ordonnés par partition : Les messages sont répartis entre les partitions via une clé de sharding. Au sein de chaque partition, les messages respectent l'ordre FIFO. Les différentes partitions sont consommées indépendamment, ce qui permet d'obtenir un débit plus élevé tout en préservant l'ordre là où cela est nécessaire.

Une clé de sharding identifie la partition à laquelle appartient un message. Les messages partageant la même clé de sharding sont toujours acheminés vers la même partition et consommés dans l'ordre. Une clé de sharding est distincte d'une clé de message.

Important

Un courtier ApsaraMQ for RocketMQ détermine l'ordre de génération des messages en fonction de l'ordre dans lequel l'expéditeur utilise un seul producteur ou thread pour envoyer les messages. Si l'expéditeur utilise plusieurs producteurs ou threads pour envoyer des messages simultanément, l'ordre des messages est déterminé par l'ordre de réception par le courtier ApsaraMQ for RocketMQ. Cet ordre peut différer de l'ordre d'envoi côté métier.

Ordre de production et ordre de consommation

La livraison FIFO de bout en bout dépend de deux facteurs indépendants :

Ordre de production	Ordre de consommation	Résultat
Producteur unique, envois sériels	Consommation ordonnée (`consumeMessageOrderly`)	Garantie FIFO par clé de sharding
Plusieurs producteurs ou threads	Consommation ordonnée	L'ordre est déterminé par l'arrivée chez le courtier, et non par l'intention métier
Producteur unique, envois sériels	Consommation concurrente	L'ordre de consommation n'est pas garanti

Pour garantir un ordre strict, les deux conditions suivantes doivent être remplies : envoyez les messages depuis un thread de producteur unique et consommez-les avec consumeMessageOrderly.

Prérequis

Avant de commencer, effectuez les opérations suivantes :

Installez le SDK pour Java. Pour plus d'informations, consultez la rubrique Configurer votre environnement Java.
Créez les ressources que vous souhaitez spécifier dans le code via la console ApsaraMQ for RocketMQ. Ces ressources incluent des instances, des topics et des groupes de consommateurs. Pour plus d'informations, consultez la rubrique Créer des ressources.
Obtenez la paire AccessKey de votre compte Alibaba Cloud. Pour plus d'informations, consultez la rubrique Créer une AccessKey.

Envoyer des messages ordonnés

L'exemple suivant envoie huit messages ordonnés répartis sur deux partitions à l'aide de clés de sharding.

Remplacez les espaces réservés par vos valeurs réelles :

Espace réservé	Description	Exemple
`<your-http-endpoint>`	Endpoint HTTP disponible sur la page Instance Details de la console ApsaraMQ for RocketMQ	`http://xxx.mqrest.cn-hangzhou.aliyuncs.com`
`<your-topic>`	Nom du topic	`OrderTopic`
`<your-instance-id>`	ID de l'instance. Définissez cette valeur sur `null` ou `""` si l'instance ne possède pas de namespace	`MQ_INST_xxx`

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQProducer;
import com.aliyun.mq.http.model.TopicMessage;

import java.util.Date;

public class OrderProducer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                "<your-http-endpoint>",
                // Read credentials from environment variables to avoid hardcoding secrets
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        final String topic = "<your-topic>";
        // Set to null or "" if the instance has no namespace
        final String instanceId = "<your-instance-id>";

        // Get the producer for the specified topic
        MQProducer producer;
        if (instanceId != null && instanceId != "") {
            producer = mqClient.getProducer(instanceId, topic);
        } else {
            producer = mqClient.getProducer(topic);
        }

        try {
            // Send 8 messages, alternating between 2 sharding keys (partitions)
            for (int i = 0; i < 8; i++) {
                TopicMessage pubMsg = new TopicMessage(
                        "hello mq!".getBytes(),   // Message body
                        "A"                        // Message tag for filtering
                );
                // Sharding key determines the partition. Messages with the same
                // sharding key are delivered to the same partition in FIFO order.
                pubMsg.setShardingKey(String.valueOf(i % 2));
                pubMsg.getProperties().put("a", String.valueOf(i));

                // Publish synchronously. No exception means the send succeeded.
                TopicMessage pubResultMsg = producer.publishMessage(pubMsg);

                System.out.println(new Date() + " Send mq message success. Topic is:" + topic
                        + ", msgId is: " + pubResultMsg.getMessageId()
                        + ", bodyMD5 is: " + pubResultMsg.getMessageBodyMD5());
            }
        } catch (Throwable e) {
            // Handle send failure: retry or persist the message for later delivery
            System.out.println(new Date() + " Send mq message failed. Topic is:" + topic);
            e.printStackTrace();
        }

        mqClient.close();
    }
}

Remarque

Chaque topic prend en charge un seul type de message. Un topic créé pour les messages ordonnés ne peut ni envoyer ni recevoir de messages normaux.

Recevoir des messages ordonnés

L'exemple suivant consomme des messages ordonnés à l'aide de consumeMessageOrderly, qui utilise le sondage long (long polling) et garantit l'ordre FIFO au niveau de la partition.

Remplacez les espaces réservés par vos valeurs réelles :

Espace réservé	Description	Exemple
`<your-http-endpoint>`	Endpoint HTTP disponible sur la page Instance Details	`http://xxx.mqrest.cn-hangzhou.aliyuncs.com`
`<your-topic>`	Nom du topic	`OrderTopic`
`<your-group-id>`	ID du groupe de consommateurs	`GID_Order`
`<your-instance-id>`	ID de l'instance. Définissez cette valeur sur `null` ou `""` si l'instance ne possède pas de namespace	`MQ_INST_xxx`

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.Message;

import java.util.ArrayList;
import java.util.List;

public class OrderConsumer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                "<your-http-endpoint>",
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
                System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        final String topic = "<your-topic>";
        final String groupId = "<your-group-id>";
        // Set to null or "" if the instance has no namespace
        final String instanceId = "<your-instance-id>";

        final MQConsumer consumer;
        if (instanceId != null && instanceId != "") {
            consumer = mqClient.getConsumer(instanceId, topic, groupId, null);
        } else {
            consumer = mqClient.getConsumer(topic, groupId);
        }

        // Continuously poll for messages
        do {
            List<Message> messages = null;

            try {
                // consumeMessageOrderly guarantees partition-level FIFO order.
                // The consumer must ACK all messages in a batch before the broker
                // delivers the next batch from the same partition.
                messages = consumer.consumeMessageOrderly(
                        3,  // Max messages per batch (up to 16)
                        3   // Long polling timeout in seconds (up to 30)
                );
            } catch (Throwable e) {
                e.printStackTrace();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }

            if (messages == null || messages.isEmpty()) {
                System.out.println(Thread.currentThread().getName() + ": no new message, continue!");
                continue;
            }

            // Process messages
            System.out.println("Receive " + messages.size() + " messages:");
            for (Message message : messages) {
                System.out.println(message);
                System.out.println("ShardingKey: " + message.getShardingKey()
                        + ", a:" + message.getProperties().get("a"));
            }

            // ACK consumed messages. If the broker does not receive an ACK before
            // the retry timeout, it redelivers the message.
            {
                List<String> handles = new ArrayList<String>();
                for (Message message : messages) {
                    handles.add(message.getReceiptHandle());
                }

                try {
                    consumer.ackMessage(handles);
                } catch (Throwable e) {
                    if (e instanceof AckMessageException) {
                        AckMessageException errors = (AckMessageException) e;
                        System.out.println("Ack message fail, requestId is:"
                                + errors.getRequestId() + ", fail handles:");
                        if (errors.getErrorMessages() != null) {
                            for (String errorHandle : errors.getErrorMessages().keySet()) {
                                System.out.println("Handle:" + errorHandle
                                        + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                        + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                            }
                        }
                        continue;
                    }
                    e.printStackTrace();
                }
            }
        } while (true);
    }
}

Fonctionnement de la consommation ordonnée

Lorsque vous appelez consumeMessageOrderly, le courtier applique les comportements suivants :

Verrouillage au niveau de la partition : Le consommateur peut extraire des messages de plusieurs partitions, mais au sein de chaque partition, les messages arrivent dans l'ordre d'envoi.
Validation par lot (ACK) : Le consommateur doit acquitter (ACK) tous les messages du lot actuel avant de recevoir le lot suivant provenant de la même partition. Si le courtier ne reçoit pas d'acquittement avant l'expiration du délai de nouvelle tentative, il redistribue le message non acquitté.
Sondage long (Long polling) : Si aucun message n'est disponible, le courtier maintient la requête pendant la durée de sondage spécifiée (jusqu'à 30 secondes) et répond immédiatement dès qu'un message arrive.

Notes d'utilisation

Topics à type unique : Chaque topic prend en charge un seul type de message. Ne mélangez pas les messages ordonnés avec des messages normaux sur le même topic.
Producteur unique pour un ordre strict : Envoyez tous les messages nécessitant un ordre relatif depuis un seul producteur sur un thread unique. L'utilisation de plusieurs producteurs ou de threads concurrents rompt la garantie d'ordre.
Conception de la clé de sharding : Choisissez une clé de sharding qui regroupe les messages connexes sans créer de partitions chaudes. Par exemple, utilisez un ID de commande pour conserver tous les événements d'une commande dans l'ordre, plutôt qu'un ID client qui pourrait concentrer trop de messages sur une seule partition.
Acquittement rapide : Un acquittement tardif bloque la livraison des messages suivants pour toute la partition. Traitez et acquittez chaque lot aussi rapidement que possible.
Expiration du handle de réception : Chaque handle de réception de message possède un horodatage unique. Si le handle expire avant que vous ne l'acquittiez, l'acquittement échoue et le courtier redistribue le message.

Voir aussi

Messages ordonnés : Garanties d'ordre, types d'ordre et concepts détaillés
Créer des ressources : Configurer des instances, des topics et des groupes de consommateurs dans la console