seata源码解析：AT模式-启动分析

引言

在官方例子中，AT模式启动主要会加载：io.seata.spring.annotation.GlobalTransactionScanner 这个bean，使用AT模式必须在执行业务功能的方法上加上@GlobalTransactional注解，接下来我们来看看seata启动时做了哪些事情。

xml配置bean：dubbo-business.xml

<bean class="io.seata.spring.annotation.GlobalTransactionScanner">
    <constructor-arg value="dubbo-demo-app"/>
    <constructor-arg value="my_test_tx_group"/>
</bean>

@Override
@GlobalTransactional(timeoutMills = 300000, name = "dubbo-demo-tx")
public void purchase(String userId, String commodityCode, int orderCount) {
    LOGGER.info("purchase begin ... xid: " + RootContext.getXID());
    storageService.deduct(commodityCode, orderCount);
    orderService.create(userId, commodityCode, orderCount);

    try {
        Thread.sleep(10000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    throw new RuntimeException("xxx");

}

2.源码分析

2.1 GlobalTransactionScanner详解

GlobalTransactionScanner类继承AbstractAutoProxyCreator类，实现了InitializingBean接口。我们知道AbstractAutoProxyCreator主要功能是spring AOP中提供代理功能，InitializingBean接口是初始化类时去做一些预先准备工作。

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

2.1.1 GlobalTransactionScanner#wrapIfNecessary()详解

GlobalTransactionScanner#wrapIfNecessary()方法在spring bean 初始化时会被调用，这里主要是给加了：@GlobalTransactional 注解的方法添加分布式事务功能。方法详解：

@Override
protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
    if (disableGlobalTransaction) {//@1
        return bean;
    }
    try {
        synchronized (PROXYED_SET) {//@2
            if (PROXYED_SET.contains(beanName)) {//@3
                return bean;
            }
            interceptor = null;
            //check TCC proxy
            if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {//@4
                //TCC interceptor, proxy bean of sofa:reference/dubbo:reference, and LocalTCC
                interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));
            } else {
                Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);//@5
                Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);

                if (!existsAnnotation(new Class[]{serviceInterface})//@6
                    && !existsAnnotation(interfacesIfJdk)) {
                    return bean;
                }

                if (interceptor == null) {//@7
                    if (globalTransactionalInterceptor == null) {
                        globalTransactionalInterceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
                        ConfigurationCache.addConfigListener(
                            ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
                            (ConfigurationChangeListener)globalTransactionalInterceptor);
                    }
                    interceptor = globalTransactionalInterceptor;
                }
            }

            LOGGER.info("Bean[{}] with name [{}] would use interceptor [{}]", bean.getClass().getName(), beanName, interceptor.getClass().getName());
            if (!AopUtils.isAopProxy(bean)) {//@8
                bean = super.wrapIfNecessary(bean, beanName, cacheKey);
            } else {
                AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);//@9
                Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));//@10
                for (Advisor avr : advisor) {
                    advised.addAdvisor(0, avr);
                }
            }
            PROXYED_SET.add(beanName);//@11
            return bean;
        }
    } catch (Exception exx) {
        throw new RuntimeException(exx);
    }
}

代码@1：判断是否开启seata分布式事务，如果未开启则返回。

代码@2：加锁操作，防止并发问题。

代码@3：根据beanName判断代理类是否已经生成过，如果已经生成过代理类则返回。

代码@4：判断是否是tcc模式，如果是tcc则创建tcc的代理类。

代码@5：找到需要被代理对象的目标对象。

代码@6：判断是否有GlobalTransactional注解。

代码@7：MethodInterceptor为空，则创建MethodInterceptor代理执行对象。

代码@8：判断是否是AOP代理对象，如果不是则调用父类的AbstractAutoProxyCreator.wrapIfNecessary()方法

代码@9：

代码@10：

代码@11：将生成好的代理对象beanName放入Set集合中，用于防止同一个bean被重复创建代理。

