Seata AT模式TM處理流程圖文範例詳解

try{
  // 開啟分散式事務
  String xid = TM.beginGlobalTransaction();
  // 執行業務邏輯，包含遠端rpc呼叫
  RM1.execute(xid); -------RPC呼叫--------> RM2.execute(xid);
  // 提交分散式事務
  TM.commitGlobalTransaction(xid);
}catch(Exception e){
  // 回滾分散式事務
  TM.rollbackGlobalTransaction(xid);
}finally{
  // 恢復現場
}

@Override
    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);
        // 判斷這個方法是不是Object類中的toString()、equals()等方法
        if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
            // 通過被攔截的方法找出對應的註解GlobalTransactional和GlobalLock
            final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
            final GlobalTransactional globalTransactionalAnnotation =
                getAnnotation(method, targetClass, GlobalTransactional.class);
            final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);
            // 判斷是否開啟分散式事務，或者TM是否被降級處理，預設是沒有被降級的
            boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes);
            // 分散式事務可以正常使用
            if (!localDisable) {
                // 如果註解GlobalTransactional存在，那麼直接把裡面的設定解析成AspectTransactional
                if (globalTransactionalAnnotation != null || this.aspectTransactional != null) {
                    AspectTransactional transactional;
                    if (globalTransactionalAnnotation != null) {
                        transactional = new AspectTransactional(globalTransactionalAnnotation.timeoutMills(),
                            globalTransactionalAnnotation.name(), globalTransactionalAnnotation.rollbackFor(),
                            globalTransactionalAnnotation.rollbackForClassName(),
                            globalTransactionalAnnotation.noRollbackFor(),
                            globalTransactionalAnnotation.noRollbackForClassName(),
                            globalTransactionalAnnotation.propagation(),
                            globalTransactionalAnnotation.lockRetryInterval(),
                            globalTransactionalAnnotation.lockRetryTimes());
                    } else {
                        transactional = this.aspectTransactional;
                    }
                    // 呼叫handleGlobalTransaction處理
                    return handleGlobalTransaction(methodInvocation, transactional);
                } else if (globalLockAnnotation != null) {
                    // 呼叫handleGlobalLock處理
                    return handleGlobalLock(methodInvocation, globalLockAnnotation);
                }
            }
        }
        // 如果是Object類中的方法的話，直接呼叫，不作攔截
        return methodInvocation.proceed();
    }

