Seata入门系列【11】Seata源码分析之全局事务扫描器GlobalTransactionScanner和全局事务拦截器GlobalTransactionalInterceptor

阳光穿透心脏的1/2处 2024-03-27 18:40 69阅读 0赞

## 1 全局事务扫描器GlobalTransactionScanner ##

### 1.1 自动配置类 ###

基于Spring Boot的自动配置功能，可以在seata-spring-boot-starter中找到Seata 引入的自动配置类，当开启了seata.enabled=true时，将会加载。  
![在这里插入图片描述][92452ed1d4c14bbea9a9e0b42cb42729.png]  
扫描器是在SeataAutoConfiguration自动配置类中加载注入的：

@ConditionalOnProperty(prefix = SEATA_PREFIX, name = "enabled", havingValue = "true", matchIfMissing = true)
    public class SeataAutoConfiguration {
        
        private static final Logger LOGGER = LoggerFactory.getLogger(SeataAutoConfiguration.class);
    
        @Bean(BEAN_NAME_FAILURE_HANDLER)
        @ConditionalOnMissingBean(FailureHandler.class)
        public FailureHandler failureHandler() {
        
            return new DefaultFailureHandlerImpl();
        }
    
        @Bean
        @DependsOn({
        BEAN_NAME_SPRING_APPLICATION_CONTEXT_PROVIDER, BEAN_NAME_FAILURE_HANDLER})
        @ConditionalOnMissingBean(GlobalTransactionScanner.class)
        public GlobalTransactionScanner globalTransactionScanner(SeataProperties seataProperties, FailureHandler failureHandler) {
        
            if (LOGGER.isInfoEnabled()) {
        
                LOGGER.info("Automatically configure Seata");
            }
            // 直接new 全局事务扫描器，传入applicationId、txServiceGroup、失败处理器。
            return new GlobalTransactionScanner(seataProperties.getApplicationId(), seataProperties.getTxServiceGroup(), failureHandler);
        }
    }

### 1.2 全局事务扫描器 ###

GlobalTransactionScanner继承并实现了一系列接口，实现了众多功能。

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

#### 1.2.1 初始化客户端 ####

实现了InitializingBean接口，该接口为bean提供了初始化方法的方式，它只包括afterPropertiesSet方法，凡是继承该接口的类，在初始化bean的时候都会执行该方法。

GlobalTransactionScanner实现了afterPropertiesSet方法，主要是执行 initClient()初始化分布式事务客户端：

@Override
        public void afterPropertiesSet() {
        
        	// 查询配置service.disableGlobalTransaction 是否开启当前服务的分布式事务
            if (disableGlobalTransaction) {
        
                if (LOGGER.isInfoEnabled()) {
        
                    LOGGER.info("Global transaction is disabled.");
                }
                ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
                        (ConfigurationChangeListener)this);
                return;
            }
            // 开启了全局事务，则会初始化客户端
            if (initialized.compareAndSet(false, true)) {
        
                initClient();
            }
        }

在initClient()方法中，会初始化TM、RM，使用netty 和Seata 服务端进行通信。

private void initClient() {
        
            if (LOGGER.isInfoEnabled()) {
        
                LOGGER.info("Initializing Global Transaction Clients ... ");
            }
            // 1. 应用ID、事务分组为空则会抛出异常 
            if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
        
                throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
            }
            //init TM
            // 初始化事务管理器TM，打印日志
            TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
            if (LOGGER.isInfoEnabled()) {
        
                LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
            }
            //init RM
            // 初始化资源管理器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();
        }

#### 1.2.2 注销 ####

实现了DisposableBean接口，该接口负责在Bean生命周期结束前调用destroy()方法做一些收尾工作。

destroy()负责在应用关闭后，释放相关资源。

@Override
        public void destroy() {
        
            ShutdownHook.getInstance().destroyAll();
        }

#### 1.2.3 AOP 代理 ####

GlobalTransactionScanner继承了AbstractAutoProxyCreator(自动代理创建者) 抽象类，Spring 通过 AbstractAutoProxyCreator在 bean 初始化完成之后创建它的代理AOP 代理，其主要有两个核心方法：

*  wrapIfNecessary：代理增强的前置判断处理，表示是否该Bean需要增强，如果增强的话创建代理类
 *  postProcessAfterInitialization：当一个spring的Bean已经初始化完毕的时候，后置处理一些东西

GlobalTransactionScanner中的wrapIfNecessary（如果需要则进行包装）方法，主要是对存在全局事务注解的Bean，添加相应的拦截器。

