Tous les produits
Search
Centre de documentation

ApsaraMQ for RocketMQ:Send and receive ordered messages

Dernière mise à jour :Aug 09, 2026

Dans ApsaraMQ for RocketMQ, les messages ordonnés sont consommés selon un ordre strict premier entré, premier sorti (FIFO). Cette rubrique fournit des exemples de code pour envoyer et recevoir des messages ordonnés à l'aide du SDK client TCP pour .NET.

Modes d'ordonnancement

ApsaraMQ for RocketMQ prend en charge deux modes d'ordonnancement :

  • Messages globalement ordonnés : tous les messages d'un topic sont publiés et consommés selon un ordre FIFO strict.

  • Messages ordonnés par partition : les messages d'un topic sont répartis entre les partitions via une clé de sharding. Au sein de chaque partition, les messages sont consommés selon un ordre FIFO strict.

Une clé de sharding est un champ qui achemine les messages ordonnés vers une partition spécifique. Elle diffère des clés utilisées pour les messages classiques.

Important

Le broker ApsaraMQ for RocketMQ détermine l'ordre des messages selon la séquence d'envoi d'un unique producteur ou thread. Si plusieurs producteurs ou threads envoient des messages simultanément, le broker classe les messages par heure d'arrivée, ce qui peut ne pas correspondre à l'ordre métier souhaité. Pour garantir l'ordonnancement, envoyez tous les messages connexes depuis un seul producteur ou thread.

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

Prérequis

Avant de commencer, assurez-vous d'avoir :

  • Téléchargé le SDK pour .NET. Pour plus d'informations, consultez les Notes de version

  • Préparé l'environnement. Pour plus d'informations, consultez la rubrique Préparation de l'environnement

  • Créé des instances, des topics et des groupes de consommateurs dans la console ApsaraMQ for RocketMQ. Pour plus d'informations, consultez la rubrique Création de ressources

  • Obtenu une paire AccessKey pour votre compte Alibaba Cloud. Pour plus d'informations, consultez la rubrique Création d'une paire AccessKey

Envoi de messages ordonnés

L'exemple de code suivant montre comment envoyer des messages ordonnés à l'aide du SDK client TCP pour .NET. Tous les messages partagent la même clé de sharding ; ils sont donc acheminés vers la même partition et consommés dans l'ordre d'envoi.

Pour consulter le dépôt de code complet, rendez-vous sur le dépôt de code ApsaraMQ for RocketMQ.

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

Espace réservé Description Exemple
<your-group-id> ID du groupe de producteurs créé dans la console GID_example
<your-topic> Topic créé dans la console T_example_topic_name
<your-tcp-endpoint> Endpoint TCP disponible sur la page Instance Details de la console NameSrv_Addr
<your-log-path> Chemin local pour les journaux du SDK C://log
<your-sharding-key> Clé de sharding qui détermine le routage vers la partition App-Test
using System;
using ons;

public class OrderProducerExampleForEx
{
    public OrderProducerExampleForEx()
    {
    }

    static void Main(string[] args) {
        // Configure the producer properties.
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
        // and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
        // AccessKey ID for authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        // AccessKey secret for authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // Producer group ID created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "<your-group-id>");
        // Topic created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
        // TCP endpoint from the Instance Details page in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<your-tcp-endpoint>");
        // Local path for SDK logs.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "<your-log-path>");

        // Create a producer instance.
        // Producer instances are thread-safe and can send messages to different topics.
        // In most cases, one producer instance per thread is sufficient.
        OrderProducer producer = ONSFactory.getInstance().createOrderProducer(factoryInfo);

        // Start the producer.
        producer.start();

        // Create a message with topic, tag, and body.
        Message msg = new Message(factoryInfo.getPublishTopics(), "tagA", "Example message body");
        // Define the sharding key. Messages with the same sharding key
        // are sent to the same partition and consumed in order.
        string shardingKey = "<your-sharding-key>";
        for (int i = 0; i < 32; i++) {
            try
            {
                SendResultONS sendResult = producer.send(msg, shardingKey);
                Console.WriteLine("send success {0}", sendResult.getMessageId());
            }
            catch (Exception ex)
            {
                Console.WriteLine("send failure{0}", ex.ToString());
            }
        }

        // Shut down the producer before exiting the thread.
        producer.shutdown();

    }
}

Réception de messages ordonnés

L'exemple de code suivant montre comment recevoir des messages ordonnés à l'aide du SDK client TCP pour .NET. Le consommateur traite les messages selon un ordre FIFO au sein de chaque partition.

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

Espace réservé Description Exemple
<your-group-id> ID du groupe de consommateurs créé dans la console GID_example
<your-topic> Topic créé dans la console T_example_topic_name
<your-tcp-endpoint> Endpoint TCP disponible sur la page Instance Details de la console NameSrv_Addr
<your-log-path> Chemin local pour les journaux du SDK C://log
using System;
using System.Text;
using System.Threading;
using ons;

namespace demo
{

    public class MyMsgOrderListener : MessageOrderListener
    {
        public MyMsgOrderListener()
        {

        }

        ~MyMsgOrderListener()
        {
        }

        public override ons.OrderAction consume(Message value, ConsumeOrderContext context)
        {
            Byte[] text = Encoding.Default.GetBytes(value.getBody());
            Console.WriteLine(Encoding.UTF8.GetString(text));
            return ons.OrderAction.Success;
        }
    }

    class OrderConsumerExampleForEx
    {
        static void Main(string[] args)
        {
            // Configure the consumer properties.
            ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
            // AccessKey ID for authentication.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "Your access key");
            // AccessKey secret for authentication.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "Your access secret");
            // Consumer group ID created in the ApsaraMQ for RocketMQ console.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "<your-group-id>");
            // Topic created in the ApsaraMQ for RocketMQ console.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
            // TCP endpoint from the Instance Details page in the console.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<your-tcp-endpoint>");
            // Local path for SDK logs.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "<your-log-path>");

            // Create a consumer instance.
            OrderConsumer consumer = ONSFactory.getInstance().createOrderConsumer(factoryInfo);

            // Subscribe to the topic with a wildcard tag filter.
            consumer.subscribe(factoryInfo.getPublishTopics(), "*", new MyMsgOrderListener());

            // Start the consumer.
            consumer.start();

            // Keep the main thread alive to continue receiving messages.
            Thread.Sleep(30000);

            // Shut down the consumer when it is no longer needed.
            consumer.shutdown();
        }
    }
}

Étapes suivantes