Object handleGlobalTransaction(final MethodInvocation methodInvocation,
        final AspectTransactional aspectTransactional) throws Throwable {
        // 預設succeed=true
        boolean succeed = true;
        try {
            // 執行分散式事務處理邏輯
            // 詳細內容後面介紹
            return transactionalTemplate.execute(new TransactionalExecutor() {
                // 執行業務邏輯
                @Override
                public Object execute() throws Throwable {
                    return methodInvocation.proceed();
                }
                // 分散式事務名稱，沒有指定的話，就用【方法名+引數型別】命名
                public String name() {
                    String name = aspectTransactional.getName();
                    if (!StringUtils.isNullOrEmpty(name)) {
                        return name;
                    }
                    return formatMethod(methodInvocation.getMethod());
                }
                // 分散式事務資訊，其實就是@GlobalTransactional註解裡面拿到的設定
                @Override
                public TransactionInfo getTransactionInfo() {
                    // reset the value of timeout
                    int timeout = aspectTransactional.getTimeoutMills();
                    if (timeout <= 0 || timeout == DEFAULT_GLOBAL_TRANSACTION_TIMEOUT) {
                        timeout = defaultGlobalTransactionTimeout;
                    }
                    TransactionInfo transactionInfo = new TransactionInfo();
                    transactionInfo.setTimeOut(timeout);
                    transactionInfo.setName(name());
                    transactionInfo.setPropagation(aspectTransactional.getPropagation());
                    transactionInfo.setLockRetryInterval(aspectTransactional.getLockRetryInterval());
                    transactionInfo.setLockRetryTimes(aspectTransactional.getLockRetryTimes());
                    Set<RollbackRule> rollbackRules = new LinkedHashSet<>();
                    for (Class<?> rbRule : aspectTransactional.getRollbackFor()) {
                        rollbackRules.add(new RollbackRule(rbRule));
                    }
                    for (String rbRule : aspectTransactional.getRollbackForClassName()) {
                        rollbackRules.add(new RollbackRule(rbRule));
                    }
                    for (Class<?> rbRule : aspectTransactional.getNoRollbackFor()) {
                        rollbackRules.add(new NoRollbackRule(rbRule));
                    }
                    for (String rbRule : aspectTransactional.getNoRollbackForClassName()) {
                        rollbackRules.add(new NoRollbackRule(rbRule));
                    }
                    transactionInfo.setRollbackRules(rollbackRules);
                    return transactionInfo;
                }
            });
        } catch (TransactionalExecutor.ExecutionException e) {
            // 發生異常
            TransactionalExecutor.Code code = e.getCode();
            switch (code) {
                // 已經回滾過了
                case RollbackDone:
                    throw e.getOriginalException();
                // 開啟分散式事務失敗
                case BeginFailure:
                    // 分散式事務失敗
                    succeed = false;
                    // 呼叫失敗處理邏輯
                    failureHandler.onBeginFailure(e.getTransaction(), e.getCause());
                    throw e.getCause();
                // 分散式事務提交失敗
                case CommitFailure:
                    // 分散式事務失敗
                    succeed = false;
                    // 呼叫失敗處理邏輯
                    failureHandler.onCommitFailure(e.getTransaction(), e.getCause());
                    throw e.getCause();
                // 回滾失敗
                case RollbackFailure:
                    // 呼叫失敗處理邏輯
                    failureHandler.onRollbackFailure(e.getTransaction(), e.getOriginalException());
                    throw e.getOriginalException();
                // 回滾重試
                case RollbackRetrying:
                    // 呼叫失敗處理器中的回滾重試回撥邏輯
                    failureHandler.onRollbackRetrying(e.getTransaction(), e.getOriginalException());
                    throw e.getOriginalException();
                // 啥也不是，直接拋異常
                default:
                    throw new ShouldNeverHappenException(String.format("Unknown TransactionalExecutor.Code: %s", code));
            }
        } finally {
            // 如果允許TM降級，那麼這次處理完畢後，說明與TC恢復通訊，可以解除降級
            if (degradeCheck) {
                EVENT_BUS.post(new DegradeCheckEvent(succeed));
            }
        }
    }