/**
         * 将扫描以下内容，并添加相应的拦截器
         *
         * 1. TM: @GlobalTransactional注解添加GlobalTransactionalInterceptor（handleGlobalTransaction方法处理）
         * 2. @GlobalLock注解添加GlobalTransactionalInterceptor（handleGlobalLock方法处理）
         * @see io.seata.spring.annotation.GlobalLock // GlobalLock annotation
         * 3. TCC 模式，添加TccActionInterceptor 拦截器
         *  @LocalTCC 
         *  @TwoPhaseBusinessAction 
         *  RemotingParser接口 
         */
        @Override
        protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
        
            try {
        
                synchronized (PROXYED_SET) {
        
                // 1. 如果已被代理，则跳过该Bean ,PROXYED_SET是一个Set<String> 集合
                    if (PROXYED_SET.contains(beanName)) {
        
                        return bean;
                    }
                    interceptor = null;
                    //check TCC proxy
                    //  2. 判断是否开启TCC模式，是则添加TCC 拦截器
                    if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {
        
                        //TCC interceptor, proxy bean of sofa:reference/dubbo:reference, and LocalTCC
                        interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));
                        ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
                            (ConfigurationChangeListener)interceptor);
                    } else {
        
                    // 3. 查询Bean的 Class 类型
                        Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
                        Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);
    					// 如果类或者方法上没有GlobalTransactional或者GlobalLock 注解，直接返回Bean
                        if (!existsAnnotation(new Class[]{
        serviceInterface})
                            && !existsAnnotation(interfacesIfJdk)) {
        
                            return bean;
                        }
    					// 4. 初始化GlobalTransactionalInterceptor
                        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());
                    // 5. 判断是否已经是AOP代理类，不是则走一遍父类的包装
                    if (!AopUtils.isAopProxy(bean)) {
        
                        bean = super.wrapIfNecessary(bean, beanName, cacheKey);
                    } else {
        
                    	// 6. 代理对象添加拦截器
                        AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
                        Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
                        for (Advisor avr : advisor) {
        
                            advised.addAdvisor(0, avr);
                        }
                    }
                    // 添加到已处理的集合中，标识该bean已经处理过了
                    PROXYED_SET.add(beanName);
                    return bean;
                }
            } catch (Exception exx) {
        
                throw new RuntimeException(exx);
            }
        }

#### 1.2.4 总结 ####

1.  自动配置类，注入全局事务扫描器GlobalTransactionScanner
2.  GlobalTransactionScanner会初始化TM、RM客户端
3.  扫描Bean，添加了全局事务注解的类和方法，添加对应的拦截器
4.  应用关闭，释放相关资源

## 2 全局事务拦截器GlobalTransactionalInterceptor ##

### 2.1 前言 ###

在上篇文档，我们分析了全局事务扫描器会为使用了分布式事务注解的Bean 添加拦截器，接下来我们分析下全局事务拦截器GlobalTransactionalInterceptor是如何进行拦截的？

### 2.2 方法代理拦截 ###

GlobalTransactionalInterceptor 实现了MethodInterceptor接口，MethodInterceptor是AOP中的拦截器，它拦截的目标是方法，实现该接口可以对需要增强的方法进行增强，扫描器扫描到标注了@GlobalTransaction的方法，就会添加这个AOP，所以执行目标方法时，实际是进入了拦截器的invoke方法。

在invoke方法中，会获取当前执行方法上的注解信息，然后开启全局事务：

@Override
        public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
        
        	// 目标的Class 类型
            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);
                // 获取@GlobalTransactional注解
                final GlobalTransactional globalTransactionalAnnotation =
                    getAnnotation(method, targetClass, GlobalTransactional.class);
                // 获取@GlobalLock 注解
                final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);
                boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes);
                if (!localDisable) {
        
                    if (globalTransactionalAnnotation != null) {
        
                    // 执行@GlobalTransactional全局事务
                        return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
                    } else if (globalLockAnnotation != null) {
        
                    // 执行@GlobalLock事务
                        return handleGlobalLock(methodInvocation, globalLockAnnotation);
                    }
                }
            }
    		// 没有注解，直接执行
            return methodInvocation.proceed();
        }

### 2.3 事务执行器 ###

在使用了@GlobalTransactional注解时，会进入到handleGlobalTransaction方法中，该方法首先会创建一个事务执行器，然后使用事务处理模板类，进入execute方法执行。  
![在这里插入图片描述][5840caaa522148a5a3d9d2f44b6996e8.jpeg]  
TransactionalExecutor 事务执行器接口，提供了两个方法，具体说明如下：

