Rocket MQ 4.3.0分散式事務訊息初析
阿新 • • 發佈：2018-11-11
前言
從4.3.0版本開始支援事務訊息，這是一個令人振奮的訊息，線上目前4.2.0，在正式投產使用之前先進性簡單分析。幫助使用者實現類似 X/Open XA 的分佈事務功能，通過 MQ 事務訊息能達到分散式事務的最終一致。
基礎概念
- 事務訊息：MQ 提供類似 X/Open XA 的分佈事務功能，通過 MQ 事務訊息能達到分散式事務的最終一致。
- 半訊息：暫不能投遞的訊息，傳送方已經將訊息成功傳送到了 MQ 服務端，但是服務端未收到生產者對該訊息的二次確認，此時該訊息被標記成“暫不能投遞”狀態，處於該種狀態下的訊息即半訊息。
- 訊息回查：由於網路閃斷、生產者應用重啟等原因，導致某條事務訊息的二次確認丟失，MQ 服務端通過掃描發現某條訊息長期處於“半訊息”時，需要主動向訊息生產者詢問該訊息的最終狀態（Commit 或是 Rollback），該過程即訊息回查。
實現原理
其中：
- 傳送方向 MQ 服務端傳送訊息。
- MQ Server 將訊息持久化成功之後，向傳送方 ACK 確認訊息已經發送成功，此時訊息為半訊息。
- 傳送方開始執行本地事務邏輯。
- 傳送方根據本地事務執行結果向 MQ Server 提交二次確認（Commit 或是 Rollback），MQ Server 收到 Commit 狀態則將半訊息標記為可投遞，訂閱方最終將收到該訊息；MQ Server 收到 Rollback 狀態則刪除半訊息，訂閱方將不會接受該訊息。
- 在斷網或者是應用重啟的特殊情況下，上述步驟4提交的二次確認最終未到達 MQ Server，經過固定時間後 MQ Server 將對該訊息發起訊息回查。
- 傳送方收到訊息回查後，需要檢查對應訊息的本地事務執行的最終結果。
- 傳送方根據檢查得到的本地事務的最終狀態再次提交二次確認，MQ Server 仍按照步驟4對半訊息進行操作。
事務訊息傳送對應步驟1、2、3、4，事務訊息回查對應步驟5、6、7。
code實戰
傳送事務訊息
- 傳送半訊息及執行本地事務。
public class TransactionProducerClient { private final static Logger log = ClientLogger.getLog(); // 您需要設定自己的日誌，便於排查問題 public static void main(String[] args) throws InterruptedException { final BusinessService businessService = new BusinessService(); // 本地業務 Service Properties properties = new Properties(); // 您在控制檯建立的 Producer ID 注意：事務訊息的 Producer ID 不能與其他型別訊息的 Producer ID 共用 properties.put(PropertyKeyConst.ProducerId, ""); // 阿里雲身份驗證，在阿里雲伺服器管理控制檯建立 properties.put(PropertyKeyConst.AccessKey, ""); // 阿里雲身份驗證，在阿里雲伺服器管理控制檯建立 properties.put(PropertyKeyConst.SecretKey, ""); // 設定 TCP 接入域名（此處以公共雲生產環境為例） properties.put(PropertyKeyConst.ONSAddr, "http://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()); try { 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); } catch (Exception e) { // 訊息傳送失敗，需要進行重試處理，可重新發送這條訊息或持久化這條資料進行補償處理 System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic()); e.printStackTrace(); } // demo example 防止程序退出(實際使用不需要這樣) TimeUnit.MILLISECONDS.sleep(Integer.MAX_VALUE); } }
- 提交事務訊息狀態
當本地事務執行完成（執行成功或執行失敗），需要通知伺服器當前訊息的事務狀態。 通知方式有以下兩種：
執行本地事務完成後提交
執行本地事務一直沒提交狀態，等待伺服器回查訊息的事務狀態
事務狀態有以下三種：
TransactionStatus.CommitTransaction 提交事務，允許訂閱方消費該訊息。
TransactionStatus.RollbackTransaction 回滾事務，訊息將被丟棄不允許消費。
TransactionStatus.Unknow 無法判斷狀態，期待 MQ Broker 向傳送方再次詢問該訊息對應的本地事務的狀態。
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();
}
}事務回查機制說明
-
傳送事務訊息為什麼必須要實現回查 Check 機制？
當步驟（1）中 Half 訊息傳送完成，但本地事務返回狀態為 TransactionStatus.Unknow，或者應用退出導致本地事務未提交任何狀態時，從 MQ Broker 的角度看，這條 Half 狀態的訊息的狀態是未知的。 因此 MQ Broker 會定期要求傳送方能 Check 該 Half 狀態訊息，並上報其最終狀態。
-
Check 被回撥時，業務邏輯都需要做些什麼？
MQ 事務訊息的 check 方法裡面，應該寫一些檢查事務一致性的邏輯。 MQ 傳送事務訊息時需要實現 LocalTransactionChecker 介面，用來處理 MQ Broker 主動發起的本地事務狀態回查請求；因此在事務訊息的 Check 方法中，需要完成兩件事情：
(1) 檢查該 Half 訊息對應的本地事務的狀態（commited or rollback）；
(2) 向 MQ Broker 提交該 Half 訊息本地事務的狀態。