public Object execute(TransactionalExecutor business) throws Throwable {
        // 1. 拿到整理好的@GlobalTransactional註解裡面的設定資訊
        TransactionInfo txInfo = business.getTransactionInfo();
        if (txInfo == null) {
            throw new ShouldNeverHappenException("transactionInfo does not exist");
        }
        // 1.1 獲取當前的分散式事務，如果為null的話，說明這是分散式事務的發起者；如果不為null，說明這是分散式事務的參與者
        GlobalTransaction tx = GlobalTransactionContext.getCurrent();
        // 1.2 獲取分散式事務的傳播級別，其實就是按照spring的傳播級別來一套，區別就是spring事務是本地事務，這是分散式事務，原理都一樣
        Propagation propagation = txInfo.getPropagation();
        SuspendedResourcesHolder suspendedResourcesHolder = null;
        try {
            // 這個switch裡面全都是處理分散式事務傳播級別的
            switch (propagation) {
                // 如果不支援分散式事務，如果當前存在事務，那麼先掛起當前的分散式事務，再執行業務邏輯
                case NOT_SUPPORTED:
                    // 分散式事務存在，先掛起
                    if (existingTransaction(tx)) {
                        suspendedResourcesHolder = tx.suspend();
                    }
                    // 執行業務邏輯
                    return business.execute();
                // 如果是每次都要建立一個新的分散式事務，先把當前存在的分散式事務掛起，然後建立一個新分散式事務
                case REQUIRES_NEW:
                    // 如果分散式事務存在，先掛起當前分散式事務，再建立一個新的分散式事務
                    if (existingTransaction(tx)) {
                        suspendedResourcesHolder = tx.suspend();
                        tx = GlobalTransactionContext.createNew();
                    }
                    // 之所以用break，是為了後面的程式碼和其他的傳播級別一起共用，業務邏輯肯定還是要執行的
                    break;
                // 如果支援分散式事務，如果當前不存在分散式事務，那麼直接執行業務邏輯，否則以分散式事務的方式執行業務邏輯
                case SUPPORTS:
                    // 如果不存在分散式事務，直接執行業務邏輯
                    if (notExistingTransaction(tx)) {
                        return business.execute();
                    }
                    // 否則，以分散式事務的方式執行業務邏輯
                    break;
                // 如果有分散式事務，就在當前分散式事務下執行業務邏輯，否則建立一個新的分散式事務執行業務邏輯
                case REQUIRED:
                    // If current transaction is existing, execute with current transaction,
                    // else continue and execute with new transaction.
                    break;
                // 如果不允許有分散式事務，那麼一旦發現存在分散式事務，直接拋異常；只有不存在分散式事務的時候才正常執行
                case NEVER:
                    // 存在分散式事務，拋異常
                    if (existingTransaction(tx)) {
                        throw new TransactionException(
                            String.format("Existing transaction found for transaction marked with propagation 'never', xid = %s"
                                    , tx.getXid()));
                    } else {
                        // 不存在分散式事務，執行業務邏輯
                        return business.execute();
                    }
                // 一定要有分散式事務，分散式事務不存在的話，拋異常；
                case MANDATORY:
                    // 不存在分散式事務，拋異常
                    if (notExistingTransaction(tx)) {
                        throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
                    }
                    // Continue and execute with current transaction.
                    break;
                default:
                    throw new TransactionException("Not Supported Propagation:" + propagation);
            }
            // 上面的傳播級別的邏輯處理完畢，下面就是公共的處理邏輯
            // 1.3 如果當前分散式事務沒有的話，那麼我們就要建立新的分散式事務，此時我們就是分散式事務的發起者，也就是TM本身，否則不能稱之為`TM`
            if (tx == null) {
                tx = GlobalTransactionContext.createNew();
            }
            // 開始準備幹活的條件
            // 把我們這個方法的全域性鎖設定放進當前執行緒中，並且把執行緒中已有的全域性鎖的設定取出來
            // 我們在幹完自己的活後，會把這個取出來的設定放回去的
            GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);
            try {
                // 2. 如果我們是分散式事務的發起者的話，那麼我們會和TC通訊，並且拿到一個XID；如果我們不是分散式事務的發起者的話，那麼這一步啥也不幹
                // 這個XID可以從RootContext中獲取
                beginTransaction(txInfo, tx);
                Object rs;
                try {
                    // 執行業務邏輯
                    rs = business.execute();
                } catch (Throwable ex) {
                    // 3. 發生任何異常，我們準備啟動回滾機制
                    completeTransactionAfterThrowing(txInfo, tx, ex);
                    throw ex;
                }
                // 4. 一切順利，通知提交分散式事務
                commitTransaction(tx);
                return rs;
            } finally {
                //5. 恢復現場，把之前的設定放回去
                resumeGlobalLockConfig(previousConfig);
                // 觸發回撥
                triggerAfterCompletion();
                // 清理工作
                cleanUp();
            }
        } finally {
            // 恢復之前掛起的事務
            if (suspendedResourcesHolder != null) {
                tx.resume(suspendedResourcesHolder);
            }
        }
    }

 private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
        try {
            // 回撥，預設是空回撥
            triggerBeforeBegin();
            // 發起分散式事務
            tx.begin(txInfo.getTimeOut(), txInfo.getName());
            // 回撥，預設是空回撥
            triggerAfterBegin();
        } catch (TransactionException txe) {
            throw new TransactionalExecutor.ExecutionException(tx, txe,
                TransactionalExecutor.Code.BeginFailure);
        }
    }
@Override
    public void begin(int timeout, String name) throws TransactionException {
        // 如果不是分散式事務發起者，那麼啥也不做
        if (role != GlobalTransactionRole.Launcher) {
            assertXIDNotNull();
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Ignore Begin(): just involved in global transaction [{}]", xid);
            }
            return;
        }
        assertXIDNull();
        // 如果當前已經處於分散式事務當中，那麼拋異常，因為事務發起者不可能事先處於別的分散式事務當中
        String currentXid = RootContext.getXID();
        if (currentXid != null) {
            throw new IllegalStateException("Global transaction already exists," +
                " can't begin a new global transaction, currentXid = " + currentXid);
        }
        // 發起分散式事務
        xid = transactionManager.begin(null, null, name, timeout);
        status = GlobalStatus.Begin;
        // 把xid繫結到當前執行緒中
        RootContext.bind(xid);
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Begin new global transaction [{}]", xid);
        }
    }