2.1.2 AbstractAutoProxyCreator.wrapIfNecessary()方法

protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
    if (StringUtils.hasLength(beanName) && this.targetSourcedBeans.contains(beanName)) {
        return bean;
    } else if (Boolean.FALSE.equals(this.advisedBeans.get(cacheKey))) {
        return bean;
    } else if (!this.isInfrastructureClass(bean.getClass()) && !this.shouldSkip(bean.getClass(), beanName)) {
        Object[] specificInterceptors = this.getAdvicesAndAdvisorsForBean(bean.getClass(), beanName, (TargetSource)null);//@1
        if (specificInterceptors != DO_NOT_PROXY) {//@2
            this.advisedBeans.put(cacheKey, Boolean.TRUE);
            Object proxy = this.createProxy(bean.getClass(), beanName, specificInterceptors, new SingletonTargetSource(bean));//@3
            this.proxyTypes.put(cacheKey, proxy.getClass());
            return proxy;
        } else {
            this.advisedBeans.put(cacheKey, Boolean.FALSE);
            return bean;
        }
    } else {
        this.advisedBeans.put(cacheKey, Boolean.FALSE);
        return bean;
    }
}

代码@1：这里会调用GlobalTransactionScanner.getAdvicesAndAdvisorsForBean方法获取wrapIfNecessary方法初始化的GlobalTransactionalInterceptor。

代码@2：判断是否需要创建代理。

代码@3：创建代理对象。

2.1.3 GlobalTransactionScanner.getAdvicesAndAdvisorsForBean()

@Override
protected Object[] getAdvicesAndAdvisorsForBean(Class beanClass, String beanName, TargetSource customTargetSource)
    throws BeansException {
    return new Object[]{interceptor};
}

getAdvicesAndAdvisorsForBean()方法返回wrapIfNecessary()创建的interceptor对象。

2.2 GlobalTransactionalInterceptor 详解

该类实现了MethodInterceptor方法，是动态代理的实际执行类。

@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);
    if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
        final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
        final GlobalTransactional globalTransactionalAnnotation =
            getAnnotation(method, targetClass, GlobalTransactional.class);//@1
        final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);//@2
        boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes);//@3
        if (!localDisable) {
            if (globalTransactionalAnnotation != null) {//@4
                return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
            } else if (globalLockAnnotation != null) {
                return handleGlobalLock(methodInvocation);
            }
        }
    }
    return methodInvocation.proceed();
}

代码@1：从当前方法上获取GlobalTransactional注解。

代码@2：从当前方法上获取GlobalLock注解。

代码@3：判断是否打开全局事务开关。

代码@4：如果GlobalTransactional注解不为空，则走全局事务逻辑执行handleGlobalTransaction()方法

GlobalTransactionalInterceptor.handleGlobalTransaction()方法

该方法主要功能是调用seata transactionalTemplate对象进行分布式事务处理，execute()方法中传入了一个匿名类对象：TransactionalExecutor。

它主要功能是这三个方法：

- public Object execute()：执行被代理对象的原生方法。
- public String name()：返回GlobalTransactional注解上的名称，如果名称为空则返回代理对象方法的名称。
- public TransactionInfo getTransactionInfo()：这个方法主要作用是设置分布式事务信息，比如超时时间、名称、以及传播机制和回滚信息等。

