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

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

Seata 兩階段提交

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

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

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

圖片

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

RocketMQ 事務(wù)消息

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

圖片

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

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

Seata 的改進(jìn)

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

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 消息后，如果沒(méi)有收到 commit/rollback 消息，則會(huì)回查 Producer 本地事務(wù)狀態(tài)，也就是上面代碼中的 checkLocalTransaction。這里不是檢查分支事務(wù)狀態(tài)，而是檢查全局事務(wù)狀態(tài)，使用 XID 去查詢。

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

圖片

總結(jié)

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