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.
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.
O broker confirma o recebimento da half message. Em seguida, o produtor executa a transação local.
-
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.
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.

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 |
|
Executa a transação local após o envio bem-sucedido da half message e retorna um status inicial. |
|
Verificador de transação |
|
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 |
|
|
A transação é confirmada. O consumidor pode receber a mensagem. |
|
|
A transação é revertida. A mensagem é descartada. |
|
|
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:
Determinar se a transação local correspondente à half message foi confirmada ou revertida.
Retornar o
TransactionStatusapropriado 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 |
|
|
Endpoint TCP obtido na página Instance Details no console do ApsaraMQ for RocketMQ |
|
|
|
Tópico criado no console do ApsaraMQ for RocketMQ |
|
|
|
Conteúdo da mensagem |
|
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.