從上一篇文章「分散式事務中介軟體Seata的設計原理」講了下 Seata AT 模式的一些設計原理，從中也知道了 AT 模式的三個角色（RM、TM、TC），接下來我會更新 Seata 原始碼分析系列文章。今天就來分析 Seata AT 模式在啟動的時候都做了哪些操作。

客戶端啟動邏輯

TM 是負責整個全域性事務的管理器，因此一個全域性事務是由 TM 開啟的，TM 有個全域性管理類 GlobalTransaction，結構如下：

io.seata.tm.api.GlobalTransaction

public interface GlobalTransaction {

  void begin() throws TransactionException;

  void begin(int timeout) throws TransactionException;

  void begin(int timeout, String name) throws TransactionException;

  void commit() throws TransactionException;

  void rollback() throws TransactionException;
  
  GlobalStatus getStatus() throws TransactionException;
  
  // ...
}
 

可以通過 GlobalTransactionContext 建立一個 GlobalTransaction，然後用 GlobalTransaction 進行全域性事務的開啟、提交、回滾等操作，因此我們直接用 API 方式使用 Seata AT 模式：

//init seata;
TMClient.init(applicationId, txServiceGroup);
RMClient.init(applicationId, txServiceGroup);
//trx
GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
try {
  tx.begin(60000, "testBiz");
  // 事務處理
  // ...
  tx.commit();
} catch (Exception exx) {
  tx.rollback();
  throw exx;
}
 

如果每次使用全域性事務都這樣寫，難免會造成程式碼冗餘，我們的專案都是基於 Spring 容器，這時我們可以利用 Spring AOP 的特性，用模板模式把這些冗餘程式碼封裝模版裡，參考 Mybatis-spring 也是做了這麼一件事情，那麼接下來我們來分析一下基於 Spring 的專案啟動 Seata 並註冊全域性事務時都做了哪些工作。

我們開啟一個全域性事務是在方法上加上 @GlobalTransactional註解，Seata 的 Spring 模組中，有個 GlobalTransactionScanner，它的繼承關係如下：

public class GlobalTransactionScanner extends AbstractAutoProxyCreator implements InitializingBean, ApplicationContextAware, DisposableBean {
  // ...
}
 

在基於 Spring 專案的啟動過程中，對該類會有如下初始化流程：

InitializingBean 的 afterPropertiesSet() 方法呼叫了 initClient() 方法：

io.seata.spring.annotation.GlobalTransactionScanner#initClient

TMClient.init(applicationId, txServiceGroup);
RMClient.init(applicationId, txServiceGroup);

對 TM 和 RM 做了初始化操作。

• TM 初始化

io.seata.tm.TMClient#init

public static void init(String applicationId, String transactionServiceGroup) {
  // 獲取 TmRpcClient 例項
  TmRpcClient tmRpcClient = TmRpcClient.getInstance(applicationId, transactionServiceGroup);
  // 初始化 TM Client
  tmRpcClient.init();
}

呼叫 TmRpcClient.getInstance() 方法會獲取一個 TM 客戶端例項，在獲取過程中，會建立 Netty 客戶端配置檔案物件，以及建立 messageExecutor 執行緒池，該執行緒池用於在處理各種與服務端的訊息互動，在建立 TmRpcClient 例項時，建立 ClientBootstrap，用於管理 Netty 服務的啟停，以及 ClientChannelManager，它是專門用於管理 Netty 客戶端物件池，Seata 的 Netty 部分配合使用了物件吃，後面在分析網路模組會講到。

io.seata.core.rpc.netty.AbstractRpcRemotingClient#init

public void init() {
  clientBootstrap.start();
  // 定時嘗試連線服務端
  timerExecutor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      clientChannelManager.reconnect(getTransactionServiceGroup());
    }
  }, SCHEDULE_INTERVAL_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.SECONDS);
  mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
                                                    MAX_MERGE_SEND_THREAD,
                                                    KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
                                                    new LinkedBlockingQueue<>(),
                                                    new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
  mergeSendExecutorService.submit(new MergedSendRunnable());
  super.init();
}

呼叫 TM 客戶端 init() 方法，最終會啟動 netty 客戶端（此時還未真正啟動，在物件池被呼叫時才會被真正啟動）；開啟一個定時任務，定時重新發送 RegisterTMRequest（RM 客戶端會發送 RegisterRMRequest）請求嘗試連線服務端，具體邏輯是在 NettyClientChannelManager 中的 channels 中快取了客戶端 channel，如果此時 channels 不存在獲取已過期，那麼就會嘗試連線服務端以重新獲取 channel 並將其快取到 channels 中；開啟一條單獨執行緒，用於處理非同步請求傳送，這裡用得很巧妙，之後在分析網路模組在具體對其進行分析。

