Tüm Ürünler
Search
Belge Merkezi

ApsaraMQ for RocketMQ:Send and receive transactional messages

Son Güncelleme:Oct 04, 2026

ApsaraMQ for RocketMQ, eXtended Architecture (X/Open XA) protokolüne benzer bir dağıtık işlem işleme özelliği sunar. ApsaraMQ for RocketMQ, işlemsel tutarlılığı sağlamak için işlemsel mesajları kullanır.

Aşağıdaki örnek kod, .NET için TCP istemci SDK'sını kullanarak işlemsel mesajların nasıl gönderileceğini ve alınacağını gösterir.

Zamanlanmış mesajlar Internet, Çin (Hangzhou), Çin (Pekin), Çin (Şanghay) ve Çin (Shenzhen) bölgelerinde desteklenir.

İşlemsel mesajların çalışma mekanizması

Bir işlemsel mesaj üç aşamadan geçer: yarım mesaj teslimi, yerel işlem çalıştırma ve son taahhüt veya geri alma.

  1. Üretici, bir yarım mesajı ApsaraMQ for RocketMQ aracı sunucusuna (broker) gönderir. Yarım mesaj aracı sunucusu tarafından kalıcı hale getirilir ancak henüz tüketicilere teslim edilmez.

  2. Aracı sunucu yarım mesajı onaylar. Üretici ardından yerel işlemi yürütür.

  3. Yerel işlem sonucuna bağlı olarak üretici, aracı sunucuya bir taahhüt veya geri alma durumu gönderir:

    • Taahhüt: Aracı sunucu, yarım mesajı teslim edilebilir olarak işaretler. Tüketiciler artık bu mesajı alabilir.

    • Geri alma: Aracı sunucu yarım mesajı atar. Tüketiciler mesajı asla görmez.

  4. Aracı sunucu bir taahhüt veya geri alma almazsa — örneğin, üretici çöker veya bilinmeyen bir durum döndürürse — düzenli aralıklarla üretici kümesindeki bir üreticiye bir işlem durumu denetimi isteği gönderir. Üretici yerel işlem durumunu kontrol eder ve nihai durumu bildirir.

Transactional message interaction flow

Daha fazla bilgi için bkz. İşlemsel mesajlar.

Ön koşullar

  • .NET için SDK indirildi. Daha fazla bilgi için bkz. Sürüm notları.

  • Ortam hazırlandı. Daha fazla bilgi için bkz. .NET SDK ortamını hazırlama.

  • Kodda belirtmek istediğiniz kaynaklar ApsaraMQ for RocketMQ konsolunda oluşturuldu. Bu kaynaklar arasında örnekler, konular ve tüketici grupları bulunur. Daha fazla bilgi için bkz. Kaynak oluşturma.

  • Alibaba Cloud hesabınızın AccessKey çifti alındı. Daha fazla bilgi için bkz. AccessKey oluşturma.

İşlemsel mesaj gönderme

Bir işlemsel mesaj göndermek için iki bileşen gerekir:

Bileşen Sınıf Amaç
İşlem yürütücüsü LocalTransactionExecuter Yarım mesaj başarıyla gönderildikten sonra yerel işlemi çalıştırır ve ilk durumu döndürür.
İşlem denetleyicisi LocalTransactionChecker Yerel işlem durumunu sorgular ve aracı sunucu nihai bir taahhüt veya geri alma almadığında bunu aracı sunucuya bildirir.

1. Adım: İşlem yürütücüsünü uygulama

İşlem yürütücüsü, yarım mesaj gönderildikten hemen sonra çalışır. Yerel işlem mantığını yürütür ve üç durumdan birini döndürür:

Durum Etki
TransactionStatus.CommitTransaction İşlem taahhüt edilir. Tüketici mesajı alabilir.
TransactionStatus.RollbackTransaction İşlem geri alınır. Mesaj atılır.
TransactionStatus.Unknow Durum bilinmiyor. Aracı sunucu daha sonra bunu çözmek için işlem denetleyicisini çağırır.
TransactionStatus.Unknow SDK'daki gerçek enum yazımıdır (sonundaki n harfi olmadan).
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.getUserProperties("VincentNoUser"));

            // Two messages can have the same message body but different IDs.
            // You cannot query the ID of the current message in the ApsaraMQ for RocketMQ console.
            // Use the message ID or a CRC32/MD5 hash of the message body
            // to implement idempotent processing and prevent duplicate messages.
            string msgId = value.getMsgID();

            TransactionStatus transactionStatus = TransactionStatus.Unknow;
            try {
                boolean isCommit = The execution result of the local transaction;
                if (isCommit) {
                    // Local transaction succeeded -- commit the message.
                    transactionStatus = TransactionStatus.CommitTransaction;
                } else {
                    // Local transaction failed -- roll back the message.
                    transactionStatus = TransactionStatus.RollbackTransaction;
                }
            } catch (Exception e) {
                // Handle exceptions from the local transaction.
                // Returning Unknow triggers the transaction checker later.
            }
            return transactionStatus;
        }
    }
}