public interface TransactionalExecutor {
        
    
        /**
         * 执行实际业务逻辑
         */
        Object execute() throws Throwable;
    
        /**
         * 获取事务配置或其他属性
         */
        TransactionInfo getTransactionInfo();
    }

在TransactionalExecutor声明了一个内部枚举，用于标记事务异常时的状态：

public static enum Code {
        
        	// 开始失败
            BeginFailure,
            // 提交失败
            CommitFailure,
            // 回滚失败
            RollbackFailure,
            //  回滚完成
            RollbackDone,
            // 报告失败
            ReportFailure,
            // 回滚重试中
            RollbackRetrying;
        }

handleGlobalTransaction方法中，直接使用内部类创建了一个TransactionalExecutor实例，代码如下：

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());
                    }
    				// 将注解封装为成TransactionInfo对象
                    @Override
                    public TransactionInfo getTransactionInfo() {
        
                        // reset the value of timeout
                        // 超时时间
                        int timeout = globalTrxAnno.timeoutMills();
                        if (timeout <= 0 || timeout == DEFAULT_GLOBAL_TRANSACTION_TIMEOUT) {
        
                            timeout = defaultGlobalTransactionTimeout;
                        }
    				
                        TransactionInfo transactionInfo = new TransactionInfo();
                        transactionInfo.setTimeOut(timeout);
                        // 事务名=》test()
                        transactionInfo.setName(name());
                        // 传播行为=》REQUIRED
                        transactionInfo.setPropagation(globalTrxAnno.propagation());
                        transactionInfo.setLockRetryInternal(globalTrxAnno.lockRetryInternal());
                        transactionInfo.setLockRetryTimes(globalTrxAnno.lockRetryTimes());
                        // 回滚规则
                        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;
                    }
                });

最终当前TransactionalExecutor实例，包含的事务信息如下图所示：  
![在这里插入图片描述][5272e9f3cb61484699dd393bf14b02ef.png]

### 2.4 事务模板处理类 ###

TransactionalExecutor实例创建后，它包含了当前事务注解的信息和被代理执行的方法，接着就会调用执行模板TransactionalTemplate的execute方法，该方法是执行全局事务的核心方法，其中包含了AT模式下，两个阶段处理逻辑：

public Object execute(TransactionalExecutor business) throws Throwable {
        
        	// 注解信息
            TransactionInfo txInfo = business.getTransactionInfo();
            if (txInfo == null) {
        
                throw new ShouldNeverHappenException("transactionInfo does not exist");
            } else {
        
            	// 1. 从RootContext中获取xid，没有则返回NULL,
            	// 这里是全局事务发起方，所有没有，直接返回NULL
            	// 如果是被调用方，则会存在传递过来的xid,使用xid创建GlobalTransaction 对象
                GlobalTransaction tx = GlobalTransactionContext.getCurrent();
                // 2. 事务传播行为，默认为REQUIRED
                // REQUIRED：如果本来有事务，则加入该事务，如果没有事务，则创建新的事务
                Propagation propagation = txInfo.getPropagation();
                SuspendedResourcesHolder suspendedResourcesHolder = null;
    
                Object ex;
                try {
        
                    Object var6;
                    // 3. 根据不同的传播行为，执行不同的逻辑
                    switch(propagation) {
        
                    case NOT_SUPPORTED:
                        if (this.existingTransaction(tx)) {
        
                            suspendedResourcesHolder = tx.suspend();
                        }
    
                        var6 = business.execute();
                        return var6;
                    case REQUIRES_NEW:
                        if (this.existingTransaction(tx)) {
        
                            suspendedResourcesHolder = tx.suspend();
                            tx = GlobalTransactionContext.createNew();
                        }
                        break;
                    case SUPPORTS:
                        if (this.notExistingTransaction(tx)) {
        
                            var6 = business.execute();
                            return var6;
                        }
                    case REQUIRED:
                        break;
                    case NEVER:
                        if (this.existingTransaction(tx)) {
        
                            throw new TransactionException(String.format("Existing transaction found for transaction marked with propagation 'never', xid = %s", tx.getXid()));
                        }
    
                        var6 = business.execute();
                        return var6;
                    case MANDATORY:
                        if (this.notExistingTransaction(tx)) {
        
                            throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
                        }
                        break;
                    default:
                        throw new TransactionException("Not Supported Propagation:" + propagation);
                    }
    				// 4. 不存在全局事务信息，创建一个
                    if (tx == null) {
        
                        tx = GlobalTransactionContext.createNew();
                    }
    
                    GlobalLockConfig previousConfig = this.replaceGlobalLockConfig(txInfo);
    				// 5. 执行全局事务
                    try {
        
                    	// 5.1 开启全局事务
                        this.beginTransaction(txInfo, tx);
    
                        Object rs;
                        try {
        
                        	// 5.2 执行业务逻辑
                            rs = business.execute();
                        } catch (Throwable var17) {
        
                            ex = var17;
                            // 5.3 发生异常全局回滚
                            this.completeTransactionAfterThrowing(txInfo, tx, var17);
                            throw var17;
                        }
    					// 5.4 没有异常，提交全局事务
                        this.commitTransaction(tx);
                        ex = rs;
                    } finally {
        
                    	// 释放资源
                        this.resumeGlobalLockConfig(previousConfig);
                        this.triggerAfterCompletion();
                        this.cleanUp();
                    }
                } finally {
        
                    if (suspendedResourcesHolder != null) {
        
                        tx.resume(suspendedResourcesHolder);
                    }
    
                }
    
                return ex;
            }
        }

