Apache Seata 新版本集成了 RocketMQ 事務消息
Apache Seata 是一款高性能、簡單易用的分布式事務中間件，它包含 AT、TCC、SAGA 和 XA 四種模式。
在最近發布的新版本中，Apache Seata 引入了 RocketMQ 中間件，并且跟 RocketMQ 的事務消息配合使用。今天我們來聊一聊這個話題。
Seata 兩階段提交
Seata 的分布式事務解決方案使用兩階段提交，以最流行的 TCC 模式為例，Seata 中設計了 3 個主要的角色：
- TM：管理全局事務，包括開啟全局事務，提交/回滾全局事務；
- RM：管理分支事務；
- TC: 事務協調器，管理全局事務和分支事務的狀態。
下圖是一個電商場景的分布式事務場景，包括訂單、庫存和賬戶三個服務。
圖片
TM 開啟全局事務后，在第一階段，訂單服務、庫存服務和賬戶服務分別作為一個 RM 向 TC 注冊分支事務，并且向 TC 上報分支事務執行狀態。如果三個分支事務都執行成功，TM 會給 TC 發送 commit 執行，否則發送 rollback 指令。在第二階段，TC 會根據 TM 的指令，決定分支事務 commit 或者 rollback。
RocketMQ 事務消息
作為分布式事務的解決方案，RocketMQ 的事務消息也可以供我們選擇。RocketMQ 的事務消息，主要是保證了本地方法執行和消息發送在一個分布式事務中，要不全部成功，要不全部失敗。見下圖：
圖片
這里給出一個賬戶服務和庫存服務在一個分布式事務中的例子，本地方法執行扣減賬戶金額，RocketMQ 消費者消費賬戶服務發送的消息后執行扣減庫存。RocketMQ 通過發送 half 消息來實現，下面詳細說明一下：
- 賬戶服務向 Broker 發送一條 half 消息；
- half 消息發送成功后，賬戶服務執行本地事務；
- 如果賬戶服務執行本地事務成功，則向 Broker 發送 commit 請求，否則發送 rollback 請求；
- 如果 Broker 收到的是 rollback 請求，則刪除保存的 half 消息；
- 如果 Broker 收到的是 commit 請求，則把 half 消息投遞到真實的隊列等待庫存服務來拉取，然后刪除保存的 half 消息；
- 如果 Broker 沒有收到請求，則會發送請求到 Producer 查詢本地事務狀態，然后根據 Producer 返回的本地狀態做 commit/rollback 相關處理。
Seata 的改進
Seata 的改進主要在 prepare 階段。Seata 提供了一個 SeataMQProducer 類，把 RocketMQ 中 TransactionListener 的方法加入到全局事務。見下面代碼：
SeataMQProducer(final String namespace, final String producerGroup, RPCHook rpcHook) {
super(namespace, producerGroup, rpcHook);
this.transactionListener = new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
return LocalTransactionState.UNKNOW;
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
String xid = msg.getProperty(PROPERTY_SEATA_XID);
if (StringUtils.isBlank(xid)) {
LOGGER.error("msg has no xid, msgTransactionId: {}, msg will be rollback", msg.getTransactionId());
return LocalTransactionState.ROLLBACK_MESSAGE;
}
GlobalStatus globalStatus = DefaultResourceManager.get().getGlobalStatus(SeataMQProducerFactory.ROCKET_BRANCH_TYPE, xid);
if (COMMIT_STATUSES.contains(globalStatus)) {
return LocalTransactionState.COMMIT_MESSAGE;
} else if (ROLLBACK_STATUSES.contains(globalStatus) || GlobalStatus.isOnePhaseTimeout(globalStatus)) {
return LocalTransactionState.ROLLBACK_MESSAGE;
} else if (GlobalStatus.Finished.equals(globalStatus)) {
LOGGER.error("global transaction finished, msg will be rollback, xid: {}", xid);
return LocalTransactionState.ROLLBACK_MESSAGE;
}
return LocalTransactionState.UNKNOW;
}
};
}在 prepare 階段，SeataMQProducer 向 RocketMQ Broker 發送 half 消息，執行本地事務，如果執行成功，則強行把 LocalTransactionState 改回 UNKNOW，等待 TC 發送指令，決定是 commit 或 rollback。如果執行失敗，返回 ROLLBACK_MESSAGE，TC 下發指令，回滾全局事務。
需要注意的是，Broker 收到 half 消息后，如果沒有收到 commit/rollback 消息，則會回查 Producer 本地事務狀態，也就是上面代碼中的 checkLocalTransaction。這里不是檢查分支事務狀態，而是檢查全局事務狀態，使用 XID 去查詢。
集成 RocketMQ 之后，Seata 的分布式事務調用流程如下圖，這里以 訂單服務、庫存服務兩個服務為例：
圖片
總結
Apache Seata 引入 RocketMQ 后，支持的分布式事務場景更加豐富，MQ 的異步特性也讓事務模型整體性能提升。希望本文對你理解 Seata 中 RocketMQ 的使用有所幫助。
