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

        }
 }
}

Adım 1: Yarım mesaj gönderme ve yerel işlemi çalıştırma

Yarım mesaj gönderildikten sonra iş mantığınızı çalıştırmak için LocalTransactionExecuter uygulayın. Sonuca göre CommitTransaction, RollbackTransaction veya Unknow döndürün.

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

Adım 2: İşlem durumu denetleyicisini uygulama

Broker, bir commit veya rollback almadığında işlem durumunu doğrulamak için LocalTransactionChecker.check() yöntemini çağırır. Tekrarlanan denetim çağrılarını işlemek için yinelenmezlik mantığı uygulayın.

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