@Override
    public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)
        throws TransactionException {
        // 發起分散式事務開啟的請求
        GlobalBeginRequest request = new GlobalBeginRequest();
        request.setTransactionName(name);
        request.setTimeout(timeout);
        GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
        if (response.getResultCode() == ResultCode.Failed) {
            throw new TmTransactionException(TransactionExceptionCode.BeginFailed, response.getMsg());
        }
        // 獲取拿到的xid，表示分散式事務開啟成功
        return response.getXid();
    }

private void completeTransactionAfterThrowing(TransactionInfo txInfo, GlobalTransaction tx, Throwable originalException) throws TransactionalExecutor.ExecutionException {
        // 如果異常型別和指定的型別一致，那麼發起回滾；不一致還是要提交分散式事務
        if (txInfo != null && txInfo.rollbackOn(originalException)) {
            try {
                // 回滾分散式事務
                rollbackTransaction(tx, originalException);
            } catch (TransactionException txe) {
                // 回滾失敗拋異常
                throw new TransactionalExecutor.ExecutionException(tx, txe,
                        TransactionalExecutor.Code.RollbackFailure, originalException);
            }
        } else {
            // 不是指定的異常型別，還是繼續提交分散式事務
            commitTransaction(tx);
        }
    }
private void rollbackTransaction(GlobalTransaction tx, Throwable originalException) throws TransactionException, TransactionalExecutor.ExecutionException {
        // 執行回撥，預設空回撥
        triggerBeforeRollback();
        // 回滾
        tx.rollback();
        // 執行回撥，預設空回撥
        triggerAfterRollback();
        // 就算回滾沒問題，照樣拋異常，目的應該是告知開發人員此處產生了回滾
        throw new TransactionalExecutor.ExecutionException(tx, GlobalStatus.RollbackRetrying.equals(tx.getLocalStatus())
            ? TransactionalExecutor.Code.RollbackRetrying : TransactionalExecutor.Code.RollbackDone, originalException);
    }
@Override
    public void rollback() throws TransactionException {
        // 如果是分散式事務參與者，那麼啥也不做，RM的回滾不在這裡，這是TM的回滾
        if (role == GlobalTransactionRole.Participant) {
            // Participant has no responsibility of rollback
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Ignore Rollback(): just involved in global transaction [{}]", xid);
            }
            return;
        }
        assertXIDNotNull();
        // 下面就是一個迴圈重試發起分散式事務回滾
        int retry = ROLLBACK_RETRY_COUNT <= 0 ? DEFAULT_TM_ROLLBACK_RETRY_COUNT : ROLLBACK_RETRY_COUNT;
        try {
            while (retry > 0) {
                try {
                    retry--;
                    // 發起回滾的核心程式碼
                    status = transactionManager.rollback(xid);
                    // 回滾成功跳出迴圈
                    break;
                } catch (Throwable ex) {
                    LOGGER.error("Failed to report global rollback [{}],Retry Countdown: {}, reason: {}", this.getXid(), retry, ex.getMessage());
                    // 重試失敗次數完成才會跳出迴圈
                    if (retry == 0) {
                        throw new TransactionException("Failed to report global rollback", ex);
                    }
                }
            }
        } finally {
            // 如果回滾的分散式事務就是當前的分散式事務，那麼從當前執行緒中解綁XID
            if (xid.equals(RootContext.getXID())) {
                suspend();
            }
        }
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("[{}] rollback status: {}", xid, status);
        }
    }
@Override
    public GlobalStatus rollback(String xid) throws TransactionException {
        // 準備發起請求給TC，回滾指定的分散式事務
        GlobalRollbackRequest globalRollback = new GlobalRollbackRequest();
        globalRollback.setXid(xid);
        GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback);
        return response.getGlobalStatus();
    }

