Apache Seata 是一款高性能、簡單易用的分布式事務(wù)中間件，它包含 AT、TCC、SAGA 和 XA 四種模式。

在最近發(fā)布的新版本中，Apache Seata 引入了 RocketMQ 中間件，并且跟 RocketMQ 的事務(wù)消息配合使用。今天我們來聊一聊這個話題。

Seata 兩階段提交

Seata 的分布式事務(wù)解決方案使用兩階段提交，以最流行的 TCC 模式為例，Seata 中設(shè)計了 3 個主要的角色：

• TM：管理全局事務(wù)，包括開啟全局事務(wù)，提交/回滾全局事務(wù)；
• RM：管理分支事務(wù)；
• TC: 事務(wù)協(xié)調(diào)器，管理全局事務(wù)和分支事務(wù)的狀態(tài)。

下圖是一個電商場景的分布式事務(wù)場景，包括訂單、庫存和賬戶三個服務(wù)。

圖片

TM 開啟全局事務(wù)后，在第一階段，訂單服務(wù)、庫存服務(wù)和賬戶服務(wù)分別作為一個 RM 向 TC 注冊分支事務(wù)，并且向 TC 上報分支事務(wù)執(zhí)行狀態(tài)。如果三個分支事務(wù)都執(zhí)行成功，TM 會給 TC 發(fā)送 commit 執(zhí)行，否則發(fā)送 rollback 指令。在第二階段，TC 會根據(jù) TM 的指令，決定分支事務(wù) commit 或者 rollback。

RocketMQ 事務(wù)消息

作為分布式事務(wù)的解決方案，RocketMQ 的事務(wù)消息也可以供我們選擇。RocketMQ 的事務(wù)消息，主要是保證了本地方法執(zhí)行和消息發(fā)送在一個分布式事務(wù)中，要不全部成功，要不全部失敗。見下圖：

圖片

這里給出一個賬戶服務(wù)和庫存服務(wù)在一個分布式事務(wù)中的例子，本地方法執(zhí)行扣減賬戶金額，RocketMQ 消費(fèi)者消費(fèi)賬戶服務(wù)發(fā)送的消息后執(zhí)行扣減庫存。RocketMQ 通過發(fā)送 half 消息來實現(xiàn)，下面詳細(xì)說明一下：

1. 賬戶服務(wù)向 Broker 發(fā)送一條 half 消息；
2. half 消息發(fā)送成功后，賬戶服務(wù)執(zhí)行本地事務(wù)；
3. 如果賬戶服務(wù)執(zhí)行本地事務(wù)成功，則向 Broker 發(fā)送 commit 請求，否則發(fā)送 rollback 請求；
4. 如果 Broker 收到的是 rollback 請求，則刪除保存的 half 消息；
5. 如果 Broker 收到的是 commit 請求，則把 half 消息投遞到真實的隊列等待庫存服務(wù)來拉取，然后刪除保存的 half 消息；
6. 如果 Broker 沒有收到請求，則會發(fā)送請求到 Producer 查詢本地事務(wù)狀態(tài)，然后根據(jù) Producer 返回的本地狀態(tài)做 commit/rollback 相關(guān)處理。

Seata 的改進(jìn)

Seata 的改進(jìn)主要在 prepare 階段。Seata 提供了一個 SeataMQProducer 類，把 RocketMQ 中 TransactionListener 的方法加入到全局事務(wù)。見下面代碼：

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 發(fā)送 half 消息，執(zhí)行本地事務(wù)，如果執(zhí)行成功，則強(qiáng)行把 LocalTransactionState 改回 UNKNOW，等待 TC 發(fā)送指令，決定是 commit 或 rollback。如果執(zhí)行失敗，返回 ROLLBACK_MESSAGE，TC 下發(fā)指令，回滾全局事務(wù)。

需要注意的是，Broker 收到 half 消息后，如果沒有收到 commit/rollback 消息，則會回查 Producer 本地事務(wù)狀態(tài)，也就是上面代碼中的 checkLocalTransaction。這里不是檢查分支事務(wù)狀態(tài)，而是檢查全局事務(wù)狀態(tài)，使用 XID 去查詢。

集成 RocketMQ 之后，Seata 的分布式事務(wù)調(diào)用流程如下圖，這里以 訂單服務(wù)、庫存服務(wù)兩個服務(wù)為例：

圖片

總結(jié)

Apache Seata 引入 RocketMQ 后，支持的分布式事務(wù)場景更加豐富，MQ 的異步特性也讓事務(wù)模型整體性能提升。希望本文對你理解 Seata 中 RocketMQ 的使用有所幫助。