ApsaraMQ for RocketMQ:Sample code for .NET SDK 1.x/2.x

using System;
using ons;

public class ProducerExampleForEx
{
    public ProducerExampleForEx()
    {
    }

    static void Main(string[] args) {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // Instance username and password from the Intelligent Authentication tab
        // of the Access Control page in the ApsaraMQ for RocketMQ console.
        // Do not use your Alibaba Cloud account AccessKey pair.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<your-instance-username>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<your-instance-password>");
        // Do not specify the instance ID for 5.x instances.

        // Group ID created in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "<your-group-id>");
        // Topic created in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
        // Endpoint from the console. Use domain:port only -- no http:// prefix, no resolved IP.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "<your-endpoint>");
        // Log path.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");

        // Create a producer. Producers are thread-safe; in most cases, each thread requires only one instance.
        Producer producer = ONSFactory.getInstance().createProducer(factoryInfo);

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

        // Create and send messages.
        Message msg = new Message(factoryInfo.getPublishTopics(), "tagA", "Example message body");
        msg.setKey(Guid.NewGuid().ToString());
        for (int i = 0; i < 32; i++) {
            try
            {
                SendResultONS sendResult = producer.send(msg);
                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();

    }
}

using System;
using System.Threading;
using System.Text;
using ons;

// Callback invoked when a message arrives from the broker.
public class MyMsgListener : MessageListener
{
    public MyMsgListener()
    {
    }

    ~MyMsgListener()
    {
    }

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

public class ConsumerExampleForEx
{
    public ConsumerExampleForEx()
    {
    }

    static void Main(string[] args) {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // Instance credentials from the Intelligent Authentication tab
        // of the Access Control page. Do not use Alibaba Cloud account keys.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<your-instance-username>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<your-instance-password>");
        // Do not specify the instance ID for 5.x instances.

        // Consumer group ID created in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "<your-group-id>");
        // Topic created in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
        // TCP endpoint from the Instance Details page in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<your-endpoint>");
        // Log path.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");
        // Consumption mode: clustering (default) or broadcasting.
        // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.CLUSTERING);
        // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.BROADCASTING);

        // Create a push consumer.
        PushConsumer consumer = ONSFactory.getInstance().createPushConsumer(factoryInfo);

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

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

        // Demo only. In production, keep the process running.
        Thread.Sleep(300000);

        // Shut down the consumer before exiting the process.
        consumer.shutdown();
    }
}

using System;
using ons;

public class OrderProducerExampleForEx
{
    public OrderProducerExampleForEx()
    {
    }

    static void Main(string[] args) {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // Instance credentials.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<your-instance-username>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<your-instance-password>");
        // Do not specify the instance ID for 5.x instances.

        factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "<your-group-id>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "<your-endpoint>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");

        // Create an order producer (not a regular Producer).
        OrderProducer producer = ONSFactory.getInstance().createOrderProducer(factoryInfo);

        producer.start();

        Message msg = new Message(factoryInfo.getPublishTopics(), "tagA", "Example message body");
        // Messages with the same sharding key are delivered in order.
        string shardingKey = "App-Test";
        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());
            }
        }

        producer.shutdown();

    }
}

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)
        {
            ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
            // Instance credentials.
            factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<your-instance-username>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<your-instance-password>");
            // Do not specify the instance ID for 5.x instances.

            factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "<your-group-id>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "<your-endpoint>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");

            // Create an order consumer (not a regular PushConsumer).
            OrderConsumer consumer = ONSFactory.getInstance().createOrderConsumer(factoryInfo);

            consumer.subscribe(factoryInfo.getPublishTopics(), "*",new MyMsgOrderListener());

            consumer.start();

            // Demo only. In production, keep the process running.
            Thread.Sleep(30000);

            consumer.shutdown();
        }
    }
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Runtime.InteropServices;
using ons;

namespace ons
{
    class onscsharp
    {
        static void Main(string[] args)
        {
            ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
            factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "<your-group-id>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "<your-endpoint>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty.MsgContent, "<your-message-content>");
            // Instance credentials.
            factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<your-instance-username>");
            factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<your-instance-password>");
            // Do not specify the instance ID for 5.x instances.

            Producer pProducer = ONSFactory.getInstance().createProducer(factoryInfo);

            // Call start() once before sending messages.
            pProducer.start();

            Message msg = new Message(
                factoryInfo.getPublishTopics(),
                "TagA",
                factoryInfo.getMessageContent()
            );

            // Optional: set a business key for message tracing in the console.
            msg.setKey("ORDERID_100");

