Seata分布式事务源码分析

Posted 2022-12-12 赵广陆

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Seata分布式事务源码分析相关的知识，希望对你有一定的参考价值。

1 Seata常见注

@GlobalTransactional ：注解在业务方法上用来开启全局事务，可以自定义超时时间、全局事务的名字、回滚时调用的类等。
@GlobalLock ：声明事务仅执行在本地RM中，但是本次事务确保在更新状态下的操作记录不会被其他全局事务操作。即将本地事务的执行纳入seata分布式事务的管理，一起竞争全局锁，保证全局事务在执行的时候，本地业务不可以操作全局事务中的记录。
GlobalTransactionalInterceptor：实现了spring的 MethondInterceptor，声明一个环绕通知。拦截器主要就是判断是否有上面两个注解，并调用相应处理方法。
MethodDesc：持有方法和方法上全局事务注解，model类。
TccActionInterceptor：是 TCC 模式下独有的方法拦截器，也实现了 spring 的MethondInterceptor。

1.1 环绕通知(拦截器)分析

spring是java开发必备框架，类一般都是通过spring进行管理。那spring之外的框架，尤其是想使用spring代理的类的框架，如何借助spring的扩展点来操作bean？
这篇文章主要分析 Seata如何借助 spring扩展点对代理的bean进行操作，生成Seata想要的数据库代理类和对被全局事务注解的bean 织入不同事务模式对应的 advisor 实现类。

目前的seata版本有两种 Advisor，一个是GlobalTransactionalInterceptor，一个是TccActionInterceptor。
GlobalTranscationInterceptor是在scanner检测到不是TCC模式下的bean，并且方法上有@GlobalTransactional注解或者@GlobalLock注解才织入的。
TccActionInterceptor是当scanner检测到bean的父类接口中有@LocalTCC注解，才织入的。

GlobalTransactionalInterceptor 拦截器有三个成员变量：TransactionnalTemplate、GlobalLockTemplate、FailureHandler，分别处理有全局事务注解的方法、和全局锁注解的方法和失败的情况。

拦截器invoke()方法主要逻辑是根据注解的类型找到对应的模板进行执行，用到了模板模式，具体模板的逻辑其他模块进行解析。

1.2 全局事务扫描类分析

在GlobalTransactionalScanner中，通过利用spring的扩展点，对bean进行自己想要的处理。

这个类继承了 AbstractAutoProxyCreator并实现了三个类InitializingBean、ApplicationContextAware、DisposableBean。

实现的类主要是为了获取spring的上下文 ApplicationContext 和在实例化完成时进行 TM 、RM的初始化，并注册销毁时调用的钩子，比较简单。这里有一个可以优化的地方，就是当项目中没有扫描到存在@GlobalTransactional注解的方法时，可以不用进行TM的初始化。

重点在继承AbstractAutoProxyCreate，继承这个类，获得了两次操作bean的机会。
第一次通过使用 postProcessAfterInitialization()，在bean被创建后，属性注入之前，对bean进行处理。

这里主要就是对 bean类型为为DataSource子类的时候，进行bean的代理，把原有的DataSource对象换成自己想要的DataSourceProxy对象。这里可能会被重复调用，所以多了一个判断，如果被代理过了，则交给父类处理。

wrapIfNecessary()的调用时机是在postProcessAfterInitialization()之后

主要就是获取bean的类型，根据bean的类型来实例化对应的adsivor。

这里有一个问题，被远程调用的方法没有注解，如何纳入全局事务?
猜测：在进行远程调用的时候，带上了事务id，远程业务模块拦截器通过事务id生成本地事务。

这里当不是aop代理的bean时，手动调用父类方法，父类会调用 getAdvicesAndAdvisorsForBean()来获取上面实例化好的interceptor。如果是aop代理，就把实例化好的interceptor拦截器增加到Advisor[]数组中，放置在第一位。
spring在执行完 wrapIfNecessery()后，会继续进行aop代理类的实例化。

1.3 总结

