Java SDK如何进行事务消息收发

TCP 接入点域名，请 前往查看。
发送事务消息包含以下两个步骤：

1. 
   发送半消息及执行本地事务package com.alibaba.webx.TryHsf.app1;
3. import com.aliyun.openservices.ons.api.Message;
4. import com.aliyun.openservices.ons.api.PropertyKeyConst;
5. import com.aliyun.openservices.ons.api.SendResult;
6. import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
7. import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
8. import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
9. import java.util.Properties;
10. import java.util.concurrent.TimeUnit;
12. public class TransactionProducerClient {
13. private final static Logger log = ClientLogger.getLog(); // 用户需要设置自己的log, 记录日志便于排查问题
15. public static void main(String[] args) throws InterruptedException {
16.      final BusinessService businessService = new BusinessService(); // 本地业务Service
17.      Properties properties = new Properties();
18.      // 您在控制台创建的Producer ID。注意：事务消息的Producer ID不能与其他类型消息的Producer ID共用
19.      properties.put(PropertyKeyConst.ProducerId, "");
20.      // 阿里云身份验证，在阿里云服务器管理控制台创建
21.      properties.put(PropertyKeyConst.AccessKey, "");
22.      // 阿里云身份验证，在阿里云服务器管理控制台创建
23.      properties.put(PropertyKeyConst.SecretKey, "");
24.      // 设置 TCP 接入域名（此处以公共云生产环境为例）
25.      properties.put(PropertyKeyConst.ONSAddr,
26.        "http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal");
28.      TransactionProducer producer = ONSFactory.createTransactionProducer(properties,
29.              new LocalTransactionCheckerImpl());
30.      producer.start();
31.      Message msg = new Message("Topic", "TagA", "Hello MQ transaction===".getBytes());
32.      // 输入您在控制台创建的Topic
33.      SendResult sendResult = producer.send(msg, new LocalTransactionExecuter() {
34.          @Override
35.          public TransactionStatus execute(Message msg, Object arg) {
36.              // 消息ID(有可能消息体一样，但消息ID不一样, 当前消息ID在控制台无法查询)
37.              String msgId = msg.getMsgID();
38.              // 消息体内容进行crc32, 也可以使用其它的如MD5
39.              long crc32Id = HashUtil.crc32Code(msg.getBody());
40.              // 消息ID和crc32id主要是用来防止消息重复
41.              // 如果业务本身是幂等的, 可以忽略, 否则需要利用msgId或crc32Id来做幂等
42.              // 如果要求消息绝对不重复, 推荐做法是对消息体body使用crc32或md5来防止重复消息
43.              Object businessServiceArgs = new Object();
44.              TransactionStatus transactionStatus = TransactionStatus.Unknow;
45.              try {
46.                  boolean isCommit =
47.                          businessService.execbusinessService(businessServiceArgs);
48.                  if (isCommit) {
49.                      // 本地事务成功、提交消息
50.                      transactionStatus = TransactionStatus.CommitTransaction;
51.                  } else {
52.                      // 本地事务失败、回滚消息
53.                      transactionStatus = TransactionStatus.RollbackTransaction;
54.                  }
55.              } catch (Exception e) {
56.                  log.error("Message Id:{}", msgId, e);
57.              }
58.              System.out.println(msg.getMsgID());
59.              log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
60.              return transactionStatus;
61.          }
62.      }, null);
63.      // demo example 防止进程退出(实际使用不需要这样)
64.      TimeUnit.MILLISECONDS.sleep(Integer.MAX_VALUE);
65. }
66. }

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

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

2. public class LocalTransactionCheckerImpl implements LocalTransactionChecker {
3.     private final static Logger log = ClientLogger.getLog();
4.     final  BusinessService businessService = new BusinessService();
6.     @Override
7.     public TransactionStatus check(Message msg) {
8.         //消息ID(有可能消息体一样，但消息ID不一样, 当前消息属于Half 消息，所以消息ID在控制台无法查询)
9.         String msgId = msg.getMsgID();
10.         //消息体内容进行crc32, 也可以使用其它的方法如MD5
11.         long crc32Id = HashUtil.crc32Code(msg.getBody());
12.         //消息ID、消息本 crc32Id主要是用来防止消息重复
13.         //如果业务本身是幂等的, 可以忽略, 否则需要利用msgId或crc32Id来做幂等
14.         //如果要求消息绝对不重复, 推荐做法是对消息体使用crc32或md5来防止重复消息.
15.         //业务自己的参数对象, 这里只是一个示例, 实际需要用户根据情况来处理
16.         Object businessServiceArgs = new Object();
17.         TransactionStatus transactionStatus = TransactionStatus.Unknow;
18.         try {
19.             boolean isCommit = businessService.checkbusinessService(businessServiceArgs);
20.             if (isCommit) {
21.                 //本地事务已成功、提交消息
22.                 transactionStatus = TransactionStatus.CommitTransaction;
23.             } else {
24.                 //本地事务已失败、回滚消息
25.                 transactionStatus = TransactionStatus.RollbackTransaction;
26.             }
27.         } catch (Exception e) {
28.             log.error("Message Id:{}", msgId, e);
29.         }
30.         log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
31.         return transactionStatus;
32.     }
33. }

1. import java.util.zip.CRC32;
2. public class HashUtil {
3.     public static long crc32Code(byte[] bytes) {
4.         CRC32 crc32 = new CRC32();
5.         crc32.update(bytes);
6.         return crc32.getValue();
7.     }
8. }

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