            // Set the delivery time in milliseconds. The message is delivered after this timestamp.
            // In this example, the message is delivered 3 seconds from now.
            long deliverTime = System.currentTimeMillis() + 3000;
            msg.setStartDeliverTime(deliverTime);

            try
            {
                SendResultONS sendResult = pProducer.send(msg);
            }
            catch(ONSClientException e)
            {
                // Handle the send failure.
            }

            // Shut down the producer before exiting. Skipping shutdown may cause memory leaks.
            pProducer.shutdown();

        }
 }
}

Étape 1 : Envoyer un demi-message et exécuter la transaction locale

Implémentez l'interface LocalTransactionExecuter pour exécuter votre logique métier après l'envoi du demi-message. Renvoyez CommitTransaction, RollbackTransaction ou Unknow selon le résultat obtenu.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Runtime.InteropServices;
using ons;

namespace ons
{
public class MyLocalTransactionExecuter : LocalTransactionExecuter
{
    public MyLocalTransactionExecuter()
    {
    }

    ~MyLocalTransactionExecuter()
    {
    }
    public override TransactionStatus execute(Message value)
    {
            Console.WriteLine("execute topic: {0}, tag:{1}, key:{2}, msgId:{3},msgbody:{4}, userProperty:{5}",
            value.getTopic(), value.getTag(), value.getKey(), value.getMsgID(), value.getBody(), value.getUserProperty("VincentNoUser"));

            // Use the message ID and a CRC32/MD5 hash of the body to deduplicate messages.
            string msgId = value.getMsgID();

            TransactionStatus transactionStatus = TransactionStatus.Unknow;
            try {
                boolean isCommit = Execution result of the local transaction;
                if (isCommit) {
                    // Commit the message if the local transaction succeeds.
                    transactionStatus = TransactionStatus.CommitTransaction;
                } else {
                    // Roll back the message if the local transaction fails.
                    transactionStatus = TransactionStatus.RollbackTransaction;
                }
            } catch (Exception e) {
                // Handle the exception.
            }
            return transactionStatus;
    }
}
class onscsharp
{

    static void Main(string[] args)
    {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "<your-endpoint>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty.MsgContent, "<your-message-content>");
        // Instance credentials.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<your-instance-username>");
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<your-instance-password>");
        // Do not specify the instance ID for 5.x instances.

        // Create a transaction producer with a transaction checker.
        LocalTransactionChecker myChecker = new MyLocalTransactionChecker();
        TransactionProducer pProducer =ONSFactory.getInstance().createTransactionProducer(factoryInfo,ref myChecker);

        // Call start() once. After startup, send messages concurrently from multiple threads.
        pProducer.start();

            Message msg = new Message(
            factoryInfo.getPublishTopics(),
            "TagA",
            factoryInfo.getMessageContent()
        );

        // Optional: set a business key for message tracing in the console.
        msg.setKey("ORDERID_100");

        try
        {
            LocalTransactionExecuter myExecuter = new MyLocalTransactionExecuter();
            SendResultONS sendResult = pProducer.send(msg, ref myExecuter);
        }
        catch(ONSClientException e)
        {
            Console.WriteLine("\nexception of sendmsg:{0}",e.what() );
        }

        // Shut down the producer before exiting. The producer cannot be restarted after shutdown.
        pProducer.shutdown();
    }
}
}

Étape 2 : Implémenter le vérificateur de statut de transaction

Le broker invoque la méthode LocalTransactionChecker.check() pour vérifier le statut de la transaction lorsqu'il n'a reçu ni confirmation ni annulation. Implémentez une logique idempotente pour gérer les appels de vérification répétés.

public class MyLocalTransactionChecker : LocalTransactionChecker
{
    public MyLocalTransactionChecker()
    {
    }
    ~MyLocalTransactionChecker()
    {
    }
    public override TransactionStatus check(Message value)
    {
            Console.WriteLine("check topic: {0}, tag:{1}, key:{2}, msgId:{3},msgbody:{4}, userProperty:{5}",
            value.getTopic(), value.getTag(), value.getKey(), value.getMsgID(), value.getBody(), value.getUserProperty("VincentNoUser"));
            // Use the message ID and a CRC32/MD5 hash of the body to deduplicate.
            string msgId = value.getMsgID();

            TransactionStatus transactionStatus = TransactionStatus.Unknow;
            try {
                boolean isCommit = Execution result of the local transaction;
                if (isCommit) {
                    // Commit the message if the local transaction succeeded.
                    transactionStatus = TransactionStatus.CommitTransaction;
                } else {
                    // Roll back the message if the local transaction failed.
                    transactionStatus = TransactionStatus.RollbackTransaction;
                }
            } catch (Exception e) {
                // Handle the exception.
            }
            return transactionStatus;
    }
}