private Object handleGlobalTransaction(final MethodInvocation methodInvocation,
    final GlobalTransactional globalTrxAnno) throws Throwable {
    boolean succeed = true;
    try {
        return transactionalTemplate.execute(new TransactionalExecutor() {
            @Override
            public Object execute() throws Throwable {
                return methodInvocation.proceed();
            }

            public String name() {
                String name = globalTrxAnno.name();
                if (!StringUtils.isNullOrEmpty(name)) {
                    return name;
                }
                return formatMethod(methodInvocation.getMethod());
            }

            @Override
            public TransactionInfo getTransactionInfo() {
                TransactionInfo transactionInfo = new TransactionInfo();
                transactionInfo.setTimeOut(globalTrxAnno.timeoutMills());
                transactionInfo.setName(name());
                transactionInfo.setPropagation(globalTrxAnno.propagation());
                Set<RollbackRule> rollbackRules = new LinkedHashSet<>();
                for (Class<?> rbRule : globalTrxAnno.rollbackFor()) {
                    rollbackRules.add(new RollbackRule(rbRule));
                }
                for (String rbRule : globalTrxAnno.rollbackForClassName()) {
                    rollbackRules.add(new RollbackRule(rbRule));
                }
                for (Class<?> rbRule : globalTrxAnno.noRollbackFor()) {
                    rollbackRules.add(new NoRollbackRule(rbRule));
                }
                for (String rbRule : globalTrxAnno.noRollbackForClassName()) {
                    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 {
        if (degradeCheck) {
            EVENT_BUS.post(new DegradeCheckEvent(succeed));
        }
    }
}

2.2.1 TransactionalTemplate详解

该类是seata分布式事务执行业务逻辑模板类，它里面定义了获取事务信息、处理隔离级别、开启分布式事务、执行业务逻辑，以及提交分布式事务和回顾分布式事务等等功能。

TransactionalTemplate.execute()方法

public Object execute(TransactionalExecutor business) throws Throwable {
    // 1 get transactionInfo
    TransactionInfo txInfo = business.getTransactionInfo();//@1
    if (txInfo == null) {
        throw new ShouldNeverHappenException("transactionInfo does not exist");
    }
    // 1.1 get or create a transaction
    GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();//@2

    // 1.2 Handle the Transaction propatation and the branchType
    Propagation propagation = txInfo.getPropagation();//@3
    SuspendedResourcesHolder suspendedResourcesHolder = null;
    try {
        switch (propagation) {//@4
            case NOT_SUPPORTED:
                suspendedResourcesHolder = tx.suspend(true);
                return business.execute();
            case REQUIRES_NEW:
                suspendedResourcesHolder = tx.suspend(true);
                break;
            case SUPPORTS:
                if (!existingTransaction()) {
                    return business.execute();
                }
                break;
            case REQUIRED:
                break;
            case NEVER:
                if (existingTransaction()) {
                    throw new TransactionException(
                            String.format("Existing transaction found for transaction marked with propagation 'never',xid = %s"
                                    ,RootContext.getXID()));
                } else {
                    return business.execute();
                }
            case MANDATORY:
                if (!existingTransaction()) {
                    throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
                }
                break;
            default:
                throw new TransactionException("Not Supported Propagation:" + propagation);
        }


        try {

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

            Object rs = null;
            try {

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

            } catch (Throwable ex) {

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

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

            return rs;
        } finally {
            //5. clear
            triggerAfterCompletion();//@9
            cleanUp();
        }
    } finally {
        tx.resume(suspendedResourcesHolder);
    }

}

上面execute()方法有相关的注释，代码也非常清晰，我们来简单看看他的注释：

代码@1：获取全局事务信息。

代码@2：获取或创建GlobalTransaction事务执行对象。

代码@3：获取全局事务传播机制，seata全局事务传播机制和jdbc中的类似。

代码@4：根据事务传播机制执行不同的业务逻辑。

代码@5：开始全局事务。

代码@6：执行当前业务方法。

代码@7：如果业务方法抛出异常，则对业务方法进行回滚操作。

代码@8：提交分布式事务。

代码@9：清除一些回调的hook。

TransactionalTemplate.beginTransaction()方法

beginTransaction()方法比较简单，它会调用GlobalTransaction.begin()方法开启事务。

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);
    }
}

TransactionalTemplate.commitTransaction()方法

commitTransaction()方法比较简单，它会调用GlobalTransaction.commit()方法开始提交事务。

private void commitTransaction(GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
    try {
        triggerBeforeCommit();
        tx.commit();
        triggerAfterCommit();
    } catch (TransactionException txe) {
        // 4.1 Failed to commit
        throw new TransactionalExecutor.ExecutionException(tx, txe,
            TransactionalExecutor.Code.CommitFailure);
    }
}

2.2.2 DefaultGlobalTransaction详解

DefaultGlobalTransaction分布式事务的默认实现类，它主要功能有事务开始、提交、回滚，以及对分布式事务调用出现异常等场景做重试处理。因为它代码逻辑都比较简单，这里就不做具体分析。

2.2.3 DefaultTransactionManager详解

默认事务管理器，它主要功能是负责和TC进行事务的开始、提交、回滚等操作，是与TC进行实际交互的类，底层调用Netty网络通信方法发送RPC请求给TC。

@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());
    }
    return response.getXid();
}