Seata中的bean有三种：normalTccBean、localTccBean(被@LocalTcc注解)、normalBean（被@GlobalTransactional或者@GlobalLock注解），其中normalTccBean是不会被 wrap 的，其他两个会被代理，通过wrapIfNecessery()，织入advisor。

更多好文请关注微信公众号：我手一杯

2 GlobalLock注解使用场景及源码分析

2.1 GlobalLock源码分析

// 声明事务仅在单个本地RM中执行
// 但事务需要确保要更新（或选择更新）的记录不在全局事务中
// 在上述情况下，使用此注解而不是@GlobalTransaction将有助于提高性能。
// @see io.seata.spring.annotation.GlobalTransactionScanner#wrapIfNecessary(Object, String, Object)用于TM、GlobalLock和TCC模式的扫描器
// @see io.seata.spring.annotation.GlobalTransactionalInterceptor#handleGlobalLock(MethodInvocation)@GlobalLock的拦截器
// @see io.seata.spring.annotation.datasource.SeataAutoDataSourceProxyAdvice#invoke(MethodInvocation) GlobalLockLogic和AT/XA模式的拦截器
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD,ElementType.TYPE)
@Inherited
public @interface GlobalLock 
    /**
     * 自定义全局锁重试间隔（单位：毫秒）
     * 您可以使用它覆盖“client.rm.lock.retryInterval”的全局配置，默认10
     * 注意：0或负数将不起作用（这意味着返回到全局配置）
     */
    int lockRetryInternal() default 0;

    /**
     * 自定义全局锁重试次数
     * 您可以使用它覆盖“client.rm.lock.retryTimes”的全局配置，默认30
     * 注：负数无效（这意味着返回全局配置
     */
    int lockRetryTimes() default -1;

源码注释大概意思：对于某条数据进行更新操作，如果全局事务正在进行，当某个本地事务需要更新该数据时，需要使用@GlobalLock确保其不会对全局事务正在操作的数据进行修改。

2.2 问题场景

我们参考下图，搭建一个测试案例：

2.2.1 编写代码

首先编写一个全局事务，调用订单服务下订单，扣除余额-1。

    @GlobalTransactional(rollbackFor = Throwable.class, timeoutMills = 300000)
    public void test() throws InterruptedException 
        log.info("Assign Service Begin ... xid: " + RootContext.getXID() + "\\n");
        //1.创建账户 扣款
        AccountTbl accountTbl = accountTblMapper.selectById(11111111);
        AccountTbl accountTbl1 = accountTbl.setMoney(accountTbl.getMoney() - 1);
        accountTblMapper.updateById(accountTbl1);
        //2.创建订单
        orderClint.insert(accountTbl.getUserId() + "", "iphone11", 1 + "");
        // 休眠5秒
        TimeUnit.SECONDS.sleep(5);
        int i = 5 / 0;
         //模拟异常

在编写一个本地@Transactional事务，直接扣除余额-1。

    @GetMapping("/GlobalLock")
    @Transactional
    public Object GlobalLock() 
        AccountTbl accountTbl = accountTblMapper.selectById(11111111);
        AccountTbl accountTbl1 = accountTbl.setMoney(accountTbl.getMoney()-1);
        accountTblMapper.updateById(accountTbl1);
        return "成功执行！！！";

2.2.2 测试

数据库修改余额为100 元，然后测试全局事务接口，发现异常时能正常全局回滚。

在执行全局事务的过程中，调用GlobalLock接口，修改数据，因为全局事务接口中休眠了5秒，所以需要在访问全局接口打印全局事务日志后，快速访问GlobalLock接口。

这个时候会发现，全局事务第二阶段回滚失败，并一直在重试：

原因分析：因为在全局事务执行的过程中，一阶段会直接提交本地事务，其他本地事务可直接修改该数据，所以会导致全局事务二阶段回滚时，发现数据被修改过，认为数据已经脏了，回滚失败。

2.2.3 解决方案

解决方案：