io.seata.core.rpc.netty.AbstractRpcRemoting#init

public void init() {
  timerExecutor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      for (Map.Entry<Integer, MessageFuture> entry : futures.entrySet()) {
        if (entry.getValue().isTimeout()) {
          futures.remove(entry.getKey());
          entry.getValue().setResultMessage(null);
          if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("timeout clear future: {}", entry.getValue().getRequestMessage().getBody());
          }
        }
      }

      nowMills = System.currentTimeMillis();
    }
  }, TIMEOUT_CHECK_INTERNAL, TIMEOUT_CHECK_INTERNAL, TimeUnit.MILLISECONDS);
}

在 AbstractRpcRemoting 的 init 方法中，又是開啟了一個定時任務，該定時任務主要是用於定時清除 futures 已過期的 futrue，futures 是儲存傳送請求需要返回結果的 future 物件，該物件有個超時時間，過了超時時間就會自動拋異常，因此需要定時清除已過期的 future 物件。

• RM 初始化

io.seata.rm.RMClient#init

public static void init(String applicationId, String transactionServiceGroup) {
  RmRpcClient rmRpcClient = RmRpcClient.getInstance(applicationId, transactionServiceGroup);
  rmRpcClient.setResourceManager(DefaultResourceManager.get());
  rmRpcClient.setClientMessageListener(new RmMessageListener(DefaultRMHandler.get()));
  rmRpcClient.init();
}

RmRpcClient.getInstance 處理邏輯與 TM 大致相同；ResourceManager 是 RM 資源管理器，負責分支事務的註冊、提交、上報、以及回滾操作，以及全域性鎖的查詢操作，DefaultResourceManager 會持有當前所有的 RM 資源管理器，進行統一呼叫處理，而 get() 方法主要是載入當前的資源管理器，主要用了類似 SPI 的機制，進行靈活載入，如下圖，Seata 會掃描 META-INF/services/ 目錄下的配置類並進行動態載入。

ClientMessageListener 是 RM 訊息處理監聽器，用於負責處理從 TC 傳送過來的指令，並對分支進行分支提交、分支回滾，以及 undo log 檔案刪除操作；最後 init 方法跟 TM 邏輯也大體一致；DefaultRMHandler 封裝了 RM 分支事務的一些具體操作邏輯。

接下來再看看 wrapIfNecessary 方法究竟做了哪些操作。

io.seata.spring.annotation.GlobalTransactionScanner#wrapIfNecessary

protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
  // 判斷是否有開啟全域性事務
  if (disableGlobalTransaction) {
    return bean;
  }
  try {
    synchronized (PROXYED_SET) {
      if (PROXYED_SET.contains(beanName)) {
        return bean;
      }
      interceptor = null;
      //check TCC proxy
      if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {
        //TCC interceptor, proxy bean of sofa:reference/dubbo:reference, and LocalTCC
        interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));
      } else {
        Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
        Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);

        // 判斷 bean 中是否有 GlobalTransactional 和 GlobalLock 註解
        if (!existsAnnotation(new Class[]{serviceInterface})
            && !existsAnnotation(interfacesIfJdk)) {
          return bean;
        }

        if (interceptor == null) {
          // 建立代理類
          interceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
        }
      }

      LOGGER.info("Bean[{}] with name [{}] would use interceptor [{}]",
                  bean.getClass().getName(), beanName, interceptor.getClass().getName());
      if (!AopUtils.isAopProxy(bean)) {
        bean = super.wrapIfNecessary(bean, beanName, cacheKey);
      } else {
        AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
        // 執行包裝目標物件到代理物件  
        Advisor[] advisor = super.buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
        for (Advisor avr : advisor) {
          advised.addAdvisor(0, avr);
        }
      }
      PROXYED_SET.add(beanName);
      return bean;
    }
  } catch (Exception exx) {
    throw new RuntimeException(exx);
  }
}

GlobalTransactionScanner 繼承了 AbstractAutoProxyCreator，用於對 Spring AOP 支援，從程式碼中可看出，用GlobalTransactionalInterceptor 代替了被 GlobalTransactional 和 GlobalLock 註解的方法。

GlobalTransactionalInterceptor 實現了 MethodInterceptor：

io.seata.spring.annotation.GlobalTransactionalInterceptor#invoke

public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
  Class<?> targetClass = methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null;
  Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
  final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);

  final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, GlobalTransactional.class);
  final GlobalLock globalLockAnnotation = getAnnotation(method, GlobalLock.class);
  if (globalTransactionalAnnotation != null) {
    // 全域性事務註解
    return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
  } else if (globalLockAnnotation != null) {
    // 全域性鎖註解
    return handleGlobalLock(methodInvocation);
  } else {
    return methodInvocation.proceed();
  }
}