@Override
public GlobalStatus commit(String xid) throws TransactionException {
    GlobalCommitRequest globalCommit = new GlobalCommitRequest();
    globalCommit.setXid(xid);
    GlobalCommitResponse response = (GlobalCommitResponse) syncCall(globalCommit);
    return response.getGlobalStatus();
}

@Override
public GlobalStatus rollback(String xid) throws TransactionException {
    GlobalRollbackRequest globalRollback = new GlobalRollbackRequest();
    globalRollback.setXid(xid);
    GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback);
    return response.getGlobalStatus();
}

2.3 启动TM、RM Netty客户端

@Override
public void afterPropertiesSet() {
    if (disableGlobalTransaction) {
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Global transaction is disabled.");
        }
        return;
    }
    initClient();
}

private void initClient() {
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Initializing Global Transaction Clients ... ");
    }
    if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
        throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
    }
    //init TM
    TMClient.init(applicationId, txServiceGroup);
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
    }
    //init RM
    RMClient.init(applicationId, txServiceGroup);
    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
    }

    if (LOGGER.isInfoEnabled()) {
        LOGGER.info("Global Transaction Clients are initialized. ");
    }
    registerSpringShutdownHook();

}

public class TMClient {

    /**
     * Init.
     *
     * @param applicationId           the application id
     * @param transactionServiceGroup the transaction service group
     */
    public static void init(String applicationId, String transactionServiceGroup) {
        TmNettyRemotingClient tmNettyRemotingClient = TmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup);
        tmNettyRemotingClient.init();
    }

}

TmNettyRemotingClient

@Override
public void init() {
    // registry processor
    registerProcessor();
    if (initialized.compareAndSet(false, true)) {
        super.init();
    }
}

private void registerProcessor() {
    // 1.registry TC response processor
    ClientOnResponseProcessor onResponseProcessor =
        new ClientOnResponseProcessor(mergeMsgMap, super.getFutures(), getTransactionMessageHandler());
    super.registerProcessor(MessageType.TYPE_SEATA_MERGE_RESULT, onResponseProcessor, null);
    super.registerProcessor(MessageType.TYPE_GLOBAL_BEGIN_RESULT, onResponseProcessor, null);
    super.registerProcessor(MessageType.TYPE_GLOBAL_COMMIT_RESULT, onResponseProcessor, null);
    super.registerProcessor(MessageType.TYPE_GLOBAL_REPORT_RESULT, onResponseProcessor, null);
    super.registerProcessor(MessageType.TYPE_GLOBAL_ROLLBACK_RESULT, onResponseProcessor, null);
    super.registerProcessor(MessageType.TYPE_GLOBAL_STATUS_RESULT, onResponseProcessor, null);
    super.registerProcessor(MessageType.TYPE_REG_CLT_RESULT, onResponseProcessor, null);
    // 2.registry heartbeat message processor
    ClientHeartbeatProcessor clientHeartbeatProcessor = new ClientHeartbeatProcessor();
    super.registerProcessor(MessageType.TYPE_HEARTBEAT_MSG, clientHeartbeatProcessor, null);
}

