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 消息來實現，下面詳細說明一下：

1. 賬戶服務向 Broker 發送一條 half 消息；
2. half 消息發送成功后，賬戶服務執行本地事務；
3. 如果賬戶服務執行本地事務成功，則向 Broker 發送 commit 請求，否則發送 rollback 請求；
4. 如果 Broker 收到的是 rollback 請求，則刪除保存的 half 消息；
5. 如果 Broker 收到的是 commit 請求，則把 half 消息投遞到真實的隊列等待庫存服務來拉取，然后刪除保存的 half 消息；
6. 如果 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 的使用有所幫助。