以上是代理方法執行的邏輯邏輯，其中 handleGlobalTransaction() 方法裡面呼叫了 TransactionalTemplate 模版：

io.seata.spring.annotation.GlobalTransactionalInterceptor#handleGlobalTransaction

private Object handleGlobalTransaction(final MethodInvocation methodInvocation,
                                       final GlobalTransactional globalTrxAnno) throws Throwable {
  try {
    return transactionalTemplate.execute(new TransactionalExecutor() {
      @Override
      public Object execute() throws Throwable {
        return methodInvocation.proceed();
      }
      @Override
      public TransactionInfo getTransactionInfo() {
        // ...
      }
    });
  } catch (TransactionalExecutor.ExecutionException e) {
    // ...
  }
}

handleGlobalTransaction() 方法執行了就是 TransactionalTemplate 模版類的 execute 方法：

io.seata.tm.api.TransactionalTemplate#execute

public Object execute(TransactionalExecutor business) throws Throwable {
  // 1. get or create a transaction
  GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();

  // 1.1 get transactionInfo
  TransactionInfo txInfo = business.getTransactionInfo();
  if (txInfo == null) {
    throw new ShouldNeverHappenException("transactionInfo does not exist");
  }
  try {

    // 2. begin transaction
    beginTransaction(txInfo, tx);

    Object rs = null;
    try {

      // Do Your Business
      rs = business.execute();

    } catch (Throwable ex) {

      // 3.the needed business exception to rollback.
      completeTransactionAfterThrowing(txInfo,tx,ex);
      throw ex;
    }

    // 4. everything is fine, commit.
    commitTransaction(tx);

    return rs;
  } finally {
    //5. clear
    triggerAfterCompletion();
    cleanUp();
  }
}

以上是不是有一種似曾相識的感覺？沒錯，以上就是我們使用 API 時經常寫的冗餘程式碼，現在 Spring 通過代理模式，把這些冗餘程式碼都封裝帶模版裡面了，它將那些冗餘程式碼統統封裝起來統一流程處理，並不需要你顯示寫出來了，有興趣的也可以去看看 Mybatis-spring 的原始碼，也是寫得非常精彩。

服務端處理邏輯

服務端收到客戶端的連線，那當然是將其 channel 也快取起來，前面也說到客戶端會發送 RegisterRMRequest/RegisterTMRequest 請求給服務端，服務端收到後會呼叫 ServerMessageListener 監聽器處理：

io.seata.core.rpc.ServerMessageListener

public interface ServerMessageListener {
  // 處理各種事務，如分支註冊、分支提交、分支上報、分支回滾等等
  void onTrxMessage(RpcMessage request, ChannelHandlerContext ctx, ServerMessageSender sender);
    // 處理 RM 客戶端的註冊連線
  void onRegRmMessage(RpcMessage request, ChannelHandlerContext ctx,
                      ServerMessageSender sender, RegisterCheckAuthHandler checkAuthHandler);
  // 處理 TM 客戶端的註冊連線
  void onRegTmMessage(RpcMessage request, ChannelHandlerContext ctx,
                      ServerMessageSender sender, RegisterCheckAuthHandler checkAuthHandler);
  // 服務端與客戶端保持心跳
  void onCheckMessage(RpcMessage request, ChannelHandlerContext ctx, ServerMessageSender sender)

}

ChannelManager 是服務端 channel 的管理器，服務端每次和客戶端通訊，都需要從 ChannelManager 中獲取客戶端對應的 channel，它用於儲存 TM 和 RM 客戶端 channel 的快取結構如下：

/**
 * resourceId -> applicationId -> ip -> port -> RpcContext
 */
private static final ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer,
RpcContext>>>>
  RM_CHANNELS = new ConcurrentHashMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer,
RpcContext>>>>();

/**
 * ip+appname,port
 */
private static final ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> TM_CHANNELS
  = new ConcurrentHashMap<String, ConcurrentMap<Integer, RpcContext>>();

以上的 Map 結構有點複雜：

RM_CHANNELS：

1. resourceId 指的是 RM client 的資料庫地址；
2. applicationId 指的是 RM client 的服務 Id，比如 springboot 的配置 spring.application.name=account-service 中的 account-service 即是 applicationId；
3. ip 指的是 RM client 服務地址；
4. port 指的是 RM client 服務地址；
5. RpcContext 儲存了本次註冊請求的資訊。

TM_CHANNELS：

1. ip+appname：這裡的註釋應該是寫錯了，應該是 appname+ip，即 TM_CHANNELS 的 Map 結構第一個 key 為 appname+ip；
2. port：客戶端的埠號。