### 2.5 全局事务 ###

GlobalTransaction就是Seata 中的全局事务，默认实现类为DefaultGlobalTransaction，实现了其开始、提交、回滚等方法。

public interface GlobalTransaction {
        
    
        /**
         * Begin a new global transaction with default timeout and name.
         *
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         */
        void begin() throws TransactionException;
    
        /**
         * Begin a new global transaction with given timeout and default name.
         *
         * @param timeout Global transaction timeout in MILLISECONDS
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         */
        void begin(int timeout) throws TransactionException;
    
        /**
         * Begin a new global transaction with given timeout and given name.
         *
         * @param timeout Given timeout in MILLISECONDS.
         * @param name    Given name.
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         */
        void begin(int timeout, String name) throws TransactionException;
    
        /**
         * Commit the global transaction.
         *
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         */
        void commit() throws TransactionException;
    
        /**
         * Rollback the global transaction.
         *
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         */
        void rollback() throws TransactionException;
    
        /**
         * Suspend the global transaction.
         *
         * @return the SuspendedResourcesHolder which holds the suspend resources
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * @see SuspendedResourcesHolder
         */
        SuspendedResourcesHolder suspend() throws TransactionException;
    
        /**
         * Resume the global transaction.
         *
         * @param suspendedResourcesHolder the suspended resources to resume
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         * @see SuspendedResourcesHolder
         */
        void resume(SuspendedResourcesHolder suspendedResourcesHolder) throws TransactionException;
    
        /**
         * Ask TC for current status of the corresponding global transaction.
         *
         * @return Status of the corresponding global transaction.
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         * @see GlobalStatus
         */
        GlobalStatus getStatus() throws TransactionException;
    
        /**
         * Get XID.
         *
         * @return XID. xid
         */
        String getXid();
    
        /**
         * report the global transaction status.
         *
         * @param globalStatus global status.
         *
         * @throws TransactionException Any exception that fails this will be wrapped with TransactionException and thrown
         * out.
         */
        void globalReport(GlobalStatus globalStatus) throws TransactionException;
    
        /**
         * local status of the global transaction.
         *
         * @return Status of the corresponding global transaction.
         * @see GlobalStatus
         */
        GlobalStatus getLocalStatus();
    
        /**
         * get global transaction role.
         *
         * @return global transaction Role.
         * @see GlobalTransactionRole
         */
        GlobalTransactionRole getGlobalTransactionRole();
    
    }

### 2.6 总结 ###

1.  GlobalTransactionalInterceptor使用AOP对添加了事务注解的方法进行代理执行
2.  创建事务执行器TransactionalExecutor，实际封装了被代理的业务逻辑执行对象和注解信息
3.  调用TransactionalTemplate，进行分布式事务的两个阶段执行，发生异常回滚，无异常全局提交

[92452ed1d4c14bbea9a9e0b42cb42729.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/06613bd5c89042d3b82ff3410bbbc96e.png
[5840caaa522148a5a3d9d2f44b6996e8.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/715d0d96106e4d8ab555fa98810bdc0b.png
[5272e9f3cb61484699dd393bf14b02ef.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/aa1fb4da6383420fb195246986f6ff59.png