private void commitTransaction(GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
        try {
            // 回撥，預設空回撥
            triggerBeforeCommit();
            // 分散式事務提交
            tx.commit();
            // 回撥，預設空回撥
            triggerAfterCommit();
        } catch (TransactionException txe) {
            // 4.1 提交出異常，提交失敗
            throw new TransactionalExecutor.ExecutionException(tx, txe,
                TransactionalExecutor.Code.CommitFailure);
        }
    }
@Override
    public void commit() throws TransactionException {
        // 如果只是分散式事務參與者，那麼啥也不幹，TM只能有一個，哈哈
        if (role == GlobalTransactionRole.Participant) {
            // Participant has no responsibility of committing
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Ignore Commit(): just involved in global transaction [{}]", xid);
            }
            return;
        }
        assertXIDNotNull();
        // 分散式事務提交也是有重試的
        int retry = COMMIT_RETRY_COUNT <= 0 ? DEFAULT_TM_COMMIT_RETRY_COUNT : COMMIT_RETRY_COUNT;
        try {
            while (retry > 0) {
                try {
                    retry--;
                    // 發起分散式事務提交
                    status = transactionManager.commit(xid);
                    // 提交成功跳出迴圈
                    break;
                } catch (Throwable ex) {
                    LOGGER.error("Failed to report global commit [{}],Retry Countdown: {}, reason: {}", this.getXid(), retry, ex.getMessage());
                    // 重試結束，依然失敗就拋異常
                    if (retry == 0) {
                        throw new TransactionException("Failed to report global commit", ex);
                    }
                }
            }
        } finally {
            // 如果提交的分散式事務就是當前事務，那麼需要清理當前執行緒中的XID
            if (xid.equals(RootContext.getXID())) {
                suspend();
            }
        }
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("[{}] commit status: {}", xid, status);
        }
    }
@Override
    public GlobalStatus commit(String xid) throws TransactionException {
        // 發起分散式事務提交請求，這是與TC通訊
        GlobalCommitRequest globalCommit = new GlobalCommitRequest();
        globalCommit.setXid(xid);
        GlobalCommitResponse response = (GlobalCommitResponse) syncCall(globalCommit);
        return response.getGlobalStatus();
    }

private Object handleGlobalLock(final MethodInvocation methodInvocation, final GlobalLock globalLockAnno) throws Throwable {
        // 模版模式實現全域性鎖
        return globalLockTemplate.execute(new GlobalLockExecutor() {
            // 執行業務邏輯
            @Override
            public Object execute() throws Throwable {
                return methodInvocation.proceed();
            }
            // 獲取全域性鎖設定
            // 一個是全域性鎖重試間隔時間
            // 一個是全域性鎖重試次數
            @Override
            public GlobalLockConfig getGlobalLockConfig() {
                GlobalLockConfig config = new GlobalLockConfig();
                config.setLockRetryInterval(globalLockAnno.lockRetryInterval());
                config.setLockRetryTimes(globalLockAnno.lockRetryTimes());
                return config;
            }
        });
    }
public Object execute(GlobalLockExecutor executor) throws Throwable {
        // 判斷當前是否有全域性鎖
        boolean alreadyInGlobalLock = RootContext.requireGlobalLock();
        // 如果沒有全域性鎖，那麼在當前執行緒中設定需要全域性鎖標識
        if (!alreadyInGlobalLock) {
            RootContext.bindGlobalLockFlag();
        }
        // 把全域性鎖的設定設定進當前執行緒，並把執行緒中已有的全域性鎖設定拿出來，後面恢復現場需要用
        GlobalLockConfig myConfig = executor.getGlobalLockConfig();
        GlobalLockConfig previousConfig = GlobalLockConfigHolder.setAndReturnPrevious(myConfig);
        try {
            // 執行業務邏輯
            return executor.execute();
        } finally {
            // 清除執行緒中的全域性鎖標記
            if (!alreadyInGlobalLock) {
                RootContext.unbindGlobalLockFlag();
            }
            // 恢復現場
            if (previousConfig != null) {
                GlobalLockConfigHolder.setAndReturnPrevious(previousConfig);
            } else {
                GlobalLockConfigHolder.remove();
            }
        }
    }