2.4 dubbo适配

Seata适配dubbo框架实现分布式事务能力主要是通过dubbo Filter机制做拓展，ApacheDubboTransactionPropagationFilter类实现了Dubbo Filter接口，@Activate注解标识它为Provider、Consumer都可用。

下面我们来详细分析下Seata提供的Dubbo适配类代码：

@Activate(group = {DubboConstants.PROVIDER, DubboConstants.CONSUMER}, order = 100)
public class ApacheDubboTransactionPropagationFilter implements Filter {

    private static final Logger LOGGER = LoggerFactory.getLogger(ApacheDubboTransactionPropagationFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String xid = RootContext.getXID();//@1
        String branchType = RootContext.getBranchType();//@2

        String rpcXid = getRpcXid();//@3
        String rpcBranchType = RpcContext.getContext().getAttachment(RootContext.KEY_BRANCH_TYPE);//@4
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("xid in RootContext[{}] xid in RpcContext[{}]", xid, rpcXid);
        }
        boolean bind = false;
        if (xid != null) {//@5
            RpcContext.getContext().setAttachment(RootContext.KEY_XID, xid);
            RpcContext.getContext().setAttachment(RootContext.KEY_BRANCH_TYPE, branchType);
        } else {
            if (rpcXid != null) {//@6
                RootContext.bind(rpcXid);//@7
                if (StringUtils.equals(BranchType.TCC.name(), rpcBranchType)) {
                    RootContext.bindBranchType(BranchType.TCC);
                }
                bind = true;
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("bind xid [{}] branchType [{}] to RootContext", rpcXid, rpcBranchType);
                }
            }
        }
        try {
            return invoker.invoke(invocation);//@8
        } finally {
            if (bind) {//@9
                String unbindXid = RootContext.unbind();
                String previousBranchType = RootContext.getBranchType();
                if (StringUtils.equals(BranchType.TCC.name(), previousBranchType)) {
                    RootContext.unbindBranchType();
                }
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("unbind xid [{}] branchType [{}] from RootContext", unbindXid, previousBranchType);
                }
                if (!rpcXid.equalsIgnoreCase(unbindXid)) {
                    LOGGER.warn("xid in change during RPC from {} to {},branchType from {} to {}", rpcXid, unbindXid,
                            rpcBranchType != null ? rpcBranchType : "AT", previousBranchType);
                    if (unbindXid != null) {
                        RootContext.bind(unbindXid);
                        LOGGER.warn("bind xid [{}] back to RootContext", unbindXid);
                        if (StringUtils.equals(BranchType.TCC.name(), previousBranchType)) {
                            RootContext.bindBranchType(BranchType.TCC);
                            LOGGER.warn("bind branchType [{}] back to RootContext", previousBranchType);
                        }
                    }
                }
            }
        }
    }

    /**
     * get rpc xid
     * @return
     */
    private String getRpcXid() {
        String rpcXid = RpcContext.getContext().getAttachment(RootContext.KEY_XID);
        if (rpcXid == null) {
            rpcXid = RpcContext.getContext().getAttachment(RootContext.KEY_XID.toLowerCase());
        }
        return rpcXid;
    }

}

代码@1、2：从Context上下文中获取全局事务xid、branchType。

代码@3、4：从RPC中获取xid、branchType。

代码@5：如果xid不为空，则设置xid进入dubbo的RpcContext中。

代码@6、7：如果xid为空rpcXid不为空，则设置xid、branchType进入Seata RootContext对象中。

代码@8：向后执行Filter invoker

代码@9：判断context是否绑定了信息，如果绑定信息则进行解绑操作。

原文链接：https://blog.csdn.net/xiaoyi5224765/article/details/115712357