seata 1.2版本源码深入分析（一）

素颜马尾好姑娘i 2022-12-24 13:58 152阅读 0赞

Seata 是一款开源的分布式事务解决方案，致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式，为用户打造一站式的分布式解决方案。

seata AT  模式工程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

TC (Transaction Coordinator) - 事务协调者

维护全局和分支事务的状态，驱动全局事务提交或回滚。

TM (Transaction Manager) - 事务管理器

定义全局事务的范围：开始全局事务、提交或回滚全局事务。TM 是事务管理器，就是有@GlobalTransactional 注解的方法

RM (Resource Manager) - 资源管理器，参与者。

管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。RM 是资源管理器，是 TM 通过远程调用到的一方，RM 需要把数据源换成 seata 数据源进行代理。 masterDataSource 是真正连接数据库的目标数据源

![20201204142846991.png][]

DataSourceProxy 是 seata 实现了 DataSource 接口的数据源，持有了目标数据 源的引用

![20201204142936521.png][]

当远程调用到 RM 的时候，操作的是 seata 的代理数据源。

**Seata AT** **模式源码分析开始：此篇主要分析TM的生成和执行。**

**一、TM 代理的生成**

**1、**客户端源码入口：

TM 有一个@GlobalTransactional 注解的方法，那必然要生成代理的。 通过 springboot 的 spi，我们找到该注解的自动配置类。如图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

2、点击进入 SeataAutoConfiguration 类，创建生成代理的入口类：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

3、此类 GlobalTransactionScanner 是继承 AbstractAutoProxyCreator 的类，埋点生成代理，是 AOP 入口的一个父类，用来生成 bean 的代理。即当有@GlobalTransactional 注解的类实例化完成后，由于 BeanFactory 中有 GlobalTransactionScanner 这个 BeanPostProcessor 类型的类，所以代码会执 行到这里来，如图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

4、接着调用到父类  AbstractAutoProxyCreator

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

5、这个 wrapIfNecessary 是一个模板设计模式，是一个钩子方法，会调用到子类   GlobalTransactionScanner 中 来。关键代码在源码中写了注释

/**
         * bean实例化之后到这个后置处理器方法
         * aop入口
         * AbstractAutoProxyCreator 类的 postProcessAfterInitialization 方法会调到此方法
         * 此方法对spring所有的bean进行拦截
         * @param bean
         * @param beanName
         * @param cacheKey
         * @return
         */
        @Override
        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 {//默认AT模式
                        Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
                        Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);
    
                        //判断类和接口上是否加了全局事务注解 @GlobalTransactional 等
                        // 如果没有类中方法上面没有@GlobalTransactional 注解，则不需要代理直接返回 bean
                        if (!existsAnnotation(new Class[]{serviceInterface})
                            && !existsAnnotation(interfacesIfJdk)) {
                            return bean;
                        }
    
                        /**
                         * 这是生成代理后的 advice 增强类，最终用代理对象调用方法时最终会执行到这个类中的 invoke 中来
                         * 这里只支持cglib 生成动态代理类；因为GlobalTransactionalInterceptor实现了 MethodInterceptor接口的 invoke方法
                         * 而invoke方法里的内容就是相当于在 加了  @GlobalTransactional 注解方法里加了invoke 方法里的内容，其包含全局事务的开启、提交和回滚业务。
                         */
                        if (interceptor == null) {
                            interceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
                            ConfigurationFactory.getInstance().addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (ConfigurationChangeListener) interceptor);
                        }
                    }
    
                    LOGGER.info("Bean[{}] with name [{}] would use interceptor [{}]", bean.getClass().getName(), beanName, interceptor.getClass().getName());
                    if (!AopUtils.isAopProxy(bean)) {
                        //生成代理的核心在这里，调用父类的方法，父类又会钩子到子类中获取类的 advice，如果有则生成代理
                        //调用父类的方法，正式生成代理类
                       
                        bean = super.wrapIfNecessary(bean, beanName, cacheKey);
                    } else {
                        AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
                        Advisor[] advisor = 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);
            }
        }

6、进入 super.wrapIfNecessary(bean, beanName, cacheKey);调用时，会判断类是否有切面，如果有切面则生成代理， 代码如下

protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
    		if (beanName != null && this.targetSourcedBeans.contains(beanName)) {
    			return bean;
    		}
    		if (Boolean.FALSE.equals(this.advisedBeans.get(cacheKey))) {
    			return bean;
    		}
    		if (isInfrastructureClass(bean.getClass()) || shouldSkip(bean.getClass(), beanName)) {
    			this.advisedBeans.put(cacheKey, Boolean.FALSE);
    			return bean;
    		}
    
    		// Create proxy if we have advice.调用时，会判断类是否有切面，如果有切面则生成代理
    		Object[] specificInterceptors = getAdvicesAndAdvisorsForBean(bean.getClass(), beanName, null);
    		if (specificInterceptors != DO_NOT_PROXY) {
    			this.advisedBeans.put(cacheKey, Boolean.TRUE);
    			Object proxy = createProxy(
    					bean.getClass(), beanName, specificInterceptors, new SingletonTargetSource(bean));
    			this.proxyTypes.put(cacheKey, proxy.getClass());
    			return proxy;
    		}
    
    		this.advisedBeans.put(cacheKey, Boolean.FALSE);
    		return bean;
    	}