手动处理：锁表，然后直接将数据修改为正常状态，但是这种比较麻烦，需要梳理脏数据的原因，也影响业务实际运行
提前预防：使用@GlobalLock，在执行本地事务时，去获取该数据的全局锁，如果获取不到，说明该数据正在被全局事务执行，可以进行重试获取。

在本地修改事务上加上@GlobalLock，配置重试间隔为100ms，次数为100次，说明在10S内会不断重试获取全局锁，如果该记录在全局事务中，则会失败：

    @GlobalLock(lockRetryInternal = 100, lockRetryTimes = 100)
    @GetMapping("/GlobalLock")
    @Transactional
    public Object GlobalLock() 
        AccountTbl accountTbl = accountTblMapper.selectById(11111111);
        AccountTbl accountTbl1 = accountTbl.setMoney(accountTbl.getMoney() - 1);
        accountTblMapper.updateById(accountTbl1);
        return "成功执行！！！";

2.2.4 注意事项

在使用@GlobalLock注解的时候，我们需要更新之前，在查询方法中添加排它锁，比如根据ID 查询时，需要如下SQL 书写：

    <select id="selectById" parameterType="integer" resultType="com.hnmqet.demo01.entity.AccountTbl">
        SELECT id,user_id,money FROM account_tbl WHERE id=#id FOR UPDATE
    </select>

这是因为，只有添加了 FOR UPDATE，Seata 才会进行创建重试的执行器，这样事务失败时，会释放本地锁，等待一定时间再重试。如果不添加，则会一直占有本地锁，全局事务回滚需要本地锁，则全局事务就只能等@GlobalLock事务超时失败才能拿到本地锁释放全局锁，造成@GlobalLock永远获取不到全局锁。

2.3 源码分析

2.3.1. 进入拦截器

之前分析过GlobalTransactionScanner（全局事务扫描器）会扫描@GlobalLock、@GlobalTransactional注解标识的方法，并为其添加GlobalTransactionalInterceptor（全局事务拦截器）。

所以@GlobalLock标注的方法执行时，会进入到GlobalTransactionalInterceptor的invoke方法，获取@GlobalLock注解，然后进入到handleGlobalLock方法处理。

handleGlobalLock方法会创建一个GlobalLockExecutor匿名内部类，然后调用GlobalLockTemplate 的execute方法：

    Object handleGlobalLock(final MethodInvocation methodInvocation, final GlobalLock globalLockAnno) throws Throwable 
        return this.globalLockTemplate.execute(new GlobalLockExecutor() 
            public Object execute() throws Throwable 
                return methodInvocation.proceed();
            
            public GlobalLockConfig getGlobalLockConfig() 
            	// 获取@GlobalLock 注解上的配置
                GlobalLockConfig config = new GlobalLockConfig();
                config.setLockRetryInternal(globalLockAnno.lockRetryInternal());
                config.setLockRetryTimes(globalLockAnno.lockRetryTimes());
                return config;
            
        );

GlobalLockTemplate模板类只有一个方法，处理逻辑也很简单，就是将注解配置塞入线程中，结束后清理：

    public Object execute(GlobalLockExecutor executor) throws Throwable 
        boolean alreadyInGlobalLock = RootContext.requireGlobalLock();
        if (!alreadyInGlobalLock) 
            RootContext.bindGlobalLockFlag();
        
        // 将注解配置塞入ThreadLocal中
        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();

2.3.2 进入数据源代理

在执行业务逻辑时，因为配置了数据源代理，SQL 操作都会进入到代理数据源中，大概流程为PreparedStatementProxy.execute=>ExecuteTemplate.execute=>Executor.executor。

因为我们根据ID 查询数据时加了 FOR UPDATE（排它锁），所以执行器为SelectForUpdateExecutor，在这个执行方法中，就会进行全局锁的获取，这个时候会遇到以下几种情况：

