发送事务消息

目前支持的域包括公网、华东1、华北2、华东2、华南1。
MQ 事务消息交互流程如下。

发送事务消息包含两个步骤：
1.发送半消息（Half Message）及执行本地事务

2. #include "ONSFactory.h"
3. #include "ONSClientException.h"
4. using namespace ons;
6.     class MyLocalTransactionExecuter : LocalTransactionExecuter
7.     {
8.         MyLocalTransactionExecuter()
9.         {
10.         }
12.         ~MyLocalTransactionExecuter()
13.         {
14.         }
15.         virtual TransactionStatus execute(Message &value)
16.         {
17.                 // 消息ID(有可能消息体一样，但消息id不一样, 当前消息ID在console控制不可能查询)
18.                 string msgId = value.getMsgID();
19.                 // 消息体内容进行crc32, 也可以使用其它的如MD5
20.                 // 消息ID和crc32id主要是用来防止消息重复
21.                 // 如果业务本身是幂等的, 可以忽略, 否则需要利用msgId或crc32Id来做幂等
22.                 // 如果要求消息绝对不重复, 推荐做法是对消息体body使用crc32或md5来防止重复消息.
23.                 TransactionStatus transactionStatus = Unknow;
24.                 try {
25.                     boolean isCommit = 本地事务执行结果;
26.                     if (isCommit) {
27.                         // 本地事务成功、提交消息
28.                         transactionStatus = CommitTransaction;
29.                     } else {
30.                         // 本地事务失败、回滚消息
31.                         transactionStatus = RollbackTransaction;
32.                     }
33.                 } catch (...) {
34.                     //exception handle
35.                 }
36.                 return transactionStatus；
37.         }
38.     }
40.     int main(int argc, char* argv[])
41.     {
42.         ONSFactoryProperty factoryInfo;
43.         factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");//您在控制台创建的Producer ID
44.         factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );//输入您在控制台创建的Topic
45.         factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");//msg content
46.         factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "xxxxxxxxx");//阿里云身份验证，在阿里云服务器管理控制台创建
47.         factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "xxxxxxxxxxxxxxxxxxxx" );//阿里云身份验证，在阿里云服务器管理控制台创建
49.         //创建producer，MQ不负责pChecker的释放，需要业务方自行释放资源
50.         MyLocalTransactionChecker *pChecker = new MyLocalTransactionChecker();
51.         g_producer = ONSFactory::getInstance()->createTransactionProducer(factoryInfo,pChecker);
53.         //在发送消息前，必须调用start方法来启动Producer，只需调用一次即可;
54.         pProducer->start();
56.         Message msg(
57.             //Message Topic
58.             factoryInfo.getPublishTopics(),
59.             //Message Tag,可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在ONS服务器过滤      
60.             "TagA",
61.             //Message Body,不能为空，MQ不做任何干预，需要Producer与Consumer协商好一致的序列化和反序列化方式
62.             factoryInfo.getMessageContent()
63.         );
65.         // 设置代表消息的业务关键属性，请尽可能全局唯一。
66.         // 以方便您在无法正常收到消息情况下，可通过MQ Console查询消息并补发。
67.         // 注意：不设置也不会影响消息正常收发
68.         msg.setKey("ORDERID_100");
70.         //发送消息，只要不抛出异常，就代表发送成功    
71.         try
72.         {
73.             //MQ不负责pExecuter的释放，需要业务方自行释放资源
74.             MyLocalTransactionExecuter pExecuter = new MyLocalTransactionExecuter();
75.             SendResultONS sendResult = pProducer->send(msg,pExecuter);
76.         }
77.         catch(ONSClientException & e)
78.         {
79.             //自定义处理exception的细节
80.         }
81.         // 在应用退出前，必须销毁Producer对象，否则会导致内存泄露等问题
82.         pProducer->shutdown();
84.         return 0;
86.     }

2.提交事务消息状态
当本地事务执行完成（执行成功或执行失败），需要通知服务器当前消息的事务状态。通知方式有以下两种：

• 执行本地事务完成后提交；
• 执行本地事务一直没提交状态，等待服务器回查消息的事务状态。

事务状态有以下三种：

• TransactionStatus.CommitTransaction 提交事务，允许订阅方消费该消息；
• TransactionStatus.RollbackTransaction 回滚事务，消息将被丢弃不允许消费；
• 
  TransactionStatus.Unknow 无法判断状态，期待 MQ Broker 向发送方再次询问该消息对应的本地事务的状态。
  1. class MyLocalTransactionChecker : LocalTransactionChecker
  2. {
  3.      MyLocalTransactionChecker()
  4.      {
  5.      }
  7.      ~MyLocalTransactionChecker()
  8.      {
  9.      }
  11.      virtual TransactionStatus check(Message &value)
  12.      {
  13.          // 消息ID(有可能消息体一样，但消息id不一样, 当前消息ID在console控制不可能查询)
  14.          string msgId = value.getMsgID();
  15.          // 消息体内容进行crc32, 也可以使用其它的如MD5
  16.          // 消息ID和crc32id主要是用来防止消息重复
  17.          // 如果业务本身是幂等的, 可以忽略, 否则需要利用msgId或crc32Id来做幂等
  18.          // 如果要求消息绝对不重复, 推荐做法是对消息体body使用crc32或md5来防止重复消息.
  19.          TransactionStatus transactionStatus = Unknow;
  20.          try {
  21.              boolean isCommit = 本地事务执行结果;
  22.              if (isCommit) {
  23.                  // 本地事务成功、提交消息
  24.                  transactionStatus = CommitTransaction;
  25.              } else {
  26.                  // 本地事务失败、回滚消息
  27.                  transactionStatus = RollbackTransaction;
  28.              }
  29.          } catch(...) {
  30.              //exception error
  31.          }
  32.          return transactionStatus；
  33.      }
  34. }

事务回查机制说明
1、发送事务消息为什么必须要实现 check 机制？
当步骤1发送半消息完成，但本地事务返回状态为 TransactionStatus.Unknow 时，亦或是应用退出导致本地事务未提交任何状态时，从 MQ Broker 的角度看，这条半状态的消息的状态是未知的，因此 MQ Broker 会定期要求发送方能 check 该半状态消息，并上报其最终状态。
2、Check 被回调时，业务逻辑都需要做些什么？
MQ 事务消息的 check 方法里面，应该写一些检查事务一致性的逻辑。MQ 发送事务消息时需要实现 LocalTransactionChecker 接口，用来处理 MQ Broker主动发起的本地事务状态回查请求；因此在事务消息的 check 方法中，需要完成两件事情：
(1) 检查该半消息对应的本地事务的状态（commited or rollback）；
(2) 向 MQ Broker 提交该半消息本地事务的状态。
3、本地事务的不同状态对Half消息的影响？

• 
  TransactionStatus.CommitTransaction 提交事务，允许订阅方消费该消息。
• 
  TransactionStatus.RollbackTransaction 回滚事务，消息将被丢弃不允许消费。
• 
  TransactionStatus.Unknow 无法判断状态，期待 MQ Broker 向发送方再次询问该消息对应的本地事务的状态。
  具体代码详见 MyLocalTransactionChecker 的实现。