7、钩子方法进入子类GlobalTransactionScanner，获取切面：直接返回了 interceptor = new GlobalTransactionalInterceptor(failureHandlerHook);这个对象的数组，返 回了值就会生成代理对象。并且用代理对象执行的时候，根据 aop 的调用规则，最终会是一个链式调用，直接执行到 advice 中，那么代理对象调用的时候会走到 GlobalTransactionalInterceptor 的 invoke 方法中。

/**
         * 父类 AbstractAutoProxyCreator 正式生成代理的时候调用此方法，获取 interceptor 对象，继而生成有全局事务功能的代理类
         */
        @Override
        protected Object[] getAdvicesAndAdvisorsForBean(Class beanClass, String beanName, TargetSource customTargetSource)
            throws BeansException {
            return new Object[]{interceptor};
        }

如上所述代理对象就生成了。

二、**TM** **代理的执行**

1、当用代理对象调用方法时，最终会执行到 GlobalTransactionalInterceptor 的 invoke 中，代码 如下：

@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);
            final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
    
            final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, GlobalTransactional.class);
            final GlobalLock globalLockAnnotation = getAnnotation(method, GlobalLock.class);
            //再次判断是否加注解
            if (!disable && globalTransactionalAnnotation != null) {
                //如果方法上面有注解，则会执行到这里,核心方法，进入
                return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
            } else if (!disable && globalLockAnnotation != null) {
                return handleGlobalLock(methodInvocation);
            } else {
                //如果没有全局事务注解，直接调用目标方法
                return methodInvocation.proceed();
            }
        }

2、handleGlobalTransaction 进入：核心代码 execute 方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5][]

3、进入 execute方法：

public Object execute(TransactionalExecutor business) throws Throwable {
            // 1 get transactionInfo 取事务信息，上一步封装好的，可以去看一下
            TransactionInfo txInfo = business.getTransactionInfo();
            if (txInfo == null) {
                throw new ShouldNeverHappenException("transactionInfo does not exist");
            }
            // 1.1 get or create a transaction 开启事务
            GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
    
            // 1.2 Handle the Transaction propatation and the branchType
            Propagation propagation = txInfo.getPropagation();
            SuspendedResourcesHolder suspendedResourcesHolder = null;
            //事务的传播机制
            try {
                switch (propagation) {
                    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);
    
                    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();
                }
            } finally {
                tx.resume(suspendedResourcesHolder);
            }
    
        }

业务方法调到这里ctrl+t命令： rs = business.execute();

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6][]

4、进入 beginTransaction(txInfo, tx);

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7][]

5、进入：tx.begin(txInfo.getTimeOut(), txInfo.getName());

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8][]

@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();//全局事务id现在为空，如果不为空报错
            if (RootContext.getXID() != null) {
                throw new IllegalStateException();
            }
            //创建返回全局事务id（从TC服务端返回），连接服务端开启全局事务，核心代码，进入
            xid = transactionManager.begin(null, null, name, timeout);
            status = GlobalStatus.Begin;
            RootContext.bind(xid);//放在本地ThreadLocal中
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Begin new global transaction [{}]", xid);
            }
    
        }

绑定xid:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 9][]

继续：

@Override
        public Object put(String key, Object value) {
            return threadLocal.get().put(key, value);
        }

6、进入 xid = transactionManager.begin(null, null, name, timeout);

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 10][]

@Override
        public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)
            throws TransactionException {
            GlobalBeginRequest request = new GlobalBeginRequest();
            request.setTransactionName(name);
            request.setTimeout(timeout);
            //远程调用通知seata服务端，在下游配置中心取服务端相关ip等信息
            GlobalBeginResponse response = (GlobalBeginResponse)syncCall(request);
            if (response.getResultCode() == ResultCode.Failed) {
                throw new TmTransactionException(TransactionExceptionCode.BeginFailed, response.getMsg());
            }
            return response.getXid();
        }

7、进入  (GlobalBeginResponse)syncCall(request);

![20201204154710461.png][]

进入

![20201204154728174.png][]

进入:

@Override
        public Object sendMsgWithResponse(Object msg, long timeout) throws TimeoutException {
            //封装信息发送服务端，为netty nio框架基础,负载到服务器上
            String validAddress = loadBalance(getTransactionServiceGroup());
            Channel channel = clientChannelManager.acquireChannel(validAddress);
            Object result = super.sendAsyncRequestWithResponse(validAddress, channel, msg, timeout);
            return result;
        }

看到这里大家应该一目了然，这里走到尽头就到了seata服务端，后续我们会详细分析。下篇我们分析RM的一、二阶段执行，敬请期待。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/80423289a53b4130981fbe71c9df2953.png
[20201204142846991.png]: /images/20221120/57b2868615474ce499d4866bc46490d1.png
[20201204142936521.png]: https://img-blog.csdnimg.cn/20201204142936521.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20201204114157917.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201204144754987.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/202012041450178.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20201204145047205.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20201204153356132.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20201205142236710.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20201204154311149.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/2020120415441657.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20201205141804685.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/20201204154556212.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[20201204154710461.png]: https://img-blog.csdnimg.cn/20201204154710461.png
[20201204154728174.png]: https://img-blog.csdnimg.cn/20201204154728174.png