获取到全局锁，则正常执行，因为加了排它锁，其他事务都会被隔离，得等待当前事务执行完成
被全局事务占有全局锁和排它锁，则会等待全局一阶段事务提交释放本地锁，GlobalLock获取到本地锁后，等待全局事务提交，释放全局锁后，再执行，
如果全局失败，回滚时需要排它锁，这个时候，GlobalLock因为没有获取到全局锁抛出异常，会在异常中进行事务回滚，休眠一定时间，这个时候会让出排它锁，全局获取到排它锁后再进行全局回滚成功释放全局锁，GlobalLock在重试过程中，获取到全局锁，则成功执行，做到了很好的事务隔离性。

    @Override
    public T doExecute(Object... args) throws Throwable 
    	// 1. 获取数据库连接
        Connection conn = statementProxy.getConnection();
        // 2. 获取数据库元数据
        DatabaseMetaData dbmd = conn.getMetaData();
        T rs;
        Savepoint sp = null;
        boolean originalAutoCommit = conn.getAutoCommit();
        try 
            if (originalAutoCommit) 
                /*
                 * 为了在全局锁检查期间保持本地数据库锁
                 * 如果原始自动提交为true，则首先将自动提交值设置为false
                 */
                conn.setAutoCommit(false);
             else if (dbmd.supportsSavepoints()) 
                /*
                 * 为了在全局锁冲突时释放本地数据库锁
                 * 如果原始自动提交为false，则创建一个保存点，然后使用此处的保存点释放db
                 * 如有必要，在全局锁定检查期间锁定
                 */
                sp = conn.setSavepoint();
             else 
                throw new SQLException("not support savepoint. please check your db version");
            
			// 3. 创建一个锁重试控制器
            LockRetryController lockRetryController = new LockRetryController();
            ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
            // 4. SELECT id FROM account_tbl WHERE id = ? FOR UPDATE
            // 
            String selectPKSQL = buildSelectSQL(paramAppenderList);
            while (true) 
                try 
                    // #870
                    // 
                    rs = statementCallback.execute(statementProxy.getTargetStatement(), args);

                    // 尝试获取选定行的全局锁
                    // 获取主键列及值
                    TableRecords selectPKRows = buildTableRecords(getTableMeta(), selectPKSQL, paramAppenderList);
                    // 构建全局锁Key :account_tbl:11111111
                    String lockKeys = buildLockKey(selectPKRows);
                    if (StringUtils.isNullOrEmpty(lockKeys)) 
                        break;
                    

                    if (RootContext.inGlobalTransaction() || RootContext.requireGlobalLock()) 
                        // 在@GlobalTransactional或@GlobalLock下做同样的事情
                        // 这里只检查全局锁
                        statementProxy.getConnectionProxy().checkLock(lockKeys);
                     else 
                        throw new RuntimeException("Unknown situation!");
                    
                    break;
                 catch (LockConflictException lce) 
                	// 如锁被占用，会抛出锁冲突异常 ：LockConflictException
                	// 直接回滚，释放本地锁
                    if (sp != null) 
                        conn.rollback(sp);
                     else 
                        conn.rollback();
                    
                    // 触发重试，线程睡眠设置的时间，超过重试此时，则会抛出LockWaitTimeoutException 异常
                    lockRetryController.sleep(lce);
                
            
         finally 
            if (sp != null) 
                try 
                    if (!JdbcConstants.ORACLE.equalsIgnoreCase(getDbType())) 
                        conn.releaseSavepoint(sp);
                    
                 catch (SQLException e) 
                    LOGGER.error(" release save point error.", getDbType(), e);
                
            
            if (originalAutoCommit) 
                conn.setAutoCommit(true);
            
        
        return rs;

2.3.3 更新数据

在通过 FOR UPDATE 查询到数据后，再更新当前数据，因为查询和修改在一个@Transactional方法里，所以他们是一个事务，在查询的时候添加了排它锁，并且获取到了全局锁，才会执行到更新方法。

FOR UPDATE 获取到全局锁后，进入到业务的更新操作，这里和一阶段执行本地事务完全一致，之前分析过，就不赘述了。

以上是关于Seata分布式事务源码分析的主要内容，如果未能解决你的问题，请参考以下文章

seata TM源码分析

分布式事务(Seata)原理详解篇，建议收藏

seata 分布式事务框架seata1.3 AT及XA模式实例演示