以下是 RM Client 註冊邏輯：

io.seata.core.rpc.ChannelManager#registerRMChannel

public static void registerRMChannel(RegisterRMRequest resourceManagerRequest, Channel channel)
  throws IncompatibleVersionException {
  Version.checkVersion(resourceManagerRequest.getVersion());
  // 將 ResourceIds 資料庫連線連線資訊放入一個set中
  Set<String> dbkeySet = dbKeytoSet(resourceManagerRequest.getResourceIds());
  RpcContext rpcContext;
  // 從快取中判斷是否有該channel資訊
  if (!IDENTIFIED_CHANNELS.containsKey(channel)) {
    // 根據請求註冊資訊，構建 rpcContext
    rpcContext = buildChannelHolder(NettyPoolKey.TransactionRole.RMROLE, resourceManagerRequest.getVersion(),
                                    resourceManagerRequest.getApplicationId(), resourceManagerRequest.getTransactionServiceGroup(),
                                    resourceManagerRequest.getResourceIds(), channel);
    // 將 rpcContext 放入快取中
    rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
  } else {
    rpcContext = IDENTIFIED_CHANNELS.get(channel);
    rpcContext.addResources(dbkeySet);
  }
  if (null == dbkeySet || dbkeySet.isEmpty()) { return; }
  for (String resourceId : dbkeySet) {
    String clientIp;
    // 將請求資訊存入 RM_CHANNELS 中，這裡用了 java8 的 computeIfAbsent 方法操作
    ConcurrentMap<Integer, RpcContext> portMap = RM_CHANNELS.computeIfAbsent(resourceId, resourceIdKey -> new ConcurrentHashMap<>())
      .computeIfAbsent(resourceManagerRequest.getApplicationId(), applicationId -> new ConcurrentHashMap<>())
      .computeIfAbsent(clientIp = getClientIpFromChannel(channel), clientIpKey -> new ConcurrentHashMap<>());
        // 將當前 rpcContext 放入 portMap 中
    rpcContext.holdInResourceManagerChannels(resourceId, portMap);
    updateChannelsResource(resourceId, clientIp, resourceManagerRequest.getApplicationId());
  }
}

從以上程式碼邏輯能夠看出，註冊 RM client 主要是將註冊請求資訊，放入 RM_CHANNELS 快取中，同時還會從 IDENTIFIED_CHANNELS 中判斷本次請求的 channel 是否已驗證過，IDENTIFIED_CHANNELS 的結構如下：

private static final ConcurrentMap<Channel, RpcContext> IDENTIFIED_CHANNELS
  = new ConcurrentHashMap<>();

IDENTIFIED_CHANNELS 包含了所有 TM 和 RM 已註冊的 channel。

以下是 TM 註冊邏輯：

io.seata.core.rpc.ChannelManager#registerTMChannel

public static void registerTMChannel(RegisterTMRequest request, Channel channel)
  throws IncompatibleVersionException {
  Version.checkVersion(request.getVersion());
  // 根據請求註冊資訊，構建 RpcContext
  RpcContext rpcContext = buildChannelHolder(NettyPoolKey.TransactionRole.TMROLE, request.getVersion(),
                                             request.getApplicationId(),
                                             request.getTransactionServiceGroup(),
                                             null, channel);
  // 將 RpcContext 放入 IDENTIFIED_CHANNELS 快取中
  rpcContext.holdInIdentifiedChannels(IDENTIFIED_CHANNELS);
  // account-service:127.0.0.1:63353
  String clientIdentified = rpcContext.getApplicationId() + Constants.CLIENT_ID_SPLIT_CHAR
    + getClientIpFromChannel(channel);
  // 將請求資訊存入 TM_CHANNELS 快取中
  TM_CHANNELS.putIfAbsent(clientIdentified, new ConcurrentHashMap<Integer, RpcContext>());
  // 將上一步建立好的get出來，之後再將rpcContext放入這個map的value中
  ConcurrentMap<Integer, RpcContext> clientIdentifiedMap = TM_CHANNELS.get(clientIdentified);
  rpcContext.holdInClientChannels(clientIdentifiedMap);
}

TM client 的註冊大體類似，把本次註冊的資訊放入對應的快取中儲存，但比 RM client 的註冊邏輯簡單一些，主要是 RM client 會涉及分支事務資源的資訊，需要註冊的資訊也會比 TM client 多。

更多精彩文章請關注作者維護的公眾號「後端進階」，這是一個專注後端相關技術的公眾號。
關注公眾號並回復「後端」免費領取後端相關電子書籍。
歡迎分享，轉載請保留出處。

相關推薦