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

Les messages ordonnés dans ApsaraMQ for RocketMQ garantissent une livraison stricte de type premier entré, premier sorti (FIFO) au sein d'un topic ou d'une partition. Utilisez les messages ordonnés lorsque la séquence de traitement impacte votre logique métier, par exemple pour l'appariement de transactions où la première enchère à un prix donné doit l'emporter, ou pour la synchronisation des modifications de base de données où les opérations d'insertion, de mise à jour et de suppression doivent être rejouées dans l'ordre.

Les exemples C++ suivants illustrent l'envoi et la réception de messages ordonnés via le SDK client HTTP.

Fonctionnement de l'ordonnancement

ApsaraMQ for RocketMQ prend en charge deux périmètres d'ordonnancement :

Messages globalement ordonnés : tous les messages d'un topic sont livrés dans l'ordre FIFO. Ce périmètre convient lorsque chaque message doit être traité de manière séquentielle.
Messages ordonnés par partition : les messages sont répartis entre les partitions selon une clé de sharding. Au sein de chaque partition, les messages sont livrés dans l'ordre FIFO. Cette approche offre un débit plus élevé lorsque seul un sous-ensemble de messages liés nécessite un ordre spécifique, comme un regroupement par ID de commande ou par ID d'utilisateur.

La clé de sharding détermine la partition qui recevra un message. Les messages partageant la même clé de sharding aboutissent toujours dans la même partition et sont consommés dans l'ordre d'envoi. Une clé de sharding se distingue d'une clé de message.

Pour plus d'informations, consultez la rubrique Messages ordonnés.

Ordre de production

Le broker détermine l'ordre des messages en fonction de la séquence d'envoi issue d'un unique producteur ou thread. Pour préserver cet ordre :

Expédiez les messages liés depuis un seul producteur sur un seul thread.
Si plusieurs producteurs ou threads envoient des données simultanément, le broker classe les messages par heure d'arrivée, ce qui peut différer de la séquence métier prévue.

Ordre de consommation

Le consommateur extrait les messages ordonnés par partition depuis une ou plusieurs partitions et traite ceux de chaque partition dans l'ordre d'envoi. La garantie d'ordre repose sur un accusé de réception approprié :

Le consommateur doit acquitter tous les messages d'un lot avant d'extraire le lot suivant de la même partition.
Si le broker ne reçoit pas d'accusé de réception (ACK) pour un message avant l'expiration du délai, il renvoie ce message.
Chaque nouvelle tentative de livraison génère un nouveau handle de réception associé à un horodatage unique.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Le SDK client HTTP C++ est installé. Consultez la rubrique Préparer l'environnement
Une instance, un topic et un groupe de consommateurs ApsaraMQ for RocketMQ ont été créés dans la console. Consultez la rubrique Créer des ressources
Une paire AccessKey Alibaba Cloud est disponible. Consultez la rubrique Créer une paire AccessKey

Envoyer des messages ordonnés

Important

Le broker séquence les messages selon leur ordre d'arrivée provenant d'un unique producteur ou thread. Si plusieurs producteurs ou threads envoient des données simultanément, l'ordre côté broker peut diverger de l'ordre d'envoi côté métier.

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

Espace réservé	Description	Où le trouver
`${HTTP_ENDPOINT}`	Endpoint HTTP de votre instance	Page Instance Details > section HTTP Endpoint dans la console ApsaraMQ for RocketMQ
`${TOPIC}`	Nom du topic	Le topic que vous avez créé dans la console
`${INSTANCE_ID}`	ID de l'instance	Page Instance Details dans la console. Définissez cette valeur sur une chaîne vide si l'instance ne possède pas de namespace

//#include <iostream>
#include <fstream>
#include <time.h>
#include "mq_http_sdk/mq_client.h"

using namespace std;
using namespace mq::http::sdk;

