48      阿裏雲  技術社區[雲棲]

發送事務消息__Java SDK_TCP 接入(專業)_消息隊列 MQ-阿裏雲

目前支持的域包括公網測試、華東1、華北2、華東2、華南1。

MQ事務消息交互流程如下：

發送事務消息包含以下兩個步驟：

1. 發送半消息及執行本地事務

   package com.alibaba.webx.TryHsf.app1;
   import com.aliyun.openservices.ons.api.Message;
   import com.aliyun.openservices.ons.api.PropertyKeyConst;
   import com.aliyun.openservices.ons.api.SendResult;
   import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
   import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
   import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
   import java.util.Properties;
   import java.util.concurrent.TimeUnit;
   public class TransactionProducerClient {
    private final static Logger log = ClientLogger.getLog(); // 用戶需要設置自己的log, 記錄日誌便於排查問題
    public static void main(String[] args) throws InterruptedException {
        final BusinessService businessService = new BusinessService(); // 本地業務Service
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ProducerId, ""); // 您在控製台創建的Producer ID
        properties.put(PropertyKeyConst.AccessKey, ""); // 阿裏雲身份驗證，在阿裏雲服務器管理控製台創建
        properties.put(PropertyKeyConst.SecretKey, ""); // 阿裏雲身份驗證，在阿裏雲服務器管理控製台創建
        //公有雲生產環境：https://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal
        //公有雲公測環境：https://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet
        //杭州金融雲環境：https://jbponsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal
        //杭州深圳雲環境：https://mq4finance-sz.addr.aliyun.com:8080/rocketmq/nsaddr4client-internal
        properties.put(PropertyKeyConst.ONSAddr,
           "https://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal");//此處以公有雲生產環境為例
        TransactionProducer producer = ONSFactory.createTransactionProducer(properties,
                new LocalTransactionCheckerImpl());
        producer.start();
        Message msg = new Message("Topic", "TagA", "Hello MQ transaction===".getBytes());
        // 輸入您在控製台創建的Topic
        SendResult sendResult = producer.send(msg, new LocalTransactionExecuter() {
            @Override
            public TransactionStatus execute(Message msg, Object arg) {
                // 消息ID(有可能消息體一樣，但消息ID不一樣, 當前消息ID在控製台無法查詢)
                String msgId = msg.getMsgID();
                // 消息體內容進行crc32, 也可以使用其它的如MD5
                long crc32Id = HashUtil.crc32Code(msg.getBody());
                // 消息ID和crc32id主要是用來防止消息重複
                // 如果業務本身是冪等的, 可以忽略, 否則需要利用msgId或crc32Id來做冪等
                // 如果要求消息絕對不重複, 推薦做法是對消息體body使用crc32或md5來防止重複消息 
                Object businessServiceArgs = new Object();
                TransactionStatus transactionStatus = TransactionStatus.Unknow;
                try {
                    boolean isCommit =
                            businessService.execbusinessService(businessServiceArgs);
                    if (isCommit) {
                        // 本地事務成功、提交消息
                        transactionStatus = TransactionStatus.CommitTransaction;
                    } else {
                        // 本地事務失敗、回滾消息
                        transactionStatus = TransactionStatus.RollbackTransaction;
                    }
                } catch (Exception e) {
                    log.error("Message Id:{}", msgId, e);
                }
                System.out.println(msg.getMsgID());
                log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
                return transactionStatus;
            }
        }, null);
        // demo example 防止進程退出(實際使用不需要這樣)
        TimeUnit.MILLISECONDS.sleep(Integer.MAX_VALUE);
    }
   }

2. 提交事務消息狀態

   當本地事務執行完成（執行成功或執行失敗），需要通知服務器當前消息的事務狀態。通知方式有以下兩種：

   • 執行本地事務完成後提交
   • 執行本地事務一直沒提交狀態，等待服務器回查消息的事務狀態

   事務狀態有以下三種：

   • TransactionStatus.CommitTransaction 提交事務，允許訂閱方消費該消息
   • TransactionStatus.RollbackTransaction 回滾事務，消息將被丟棄不允許消費
   • TransactionStatus.Unknow 無法判斷狀態，期待MQ Broker向發送方再次詢問該消息對應的本地事務的狀態

   import com.alibaba.rocketmq.client.producer.LocalTransactionState;
   public class LocalTransactionCheckerImpl implements LocalTransactionChecker {
   private final static Logger log = ClientLogger.getLog();
   final  BusinessService businessService = new BusinessService();
   @Override
   public TransactionStatus check(Message msg) {
       //消息ID(有可能消息體一樣，但消息ID不一樣, 當前消息屬於Half 消息，所以消息ID在控製台無法查詢)
       String msgId = msg.getMsgID();
       //消息體內容進行crc32, 也可以使用其它的方法如MD5
       long crc32Id = HashUtil.crc32Code(msg.getBody());
       //消息ID、消息本 crc32Id主要是用來防止消息重複
       //如果業務本身是冪等的, 可以忽略, 否則需要利用msgId或crc32Id來做冪等
       //如果要求消息絕對不重複, 推薦做法是對消息體使用crc32或md5來防止重複消息.
       //業務自己的參數對象, 這裏隻是一個示例, 實際需要用戶根據情況來處理
       Object businessServiceArgs = new Object();
       TransactionStatus transactionStatus = TransactionStatus.Unknow;
       try {
           boolean isCommit = businessService.checkbusinessService(businessServiceArgs);
           if (isCommit) {
               //本地事務已成功、提交消息
               transactionStatus = TransactionStatus.CommitTransaction;
           } else {
               //本地事務已失敗、回滾消息
               transactionStatus = TransactionStatus.RollbackTransaction;
           }
       } catch (Exception e) {
           log.error("Message Id:{}", msgId, e);
       }
       log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
       return transactionStatus;
   }
   }

工具類

import java.util.zip.CRC32;
public class HashUtil {
    public static long crc32Code(byte[] bytes) {
        CRC32 crc32 = new CRC32();
        crc32.update(bytes);
        return crc32.getValue();
    }
}