2. Adım: İşlem denetleyicisini uygulama

İşlem denetleyicisi, aracı sunucudan gelen durum denetleme isteklerini işler. Aracı sunucu bir yarım mesaj için taahhüt veya geri alma almazsa — örneğin, uygulama çıkar veya yürütücü TransactionStatus.Unknow döndürürse — düzenli aralıklarla üretici kümesindeki bir üretici örneğini sorgular.

check yöntemi şunları yapmalıdır:

  1. Yarım mesaja karşılık gelen yerel işlemin taahhüt edilip edilmediğini veya geri alınıp alınmadığını belirleyin.

  2. Uygun TransactionStatus değerini aracı sunucuya döndürü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.getUserProperties("VincentNoUser"));

        // Two messages can have the same message body but different IDs.
        // You cannot query the ID of the current message in the ApsaraMQ for RocketMQ console.
        // Use the message ID or a CRC32/MD5 hash of the message body
        // to look up the local transaction result.
        // If your business logic is idempotent, the deduplication check is optional.
        string msgId = value.getMsgID();

        TransactionStatus transactionStatus = TransactionStatus.Unknow;
        try {
            boolean isCommit = The execution result of the local transaction;
            if (isCommit) {
                // Local transaction succeeded -- commit the message.
                transactionStatus = TransactionStatus.CommitTransaction;
            } else {
                // Local transaction failed -- roll back the message.
                transactionStatus = TransactionStatus.RollbackTransaction;
            }
        } catch (Exception e) {
            // Handle exceptions. Returning Unknow keeps the half message
            // pending and the broker will check again later.
        }
        return transactionStatus;
    }
}

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

İşlem denetleyicisi ile bir TransactionProducer oluşturun, ardından işlem yürütücüsünü tetiklemek için bir yarım mesaj gönderin.

Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:

Yer Tutucu Açıklama Örnek
<tcp-endpoint> ApsaraMQ for RocketMQ konsolundaki Instance Details sayfasından TCP uç noktası http://MQ_INST_xxx.mq-internet-access.mq-internet.aliyuncs.com:80
<topic-name> ApsaraMQ for RocketMQ konsolunda oluşturulan konu TransactionTopic
<message-body> Mesajın içeriği Hello RocketMQ
class onscsharp
{
    static void Main(string[] args)
    {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();

        // TCP endpoint. Find this value in the TCP Endpoint section
        // of the Instance Details page in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<tcp-endpoint>");

        // Topic created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<topic-name>");

        // Message content.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.MsgContent, "<message-body>");

        // Make sure the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
        // and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));

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

        // Call start() once before sending messages.
        // After the producer starts, you can send messages concurrently from multiple threads.
        pProducer.start();

        Message msg = new Message(
            factoryInfo.getPublishTopics(),   // Topic
            "TagA",                           // Tag for message filtering
            factoryInfo.getMessageContent()   // Message body
        );

        // Set a message key for tracing. Use a business identifier
        // (for example, an order ID) that is globally unique when possible.
        // You can use this key to look up the message in the ApsaraMQ for RocketMQ console.
        msg.setKey("ORDERID_100");

        // Send the half message and run the local transaction.
        try
        {
            LocalTransactionExecuter myExecuter = new MyLocalTransactionExecuter();
            SendResultONS sendResult = pProducer.send(msg, myExecuter);
        }
        catch (ONSClientException e)
        {
            Console.WriteLine("\nexception of sendmsg:{0}", e.what());
        }

        // Shut down the producer before the application exits to release resources.
        // A producer cannot be restarted after shutdown.
        pProducer.shutdown();
    }
}

İşlemsel mesajlara abonelik

Tüketiciler, işlemsel mesajları normal mesajlarla aynı şekilde işler. Özel bir abonelik mantığı gerekmez.

Abonelik örnek kodu için bkz. Mesajlara abone olma.