Todos os produtos
Search
Central de documentação

ApsaraMQ for RocketMQ:Send and receive transactional messages

Última atualização: Jun 27, 2026

O ApsaraMQ for RocketMQ oferece um recurso de processamento de transações distribuídas semelhante ao protocolo eXtended Architecture (X/Open XA). O ApsaraMQ for RocketMQ usa mensagens transacionais para garantir a consistência das transações.

O exemplo de código a seguir demonstra como enviar e receber mensagens transacionais com o SDK de cliente TCP para .NET.

Mensagens agendadas têm suporte nas regiões Internet, China (Hangzhou), China (Beijing), China (Shanghai) e China (Shenzhen).

Funcionamento das mensagens transacionais

Uma mensagem transacional passa por três fases: entrega da half message, execução da transação local e commit ou rollback final.

  1. O produtor envia uma half message ao broker do ApsaraMQ for RocketMQ. O broker persiste a half message, mas ainda não a entrega aos consumidores.

  2. O broker confirma o recebimento da half message. Em seguida, o produtor executa a transação local.

  3. Com base no resultado da transação local, o produtor envia um status de commit ou rollback ao broker:

    • Commit: O broker marca a half message como entregável. Os consumidores podem recebê-la a partir desse momento.

    • Rollback: O broker descarta a half message. Os consumidores nunca a recebem.

  4. Se o broker não receber um commit ou rollback — por exemplo, se o produtor falhar ou retornar um status desconhecido —, ele enviará periodicamente uma solicitação de verificação de status da transação a um produtor no cluster de produtores. O produtor verifica o estado da transação local e relata o status final.

Transactional message interaction flow

Para obter mais informações, consulte Mensagens transacionais.

Pré-requisitos

  • Baixe o SDK para .NET. Para obter mais informações, consulte Notas de versão.

  • Prepare o ambiente. Para obter mais informações, consulte Preparar o ambiente do SDK .NET.

  • Crie os recursos que você usará no código no console do ApsaraMQ for RocketMQ. Esses recursos incluem instâncias, tópicos e grupos de consumidores. Para obter mais informações, consulte Criar recursos.

  • Obtenha o par de AccessKey da sua conta Alibaba Cloud. Para obter mais informações, consulte Criar um AccessKey.

Enviar mensagens transacionais

O envio de uma mensagem transacional exige dois componentes:

Componente

Classe

Finalidade

Executor de transação

LocalTransactionExecuter

Executa a transação local após o envio bem-sucedido da half message e retorna um status inicial.

Verificador de transação

LocalTransactionChecker

Consulta o estado da transação local e o relata ao broker quando este não recebe um commit ou rollback final.

Etapa 1: Implementar o executor de transação

O executor de transação é acionado imediatamente após o envio da half message. Ele executa a lógica da transação local e retorna um dos três status a seguir:

Status

Efeito

TransactionStatus.CommitTransaction

A transação é confirmada. O consumidor pode receber a mensagem.

TransactionStatus.RollbackTransaction

A transação é revertida. A mensagem é descartada.

TransactionStatus.Unknow

Status desconhecido. O broker invocará posteriormente o verificador de transação para resolver a pendência.

TransactionStatus.Unknow é a grafia real do enum no SDK (sem o n final).
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;
        }
    }
}

Etapa 2: Implementar o verificador de transação

O verificador de transação responde às solicitações de verificação de status enviadas pelo broker. Se o broker não receber um commit ou rollback para uma half message — por exemplo, se a aplicação encerrar ou o executor retornar TransactionStatus.Unknow —, ele consultará periodicamente uma instância de produtor no cluster de produtores.

O método check deve:

  1. Determinar se a transação local correspondente à half message foi confirmada ou revertida.

  2. Retornar o TransactionStatus apropriado ao broker.

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

Etapa 3: Enviar uma half message e executar a transação

Crie um TransactionProducer com o verificador de transação e envie uma half message para acionar o executor de transação.

Substitua os espaços reservados a seguir pelos valores reais:

Espaço reservado

Descrição

Exemplo

<tcp-endpoint>

Endpoint TCP obtido na página Instance Details no console do ApsaraMQ for RocketMQ

http://MQ_INST_xxx.mq-internet-access.mq-internet.aliyuncs.com:80

<topic-name>

Tópico criado no console do ApsaraMQ for RocketMQ

TransactionTopic

<message-body>

Conteúdo da mensagem

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

Inscrever-se em mensagens transacionais

Os consumidores processam mensagens transacionais da mesma forma que mensagens normais. Nenhuma lógica especial de inscrição é necessária.

Para visualizar o código de exemplo de inscrição, consulte Inscrever-se em mensagens.