• Spring-Cloud-Alibaba-SEATA源码解析(三)（客户端）

    前言

            在上一篇，分析了seata at模式分布式事务的执行流程，可以看到它和spring声明式事务的过程非常相似，剩下的问题是分支事务是如何提交和回滚的，seata中的全局锁是什么？答案是seata代理了我们的数据源，这个代理数据源对数据源的某些方法进行了增强，加入了申请全局锁，生成undo_log，注册分支事务等逻辑。

    全局锁

    全局锁主要用来实现全局事务的写隔离：

    • 一阶段本地事务提交前，需要确保先拿到 全局锁 。
    • 拿不到 全局锁 ，不能提交本地事务。
    • 拿 全局锁 的尝试被限制在一定范围内，超出范围将放弃，并回滚本地事务，释放本地锁。

    以一个示例来说明：

            两个全局事务 tx1 和 tx2，分别对 a 表的 m 字段进行更新操作，m 的初始值 1000。

            tx1 先开始，开启本地事务，拿到本地锁，更新操作 m = 1000 - 100 = 900。本地事务提交前，先拿到该记录的 全局锁 ，本地提交释放本地锁。 tx2 后开始，开启本地事务，拿到本地锁，更新操作 m = 900 - 100 = 800。本地事务提交前，尝试拿该记录的 全局锁 ，tx1 全局提交前，该记录的全局锁被 tx1 持有，tx2 需要重试等待 全局锁 。

    tx1 二阶段全局提交，释放 全局锁 。tx2 拿到 全局锁 提交本地事务。

            如果 tx1 的二阶段全局回滚，则 tx1 需要重新获取该数据的本地锁，进行反向补偿的更新操作，实现分支的回滚。

            此时，如果 tx2 仍在等待该数据的 全局锁，同时持有本地锁，则 tx1 的分支回滚会失败。分支的回滚会一直重试，直到 tx2 的 全局锁 等锁超时，放弃 全局锁 并回滚本地事务释放本地锁，tx1 的分支回滚最终成功。

    因为整个过程 全局锁 在 tx1 结束前一直是被 tx1 持有的，所以不会发生 脏写 的问题。

    以上内容截取自seata官网。

    下面进入代理数据源的源码，入口是一个springboot自动配置类：SeataDataSourceAutoConfiguration

    1. @ConditionalOnBean(DataSource.class)
    2. @ConditionalOnExpression("${seata.enable:true} && ${seata.enableAutoDataSourceProxy:true} && ${seata.enable-auto-data-source-proxy:true}")
    3. public class SeataDataSourceAutoConfiguration {
    4.  
    5.     /**
    6.      * The bean seataDataSourceBeanPostProcessor.
    7.      */
    8.     @Bean(BEAN_NAME_SEATA_DATA_SOURCE_BEAN_POST_PROCESSOR)
    9.     @ConditionalOnMissingBean(SeataDataSourceBeanPostProcessor.class)
    10.     public SeataDataSourceBeanPostProcessor seataDataSourceBeanPostProcessor(SeataProperties seataProperties) {
    11.         return new SeataDataSourceBeanPostProcessor(seataProperties.getExcludesForAutoProxying(), seataProperties.getDataSourceProxyMode());
    12.     }
    13.  
    14.     /**
    15.      * The bean seataAutoDataSourceProxyCreator.
    16.      */
    17.     @Bean(BEAN_NAME_SEATA_AUTO_DATA_SOURCE_PROXY_CREATOR)
    18.     @ConditionalOnMissingBean(SeataAutoDataSourceProxyCreator.class)
    19.     public SeataAutoDataSourceProxyCreator seataAutoDataSourceProxyCreator(SeataProperties seataProperties) {
    20.         return new SeataAutoDataSourceProxyCreator(seataProperties.isUseJdkProxy(),
    21.             seataProperties.getExcludesForAutoProxying(), seataProperties.getDataSourceProxyMode());
    22.     }
    23. }

            首先是SeataDataSourceBeanPostProcessor，bean的后置处理器，重写了bean的初始化后置处理方法：postProcessAfterInitialization，在这个方法中seata会为Spring容器中的数据源bean创建代理数据源。

    1.     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
    2.         if (bean instanceof DataSource) {
    3.             //不属于排除的数据源（可在配置文件中设置）
    4.             if (!excludes.contains(bean.getClass().getName())) {
    5.                 //创建代理数据源，并和原数据源做映射
    6.                 DataSourceProxyHolder.get().putDataSource((DataSource) bean, dataSourceProxyMode);
    7.             }
    8.  
    9.             //If is SeataDataSourceProxy, return the original data source.
    10.             if (bean instanceof SeataDataSourceProxy) {
    11.                 LOGGER.info("Unwrap the bean of the data source," +
    12.                     " and return the original data source to replace the data source proxy.");
    13.                 return ((SeataDataSourceProxy) bean).getTargetDataSource();
    14.             }
    15.         }
    16.         return bean;
    17.     }
    18.  
    19.  
    20.     public SeataDataSourceProxy putDataSource(DataSource dataSource, BranchType dataSourceProxyMode) {
    21.         DataSource originalDataSource;
    22.         if (dataSource instanceof SeataDataSourceProxy) {
    23.             SeataDataSourceProxy dataSourceProxy = (SeataDataSourceProxy) dataSource;
    24.  
    25.             //只会创建一次代理
    26.             if (dataSourceProxyMode == dataSourceProxy.getBranchType()) {
    27.                 return (SeataDataSourceProxy) dataSource;
    28.             }
    29.  
    30.             //Get the original data source.
    31.             originalDataSource = dataSourceProxy.getTargetDataSource();
    32.         } else {
    33.             originalDataSource = dataSource;
    34.         }
    35.         //从映射map中获取代理数据源
    36.         SeataDataSourceProxy dsProxy = dataSourceProxyMap.get(originalDataSource);
    37.         //没有则创建
    38.         if (dsProxy == null) {
    39.             synchronized (dataSourceProxyMap) {
    40.                 dsProxy = dataSourceProxyMap.get(originalDataSource);
    41.                 if (dsProxy == null) {
    42.                     //创建代理数据源
    43.                     dsProxy = createDsProxyByMode(dataSourceProxyMode, originalDataSource);
    44.                     dataSourceProxyMap.put(originalDataSource, dsProxy);
    45.                 }
    46.             }
    47.         }
    48.         return dsProxy;
    49.     }
    50.  
    51.  
    52.     private SeataDataSourceProxy createDsProxyByMode(BranchType mode, DataSource originDs) {
    53.         //默认是AT模式，创建DataSourceProxy对象
    54.         return BranchType.XA == mode ? new DataSourceProxyXA(originDs) : new DataSourceProxy(originDs);
    55.     }

    在创建代理数据源的过程中，seata客户端会向TC注册RM：

    1.     public DataSourceProxy(DataSource targetDataSource, String resourceGroupId) {
    2.         if (targetDataSource instanceof SeataDataSourceProxy) {
    3.             LOGGER.info("Unwrap the target data source, because the type is: {}", targetDataSource.getClass().getName());
    4.             targetDataSource = ((SeataDataSourceProxy) targetDataSource).getTargetDataSource();
    5.         }
    6.         this.targetDataSource = targetDataSource;
    7.         init(targetDataSource, resourceGroupId);
    8.     }
    9.  
    10.  
    11.     private void init(DataSource dataSource, String resourceGroupId) {
    12.         this.resourceGroupId = resourceGroupId;
    13.         try (Connection connection = dataSource.getConnection()) {
    14.             //获取数据库url，类型，并填充DataSourceProxy属性
    15.             jdbcUrl = connection.getMetaData().getURL();
    16.             dbType = JdbcUtils.getDbType(jdbcUrl);
    17.             if (JdbcConstants.ORACLE.equals(dbType)) {
    18.                 userName = connection.getMetaData().getUserName();
    19.             }
    20.         } catch (SQLException e) {
    21.             throw new IllegalStateException("can not init dataSource", e);
    22.         }
    23.         //向TC注册RM，包含了本地数据库连接信息
    24.         //依然是通过netty客户端发送
    25.         DefaultResourceManager.get().registerResource(this);
    26.         if (ENABLE_TABLE_META_CHECKER_ENABLE) {
    27.             tableMetaExcutor.scheduleAtFixedRate(() -> {
    28.                 try (Connection connection = dataSource.getConnection()) {
    29.                     TableMetaCacheFactory.getTableMetaCache(DataSourceProxy.this.getDbType())
    30.                         .refresh(connection, DataSourceProxy.this.getResourceId());
    31.                 } catch (Exception ignore) {
    32.                 }
    33.             }, 0, TABLE_META_CHECKER_INTERVAL, TimeUnit.MILLISECONDS);
    34.         }
    35.  
    36.         //设置默认分支事务类型为AT模式
    37.         RootContext.setDefaultBranchType(this.getBranchType());
    38.     }

            然后是SeataAutoDataSourceProxyCreator这个bean，它继承了AbstractAutoProxyCreator，这个类大家应该很熟悉了，sping aop的核心类，用来为bean创建动态代理。SeataAutoDataSourceProxyCreator重写了父类的getAdvicesAndAdvisorsForBean和shouldSkip方法，分别用来从容器中寻找对应数据源bean Datasource的通知类和在创建动态代理时判断是否跳过：

    1. public class SeataAutoDataSourceProxyCreator extends AbstractAutoProxyCreator {
    2.     private static final Logger LOGGER = LoggerFactory.getLogger(SeataAutoDataSourceProxyCreator.class);
    3.     private final List<String> excludes;
    4.     private final Advisor advisor;
    5.  
    6.     public SeataAutoDataSourceProxyCreator(boolean useJdkProxy, String[] excludes, String dataSourceProxyMode) {
    7.         this.excludes = Arrays.asList(excludes);
    8.         this.advisor = new DefaultIntroductionAdvisor(new SeataAutoDataSourceProxyAdvice(dataSourceProxyMode));
    9.         setProxyTargetClass(!useJdkProxy);
    10.     }
    11.  
    12.     @Override
    13.     protected Object[] getAdvicesAndAdvisorsForBean(Class<?> beanClass, String beanName, TargetSource customTargetSource) throws BeansException {
    14.         if (LOGGER.isInfoEnabled()) {
    15.             LOGGER.info("Auto proxy of [{}]", beanName);
    16.         }
    17.         //获取能够匹配到的通知器，这里直接返回自己的Advisor也就是AdvisorDefaultIntroductionAdvisor，省去匹配过程 
    18.         return new Object[]{advisor};
    19.     }
    20.  
    21.     @Override
    22.     protected boolean shouldSkip(Class<?> beanClass, String beanName) {
    23.         //判断是否要跳过创建代理的过程，这里直接判断是否是DataSource类型
    24.         //如果是DataSource类型则需要为止创建代理
    25.         return !DataSource.class.isAssignableFrom(beanClass) ||
    26.             SeataProxy.class.isAssignableFrom(beanClass) ||
    27.             excludes.contains(beanClass.getName());
    28.     }
    29. }

            最后orm框架在执行sql时，会拿到spring容器中的数据源，并执行获取数据库连接connection的方法，会来到拦截器SeataAutoDataSourceProxyAdvice的invoke：

    1.     public Object invoke(MethodInvocation invocation) throws Throwable {
    2.         if (!RootContext.requireGlobalLock() && dataSourceProxyMode != RootContext.getBranchType()) {
    3.             return invocation.proceed();
    4.         }
    5.  
    6.         Method method = invocation.getMethod();
    7.         Object[] args = invocation.getArguments();
    8.         //要执行的目标方法
    9.         Method m = BeanUtils.findDeclaredMethod(dataSourceProxyClazz, method.getName(), method.getParameterTypes());
    10.         //判断是否是datasource的方法
    11.         if (m != null && DataSource.class.isAssignableFrom(method.getDeclaringClass())) {
    12.             //从DataSourceProxyHolder中获取对应数据源的映射过的代理数据源DataSourceProxy
    13.             //这里代理数据源DataSourceProxy在spinrg加载容器时已经创建完毕
    14.             SeataDataSourceProxy dataSourceProxy = DataSourceProxyHolder.get().putDataSource((DataSource) invocation.getThis(), dataSourceProxyMode);
    15.             //执行代理数据源DataSourceProxy的相对应的方法
    16.             return m.invoke(dataSourceProxy, args);
    17.         } else {
    18.             return invocation.proceed();
    19.         }
    20.     }

    DataSourceProxy代理过程：

    DataSourceProxy-->ConnectionProxy-->PreparedStatementProxy

    最终在执行sql时，PreparedStatementProxy对执行sql方法execute进行代理PreparedStatementProxy#execute：

    1.     public boolean execute() throws SQLException {
    2.         return ExecuteTemplate.execute(this, (statement, args) -> statement.execute());
    3.     }
    4.  
    5.     public static <T, S extends Statement> T execute(StatementProxy<S> statementProxy,
    6.                                                      StatementCallback<T, S> statementCallback,
    7.                                                      Object... args) throws SQLException {
    8.         return execute(null, statementProxy, statementCallback, args);
    9.     }
    10.  
    11.  
    12.     public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecognizers,
    13.                                                      StatementProxy<S> statementProxy,
    14.                                                      StatementCallback<T, S> statementCallback,
    15.                                                      Object... args) throws SQLException {
    16.         if (!RootContext.requireGlobalLock() && BranchType.AT != RootContext.getBranchType()) {
    17.             // Just work as original statement
    18.             return statementCallback.execute(statementProxy.getTargetStatement(), args);
    19.         }
    20.  
    21.         String dbType = statementProxy.getConnectionProxy().getDbType();
    22.         if (CollectionUtils.isEmpty(sqlRecognizers)) {
    23.             //获取sql类型的识别器
    24.             sqlRecognizers = SQLVisitorFactory.get(
    25.                     statementProxy.getTargetSQL(),
    26.                     dbType);
    27.         }
    28.         Executor<T> executor;
    29.         if (CollectionUtils.isEmpty(sqlRecognizers)) {
    30.             executor = new PlainExecutor<>(statementProxy, statementCallback);
    31.         } else {
    32.             if (sqlRecognizers.size() == 1) {
    33.                 SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
    34.                 //根据不同的sql类型进入相应的case
    35.                 switch (sqlRecognizer.getSQLType()) {
    36.                     case INSERT:
    37.                         executor = EnhancedServiceLoader.load(InsertExecutor.class, dbType,
    38.                                 new Class[]{StatementProxy.class, StatementCallback.class, SQLRecognizer.class},
    39.                                 new Object[]{statementProxy, statementCallback, sqlRecognizer});
    40.                         break;
    41.                     case UPDATE:
    42.                         //以update操作为例，得到一个UpdateExecutor
    43.                         executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
    44.                         break;
    45.                     case DELETE:
    46.                         executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer);
    47.                         break;
    48.                     case SELECT_FOR_UPDATE:
    49.                         executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
    50.                         break;
    51.                     default:
    52.                         executor = new PlainExecutor<>(statementProxy, statementCallback);
    53.                         break;
    54.                 }
    55.             } else {
    56.                 executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers);
    57.             }
    58.         }
    59.         T rs;
    60.         try {
    61.             //执行UpdateExecutor的execute方法执行sql
    62.             rs = executor.execute(args);
    63.         } catch (Throwable ex) {
    64.             if (!(ex instanceof SQLException)) {
    65.                 // Turn other exception into SQLException
    66.                 ex = new SQLException(ex);
    67.             }
    68.             throw (SQLException) ex;
    69.         }
    70.         return rs;
    71.     }

    来到父类BaseTransactionalExecutor#execute：

    1.     public T execute(Object... args) throws Throwable {
    2.         //从本地线程变量获取绑定了的全局事务id
    3.         String xid = RootContext.getXID();
    4.         if (xid != null) {
    5.             //把xid绑定连接ConnectionProxy
    6.             statementProxy.getConnectionProxy().bind(xid);
    7.         }
    8.         //设置GlobalLockRequire为true，意为在本地事务提交时需要获取全局锁
    9.         statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
    10.         return doExecute(args);
    11.     }
    12.  
    13.  
    14.     public T doExecute(Object... args) throws Throwable {
    15.         AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
    16.         if (connectionProxy.getAutoCommit()) {
    17.             return executeAutoCommitTrue(args);
    18.         } else {
    19.             return executeAutoCommitFalse(args);
    20.         }
    21.     }
    22.  
    23.  
    24.     protected T executeAutoCommitFalse(Object[] args) throws Exception {
    25.         if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && isMultiPk()) {
    26.             throw new NotSupportYetException("multi pk only support mysql!");
    27.         }
    28.         //获取前置镜像，其实就是select for updata查询语句，查询修改前对应的数据
    29.         TableRecords beforeImage = beforeImage();
    30.         //执行目标sql
    31.         T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
    32.         //获取后置镜像
    33.         TableRecords afterImage = afterImage(beforeImage);
    34.         //用前置镜像与后置镜像构建undo_log回滚日志，并设置到connectionProxy上
    35.         prepareUndoLog(beforeImage, afterImage);
    36.         return result;
    37.     }

    然后执行提交操作ConnectionProxy#commit：

    1.     public void commit() throws SQLException {
    2.         try {
    3.             LOCK_RETRY_POLICY.execute(() -> {
    4.                 doCommit();
    5.                 return null;
    6.             });
    7.         } catch (SQLException e) {
    8.             if (targetConnection != null && !getAutoCommit() && !getContext().isAutoCommitChanged()) {
    9.                 rollback();
    10.             }
    11.             throw e;
    12.         } catch (Exception e) {
    13.             throw new SQLException(e);
    14.         }
    15.     }
    16.  
    17.  
    18.     private void doCommit() throws SQLException {
    19.         if (context.inGlobalTransaction()) {
    20.             //如果时分布式事务的话
    21.             processGlobalTransactionCommit();
    22.         } else if (context.isGlobalLockRequire()) {
    23.             processLocalCommitWithGlobalLocks();
    24.         } else {
    25.             targetConnection.commit();
    26.         }
    27.     }
    28.  
    29.  
    30.     private void processGlobalTransactionCommit() throws SQLException {
    31.         try {
    32.             //向TC注册分支事务
    33.             register();
    34.         } catch (TransactionException e) {
    35.             recognizeLockKeyConflictException(e, context.buildLockKeys());
    36.         }
    37.         try {
    38.             //插入undo_log日志到undo_log表
    39.             UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
    40.             //本地数据库提交
    41.             targetConnection.commit();
    42.         } catch (Throwable ex) {
    43.             LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
    44.             //提交异常会向TC上报
    45.             report(false);
    46.             throw new SQLException(ex);
    47.         }
    48.         if (IS_REPORT_SUCCESS_ENABLE) {
    49.             report(true);
    50.         }
    51.         context.reset();
    52.     }
    53.  
    54.  
    55.     private void register() throws TransactionException {
    56.         if (!context.hasUndoLog() || !context.hasLockKey()) {
    57.             return;
    58.         }
    59.         //RM向TC注册分支事务，返回分支事务id，依然基于netty
    60.         Long branchId = DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(),
    61.             null, context.getXid(), null, context.buildLockKeys());
    62.         context.setBranchId(branchId);
    63.     }

            值得注意的是，如果在执行目标sql发生异常，那么本地RM就不会去注册分支事务了，直接本地就会滚，从全局来看，全局事务发起者TM通过捕获rpc调用异常发起全局事务回滚。在注册分支事务时，可能会抛出TransactionException，意为获取全局锁失败，有可能是另一个全局事务拿到了全局锁，本地重试策略会不断睡眠+重试，最终抛出LockWaitTimeoutException超时异常，本地事务回滚，进而导致全局回滚。

    总结

            以上只是分布式事务第一阶段，一阶段本地事务顺利提交，等待二阶段TC的异步通知，二阶段本地RM根据TC的反馈，如果全局回滚，则执行本地的undo_log回滚日志，如果全局提交，则删除undo_log即可，最后RM上报这一步的结果，TC释放全局锁。

  • 相关阅读:
    17.RedHat认证-Ansible自动化运维(下)
    Linux：进程概念的引入和理解
    高通导航器软件开发包使用指南（10）
    【optimtool.unconstrain】无约束优化工具箱
    Linux下企业级夜莺监控分析工具的远程访问设置【内网穿透】
    C++排序函数sort()和qsort()的参数比较函数的统一记忆方法
    spring面试题
    计算机保研er历程分享（浙软、厦大、华师、东南网安、东北、西电、中南......）
    商业计划书PPT怎么做？这个AI软件一键在线生成，做PPT再也不求人！
    动态规划4（Leetcode746使用最小花费爬楼梯）
  • 原文地址：https://blog.csdn.net/w7sss/article/details/125509717