int main() {

    MQClient mqClient(
            // The HTTP endpoint. You can obtain the endpoint in the HTTP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console.
            "${HTTP_ENDPOINT}",
            // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
	          // The AccessKey ID that is used for authentication.
	          System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	          // The AccessKey secret that is used for authentication.
	          System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
            );

    // The topic in which the message is produced. You must create the topic in the ApsaraMQ for RocketMQ console.
    string topic = "${TOPIC}";
    // The ID of the instance to which the topic belongs. You must create the instance in the ApsaraMQ for RocketMQ console.
    // If the instance has a namespace, specify the ID of the instance. If the instance does not have a namespace, set the instanceID parameter to null or an empty string. You can obtain the namespace of the instance on the Instance Details page in the ApsaraMQ for RocketMQ console.
    string instanceId = "${INSTANCE_ID}";

    MQProducerPtr producer;
    if (instanceId == "") {
        producer = mqClient.getProducerRef(topic);
    } else {
        producer = mqClient.getProducerRef(instanceId, topic);
    }

    try {
        // Cyclically send four messages.
        for (int i = 0; i < 8; i++)
        {
            PublishMessageResponse pmResp;
            // The message content.
            TopicMessage pubMsg("Hello, mq!order msg!");
            // The sharding key that is used to distribute ordered messages to a specific partition. Sharding keys can be used to identify partitions. A sharding key is different from a message key.
            pubMsg.setShardingKey(std::to_string(i % 2));
            // The custom attributes of the message.
            pubMsg.putProperty("a",std::to_string(i));
            producer->publishMessage(pubMsg, pmResp);
            cout << "Publish mq message success. Topic is: " << topic
                << ", msgId is:" << pmResp.getMessageId()
                << ", bodyMD5 is:" << pmResp.getMessageBodyMD5() << endl;
        }
    } catch (MQServerException& me) {
        cout << "Request Failed: " + me.GetErrorCode() << ", requestId is:" << me.GetRequestId() << endl;
        return -1;
    } catch (MQExceptionBase& mb) {
        cout << "Request Failed: " + mb.ToString() << endl;
        return -2;
    }

    return 0;
}

Cet exemple envoie huit messages en utilisant deux clés de sharding (0 et 1 via i % 2). Les messages partageant la même clé de sharding sont livrés à la même partition dans l'ordre d'envoi. L'appel putProperty associe un attribut personnalisé à chaque message afin d'en faciliter le suivi.

Recevoir des messages ordonnés

Appelez consumeMessageOrderly plutôt que la méthode de consommation standard pour maintenir l'ordre au niveau des partitions. En mode d'interrogation longue (long polling), le broker maintient la requête ouverte pendant la durée spécifiée si aucun message n'est disponible et répond immédiatement dès qu'un message arrive.

Outre les espaces réservés mentionnés précédemment, remplacez les éléments suivants :

Espace réservé	Description	Où le trouver
`${GROUP_ID}`	ID du groupe de consommateurs	Le groupe de consommateurs que vous avez créé dans la console ApsaraMQ for RocketMQ

#include <vector>
#include <fstream>
#include "mq_http_sdk/mq_client.h"

#ifdef _WIN32
#include <windows.h>
#else
#include <unistd.h>
#endif

using namespace std;
using namespace mq::http::sdk;

