目前支持的域包括公网、华东1、华北2、华东2、华南1。

交互流程

消息队列 RocketMQ 的事务消息的交互流程如下图所示：

发送事务消息

发送事务消息包含以下两个步骤：

1. 发送半消息（Half Message）及执行本地事务。 示例代码如下：

 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"));
             // 消息 ID(有可能消息体一样，但消息 ID 不一样。当前消息 ID 在控制台无法查询)
             string msgId = value.getMsgID();
             // 消息体内容进行 crc32, 也可以使用其它的如 MD5
             // 消息 ID 和 crc32id 主要是用来防止消息重复
             // 如果要求消息绝对不重复，推荐做法是对消息体 body 使用 crc32或 MD5 来防止重复消息
             TransactionStatus transactionStatus = TransactionStatus.Unknow;
             try {
                 boolean isCommit = 本地事务执行结果;
                 if (isCommit) {
                     // 本地事务成功则提交消息
                     transactionStatus = TransactionStatus.CommitTransaction;
                 } else {
                     // 本地事务失败则回滚消息
                     transactionStatus = TransactionStatus.RollbackTransaction;
                 }
             } catch (Exception e) {
                 //exception handle
             }
             return transactionStatus；
     }
 }
 class onscsharp
 {
     static void Main(string[] args)
     {
         ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
         factoryInfo.setFactoryProperty(factoryInfo.NAMESRV_ADDR, "XXX");//设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
         factoryInfo.setFactoryProperty(factoryInfo.ProducerId, "");//您在控制台创建的 Group ID
         factoryInfo.setFactoryProperty(factoryInfo.PublishTopics, "");// 您在控制台创建的 Topic
         factoryInfo.setFactoryProperty(factoryInfo.MsgContent, "");//message body
         factoryInfo.setFactoryProperty(factoryInfo.AccessKey, "");//AccessKey 身份验证，在消息队列管理控制台创建
         factoryInfo.setFactoryProperty(factoryInfo.SecretKey, "");//SecretKey 身份验证，在消息队列管理控制台创建
         //create transaction producer       
         ONSFactory onsfactory = new ONSFactory();
         LocalTransactionChecker myChecker = new MyLocalTransactionChecker();
         TransactionProducer pProducer = onsfactory.getInstance().createTransactionProducer(factoryInfo,ref myChecker);
         // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可，启动之后可以多线程并发发送消息
         pProducer.start();
             Message msg = new Message(
             //Message Topic
             factoryInfo.getPublishTopics(),
             //Message Tag
             "TagA",
             //Message Body
             factoryInfo.getMessageContent()
         );
         // 设置代表消息的业务关键属性，请尽可能全局唯一
         // 以方便您在无法正常收到消息情况下，可通过控制台查询消息并补发。
         // 注意：不设置也不会影响消息正常收发
         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() );
         }
         // 在应用退出前，必须销毁 Producer 对象，否则会导致内存泄露等问题；
         // shutdown 之后不能重新 start 此 producer
         pProducer.shutdown();
     }
 }
 }

2. 提交事务消息状态
   当本地事务执行完成（执行成功或执行失败），需要通知服务器当前消息的事务状态。 通知方式有以下两种：
   • 执行本地事务完成后提交。
   • 执行本地事务一直没提交状态，等待服务器回查消息的事务状态。
     事务状态有以下三种：
   • TransactionStatus.CommitTransaction 提交事务，允许订阅方消费该消息。
   • TransactionStatus.RollbackTransaction 回滚事务，消息将被丢弃不允许消费。
   • TransactionStatus.Unknow 无法判断状态，期待 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.getUserProperty("VincentNoUser"));
                // 消息 ID(有可能消息体一样，但消息 ID 不一样。当前消息 ID 在控制台无法查询)
                string msgId = value.getMsgID();
                // 消息体内容进行 crc32, 也可以使用其它的如 MD5
                // 消息 ID 和 crc32id 主要是用来防止消息重复
                // 如果业务本身是幂等的， 可以忽略，否则需要利用 msgId 或 crc32Id 来做幂等
                // 如果要求消息绝对不重复，推荐做法是对消息体 body 使用 crc32或 MD5 来防止重复消息 
                TransactionStatus transactionStatus = TransactionStatus.Unknow;
                try {
                    boolean isCommit = 本地事务执行结果;
                    if (isCommit) {
                        // 本地事务成功、提交消息
                        transactionStatus = TransactionStatus.CommitTransaction;
                    } else {
                        // 本地事务失败、回滚消息
                        transactionStatus = TransactionStatus.RollbackTransaction;
                    }
                } catch (Exception e) {
                    //exception handle
                }
                return transactionStatus；
        }
        }

事务回查机制说明

• 发送事务消息为什么必须要实现 Check 机制？
  当步骤 1 半消息发送完成，但本地事务返回状态为 TransactionStatus.Unknow 时，亦或是应用退出导致本地事务未提交任何状态时，从 Broker 的角度看，这条半状态的消息的状态是未知的，因此 Broker 会定期要求发送方能 Check 该半状态消息，并上报其最终状态。
• Check 被回调时，业务逻辑都需要做些什么？
  事务消息的 Check 方法里面，应该写一些检查事务一致性的逻辑。 消息队列 RocketMQ 发送事务消息时需要实现 LocalTransactionChecker 接口，用来处理 Broker 主动发起的本地事务状态回查请求；因此在事务消息的 Check 方法中，需要完成两件事情：
  （1）检查该半消息对应的本地事务的状态（committed or rollback）。
  （2）向 Broker 提交该半消息本地事务的状态。
• 本地事务的不同状态对半消息的影响？
  • TransactionStatus.CommitTransaction 提交事务，允许订阅方消费该消息。
  • TransactionStatus.RollbackTransaction 回滚事务，消息将被丢弃不允许消费。
  • TransactionStatus.Unknow 无法判断状态，期待 Broker 向发送方再次询问该消息对应的本地事务的状态。
    具体代码详见 MyLocalTransactionChecker 的实现。