int main() {

    MQClient mqClient(
            // The HTTP endpoint. You can obtain the endpoint in the HTTP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console.
            "${HTTP_ENDPOINT}",
            // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
	          // The AccessKey ID that is used for authentication.
	          System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	          // The AccessKey secret that is used for authentication.
	          System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
            );

    // The topic in which the message is produced. You must create the topic in the ApsaraMQ for RocketMQ console.
    string topic = "${TOPIC}";
    // The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console.
    string groupId = "${GROUP_ID}";
    // The ID of the instance to which the topic belongs. You must create the instance in the ApsaraMQ for RocketMQ console.
    // If the instance has a namespace, specify the ID of the instance. If the instance does not have a namespace, set the instanceID parameter to null or an empty string. You can obtain the namespace of the instance on the Instance Details page in the ApsaraMQ for RocketMQ console.
    string instanceId = "${INSTANCE_ID}";

    MQConsumerPtr consumer;
    if (instanceId == "") {
        consumer = mqClient.getConsumerRef(topic, groupId);
    } else {
        consumer = mqClient.getConsumerRef(instanceId, topic, groupId, "");
    }

    do {
        try {
            std::vector<Message> messages;
            // Consume messages in long polling mode. The consumer may pull partitionally ordered messages from multiple partitions. The consumer consumes messages from the same partition in the order in which the messages are sent.
            // Assume that a consumer pulls partitionally ordered messages from one partition. If the broker fails to receive an acknowledgement (ACK) for a message from the consumer, the broker delivers the message in the partition to the consumer again.
            // The consumer can consume the next batch of messages from a partition only after all messages that are pulled from the partition in the previous batch are acknowledged as consumed.
            // In long polling mode, if no message in the topic is available for consumption, the request is suspended on the broker for the specified period of time. If a message becomes available for consumption within the specified period of time, the broker immediately sends a response to the consumer. In this example, the value is specified as 3 seconds.
            consumer->consumeMessageOrderly(
                    3, // The maximum number of messages that can be consumed at a time. In this example, the value is specified as 3. The maximum value that you can specify is 16.
                    3, // The duration of a long polling cycle. Unit: seconds. In this example, the value is specified as 3. The maximum value that you can specify is 30.
                    messages
            );
            cout << "Consume: " << messages.size() << " Messages!" << endl;

            // The message consumption logic.
            std::vector<std::string> receiptHandles;
            for (std::vector<Message>::iterator iter = messages.begin();
                    iter != messages.end(); ++iter)
            {
                cout << "MessageId: " << iter->getMessageId()
                    << " PublishTime: " << iter->getPublishTime()
                    << " Tag: " << iter->getMessageTag()
                    << " Body: " << iter->getMessageBody()
                    << " FirstConsumeTime: " << iter->getFirstConsumeTime()
                    << " NextConsumeTime: " << iter->getNextConsumeTime()
                    << " ConsumedTimes: " << iter->getConsumedTimes()
                    << " Properties: " << iter->getPropertiesAsString()
                    << " ShardingKey: " << iter->getShardingKey() << endl;
                receiptHandles.push_back(iter->getReceiptHandle());
            }

            // Obtain an ACK from the consumer.
            // If the broker fails to receive an ACK for a message from the consumer before the period of time that is specified by the Message.NextConsumeTime parameter elapses, the broker delivers the message for consumption again.
            // A unique timestamp is specified for the handle of a message each time the message is consumed.
            AckMessageResponse bdmResp;
            consumer->ackMessage(receiptHandles, bdmResp);
            if (!bdmResp.isSuccess()) {
                // If the handle of a message times out, the broker cannot receive an ACK for the message from the consumer.
                const std::vector<AckMessageFailedItem>& failedItems =
                    bdmResp.getAckMessageFailedItem();
                for (std::vector<AckMessageFailedItem>::const_iterator iter = failedItems.begin();
                        iter != failedItems.end(); ++iter)
                {
                    cout << "AckFailedItem: " << iter->errorCode
                        << "  " << iter->receiptHandle << endl;
                }
            } else {
                cout << "Ack: " << messages.size() << " messages suc!" << endl;
            }
        } catch (MQServerException& me) {
            if (me.GetErrorCode() == "MessageNotExist") {
                cout << "No message to consume! RequestId: " + me.GetRequestId() << endl;
                continue;
            }
            cout << "Request Failed: " + me.GetErrorCode() + ".RequestId: " + me.GetRequestId() << endl;
#ifdef _WIN32
            Sleep(2000);
#else
            usleep(2000 * 1000);
#endif
        } catch (MQExceptionBase& mb) {
            cout << "Request Failed: " + mb.ToString() << endl;
#ifdef _WIN32
            Sleep(2000);
#else
            usleep(2000 * 1000);
#endif
        }

    } while(true);
}

Paramètres de `consumeMessageOrderly`

Paramètre	Description	Plage
Taille du lot	Nombre maximal de messages par extraction	Maximum : 16
Durée de l'interrogation longue	Durée en secondes pendant laquelle le broker maintient la requête en attente en l'absence de message	Maximum : 30

Comportement d'acquittement

Après avoir traité chaque lot, appelez ackMessage avec les handles de réception de tous les messages traités.

Verrouillage par lot : le consommateur ne peut pas extraire le lot suivant d'une partition tant que chaque message du lot actuel n'a pas été acquitté.
Nouvelle livraison après expiration : si le broker ne reçoit pas d'ACK avant NextConsumeTime, il renvoie le message avec un nouveau handle de réception contenant un horodatage unique.
**Gestion de MessageNotExist :** ce code d'erreur indique qu'aucun message n'est disponible. L'exemple gère cette situation en poursuivant la boucle d'interrogation sans délai.
Récupération d'erreur : en cas d'autres exceptions, l'exemple attend 2 secondes avant de réessayer afin d'éviter des boucles d'erreur serrées.

Notes d'utilisation

Un seul producteur par périmètre d'ordonnancement. Expédiez tous les messages nécessitant un ordre relatif depuis un unique producteur sur un seul thread. L'utilisation de plusieurs producteurs ou threads risque de rompre l'ordre prévu.
Choisissez une granularité adaptée pour les clés de sharding. Privilégiez des clés significatives pour le métier, telles que les ID de commande ou les ID utilisateur. Évitez de regrouper trop de messages sous une seule clé de sharding, car cela surchargerait une partition unique et limiterait le débit.
Acquittez rapidement. Un acquittement tardif bloque la consommation des messages suivants dans la même partition, augmentant ainsi la latence de bout en bout.
Gérez les pannes sans rompre l'ordre. Si le traitement d'un message échoue, laissez-le expirer pour déclencher une nouvelle livraison plutôt que de l'acquitter et de passer au suivant, ce qui compromettrait la garantie d'ordre.