Seata事务隔离

本文目标：帮助用户明白使用Seata AT模式时，该如何正确实现事务隔离，防止脏读脏写。

希望读者在阅读本文前，已阅读过seata官网中对AT模式的介绍，并且对数据库本地锁有所了解

（例如，两个事务同时在对同一条记录做update时，只有拿到record lock的事务才能更新成功，另一个事务在record lock未释放前只能等待，直到事务超时）

首先请看这样的一段代码，尽管看着“初级”，但持久层框架实际上帮我们做的主要事情也就这样。

  1. @Service
  2. public class StorageService {
  4.     @Autowired
  5.     private DataSource dataSource;
  7.     @GlobalTransactional
  8.     public void batchUpdate() throws SQLException {
  9.         Connection connection = null;
  10.         PreparedStatement preparedStatement = null;
  11.         try {
  12.             connection = dataSource.getConnection();
  13.             connection.setAutoCommit(false);
  14.             String sql = "update storage_tbl set count = ?" +
  15.                 "    where id = ? and commodity_code = ?";
  16.             preparedStatement = connection.prepareStatement(sql);
  17.             preparedStatement.setInt(1, 100);
  18.             preparedStatement.setLong(2, 1);
  19.             preparedStatement.setString(3, "2001");
  20.             preparedStatement.executeUpdate();
  21.             connection.commit();
  22.         } catch (Exception e) {
  23.             throw e;
  24.         } finally {
  25.             IOutils.close(preparedStatement);
  26.             IOutils.close(connection);
  27.         }
  28.     }
  30. }

从代理数据源说起

使用AT模式，最重要的事情便是代理数据源，那么用DataSourceProxy代理数据源有什么作用呢？

DataSourceProxy能帮助我们获得几个重要的代理对象

  • 通过DataSourceProxy.getConnection()获得ConnectionProxy

  • 通过ConnectionProxy.prepareStatement(...)获得StatementProxy

Seata的如何实现事务隔离，就藏在这2个Proxy中，我先概述下实现逻辑。

StatementProxy.executeXXX()的处理逻辑

  • 当调用io.seata.rm.datasource.StatementProxy.executeXXX()会将sql交给io.seata.rm.datasource.exec.ExecuteTemplate.execute(...)处理。

    • ExecuteTemplate.execute(...)方法中，Seata根据不同dbType和sql语句类型使用不同的Executer，调用io.seata.rm.datasource.exec.Executer类的execute(Object... args)。
    • 如果选了DML类型Executer，主要做了以下事情：
      • 查询前镜像（select for update，因此此时获得本地锁）
      • 执行业务sql
      • 查询后镜像
      • 准备undoLog
    • 如果你的sql是select for update则会使用SelectForUpdateExecutor（Seata代理了select for update），代理后处理的逻辑是这样的：
      • 先执行 select for update（获取数据库本地锁）
      • 如果处于@GlobalTransactional or @GlobalLock，检查是否有全局锁
      • 如果有全局锁，则未开启本地事务下会rollback本地事务,再重新争抢本地锁和全局锁,以此类推,除非拿到全局锁

ConnectionProxy.commit()的处理逻辑

  • 处于全局事务中（即，数据持久化方法带有@GlobalTransactional）
    • 注册分支事务，获取全局锁
    • undoLog数据入库
    • 让数据库commit本次事务
  • 处于@GlobalLock中（即，数据持久化方法带有@GlobalLock）
    • 向tc查询是否有全局锁存在，如存在，则抛出异常
    • 让数据库commit本次事务
  • 除了以上情况（else分支）
    • 让数据库commit本次事务

@GlobalTransactional的作用

标识一个全局事务

@GlobalLock + select for update的作用

如果像updateA()方法带有@GlobalLock + select for update，Seata在处理时，会先获取数据库本地锁，然后查询该记录是否有全局锁存在，若有，则抛出LockConflictException。

先举一个脏写的例子，再来看Seata如何防止脏写

假设你的业务代码是这样的：

  • updateAll()用来同时更新A和B表记录，updateA() updateB()则分别更新A、B表记录
  • updateAll()已经加上了@GlobalTransactional
  1. class YourBussinessService {
  3.     DbServiceA serviceA;
  4.     DbServiceB serviceB;
  6.     @GlobalTransactional
  7.     public boolean updateAll(DTO dto) {
  8.         serviceA.update(dto.getA());
  9.         serviceB.update(dto.getB());
  10.     }
  12.     public boolean updateA(DTO dto) {
  13.         serviceA.update(dto.getA());
  14.     }
  16. }
  1. class DbServiceA {
  2.     @Transactional
  3.     public boolean update(A a) {
  5.     }
  6. }

dirty-write |

怎么用Seata防止脏写？

办法一：updateA()也加上@GlobalTransactional，此时Seata会如何保证事务隔离？

  1. class DbServiceA {
  3.     @GlobalTransactional
  4.     @Transactional
  5.     public boolean updateA(DTO dto) {
  7.         serviceA.update(dto.getA());
  9.     }
  10. }
  • updateAll()先被调用（未完成），updateA()后被调用

dirty-write

办法二： @GlobalLock + select for update

  1. class DbServiceA {
  3.     @GlobalLock
  4.     @Transactional
  5.     public boolean updateA(DTO dto) {
  7.         serviceA.selectForUpdate(dto.getA());
  9.         serviceA.update(dto.getA());
  11.     }
  12. }

  • updateAll()先被调用（未完成），updateA()后被调用 dirty-write

  • 那如果是updateA()先被调用（未完成），updateAll()后被调用呢？
    由于2个业务都是要先获得本地锁，因此同样不会发生脏写

  • 一定有人会问，“这里为什么要加上select for update? 只用@GlobalLock能不能防止脏写？” 能。但请再回看下上面的图，select for update能带来这么几个好处：

    • 锁冲突更“温柔”些。如果只有@GlobalLock，检查到全局锁，则立刻抛出异常，也许再“坚持”那么一下，全局锁就释放了，抛出异常岂不可惜了。
    • 在updateA()中可以通过select for update获得最新的A，接着再做更新。

如何防止脏读？

场景： 某业务先调用updateAll()，updateAll()未执行完成，另一业务后调用queryA()

dirty-write

源码展示

  1. @Service
  2. public class StorageService {
  4.     @Autowired
  5.     private DataSource dataSource;
  7.     @GlobalTransactional
  8.     public void update() throws SQLException {
  9.         Connection connection = null;
  10.         PreparedStatement preparedStatement = null;
  11.         try {
  12.             connection = dataSource.getConnection();
  13.             connection.setAutoCommit(false);
  14.             String sql = "update storage_tbl set count = ?" +
  15.                 "    where id = ? and commodity_code = ?";
  16.             preparedStatement = connection.prepareStatement(sql);
  17.             preparedStatement.setInt(1, 100);
  18.             preparedStatement.setLong(2, 1);
  19.             preparedStatement.setString(3, "2001");
  20.             preparedStatement.execute();
  21.             connection.commit();
  22.         } catch (Exception e) {
  23.             throw e;
  24.         } finally {
  25.             IOutils.close(preparedStatement);
  26.             IOutils.close(connection);
  27.         }
  28.     }
  30. }

这段代码虽然看着很初级，没有使用持久层框架，但如果将框架帮我们做的事情抽象出来，其实也就是上面这段代码。

简单说明接下来源码介绍的脉络（主要关注和事务隔离有关的源码）

  • 代理数据源的用途
    • DataSourceProxy的作用（返回ConnectionProxy）
      • 介绍 ConnectionProxy的一个小功能（存放undolog）
    • ConnectionProxy的作用（返回StatementProxy）
    • StatementProxy.execute()的处理逻辑
      • io.seata.rm.datasource.exec.UpdateExecutor的执行逻辑（查前镜像、执行sql、查后镜像、准备undoLog）
      • SelectForUpdateExecutor的执行逻辑（挣本地锁，查全局锁。有全局锁，回滚，再争…）
    • ConnectionProxy.commit()的处理逻辑（注册分支事务（争全局锁），写入undoLog，数据库提交）
  • 介绍RootContext
  • GlobalTransactionalInterceptor的不同代理逻辑
    • 带有@GlobalTransactional如何处理
    • 带有@GlobalLock如何处理

DataSourceProxy的作用

DataSourceProxy帮助我们获得几个重要的代理对象

  • 通过DataSourceProxy.getConnection()获得ConnectionProxy

    1. package io.seata.rm.datasource;
    3. import java.sql.Connection;
    5. public class DataSourceProxy extends AbstractDataSourceProxy implements Resource {
    7.     @Override
    8.     public ConnectionProxy getConnection() throws SQLException {
    9.         Connection targetConnection = targetDataSource.getConnection();
    10.         return new ConnectionProxy(this, targetConnection);
    11.     }
    12. }

    • 现在先介绍下ConnectionProxy中的ConnectionContext，它的有一个功能是存放undoLog。

      1. package io.seata.rm.datasource;
      3. import io.seata.rm.datasource.undo.SQLUndoLog;
      5. public class ConnectionProxy extends AbstractConnectionProxy {
      7.     private ConnectionContext context = new ConnectionContext();
      9.     public void appendUndoLog(SQLUndoLog sqlUndoLog) {
      10.         context.appendUndoItem(sqlUndoLog);
      11.     }
      13. }
      1. package io.seata.rm.datasource;
      3. public class ConnectionContext {
      5.     private static final Savepoint DEFAULT_SAVEPOINT = new Savepoint() {
      6.         @Override
      7.         public int getSavepointId() throws SQLException {
      8.             return 0;
      9.         }
      11.         @Override
      12.         public String getSavepointName() throws SQLException {
      13.             return "DEFAULT_SEATA_SAVEPOINT";
      14.         }
      15.     };
      17.     private final Map<Savepoint, List<SQLUndoLog>> sqlUndoItemsBuffer = new LinkedHashMap<>();
      19.     private Savepoint currentSavepoint = DEFAULT_SAVEPOINT;
      21.     void appendUndoItem(SQLUndoLog sqlUndoLog) {
      22.         sqlUndoItemsBuffer.computeIfAbsent(currentSavepoint, k -> new ArrayList<>()).add(sqlUndoLog);
      23.     }
      25. }

通过ConnectionProxy.prepareStatement(...)获得StatementProxy

  1. package io.seata.rm.datasource;
  3. public class ConnectionProxy extends AbstractConnectionProxy {
  5.     public ConnectionProxy(DataSourceProxy dataSourceProxy, Connection targetConnection) {
  6.         super(dataSourceProxy, targetConnection);
  7.     }
  9. }
  1. package io.seata.rm.datasource;
  3. import java.sql.Connection;
  5. public abstract class AbstractConnectionProxy implements Connection {
  7.     protected Connection targetConnection;
  9.     public AbstractConnectionProxy(DataSourceProxy dataSourceProxy, Connection targetConnection) {
  10.         this.dataSourceProxy = dataSourceProxy;
  11.         this.targetConnection = targetConnection;
  12.     }
  14.     @Override
  15.     public PreparedStatement prepareStatement(String sql) throws SQLException {
  16.         String dbType = getDbType();
  17.         // support oracle 10.2+
  18.         PreparedStatement targetPreparedStatement = null;
  19.         if (BranchType.AT == RootContext.getBranchType()) { //为什么这里会返回AT？
  20.             List<SQLRecognizer> sqlRecognizers = SQLVisitorFactory.get(sql, dbType);
  21.             if (sqlRecognizers != null && sqlRecognizers.size() == 1) {
  22.                 SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
  23.                 if (sqlRecognizer != null && sqlRecognizer.getSQLType() == SQLType.INSERT) {
  24.                     TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dbType).getTableMeta(getTargetConnection(),
  25.                             sqlRecognizer.getTableName(), getDataSourceProxy().getResourceId());
  26.                     String[] pkNameArray = new String[tableMeta.getPrimaryKeyOnlyName().size()];
  27.                     tableMeta.getPrimaryKeyOnlyName().toArray(pkNameArray);
  28.                     // 如果是insert语句，这里创建的PreparedStatement需要可以返回自动生成的主键，因此使用这个prepareStatement()
  29.                     targetPreparedStatement = getTargetConnection().prepareStatement(sql,pkNameArray);
  31.                 }
  32.             }
  33.         }
  34.         if (targetPreparedStatement == null) {
  35.             targetPreparedStatement = getTargetConnection().prepareStatement(sql);
  36.         }
  37.         return new PreparedStatementProxy(this, targetPreparedStatement, sql);
  38.     }
  41.     public Connection getTargetConnection() {
  42.         return targetConnection;
  43.     }
  45. }

先在这打下个疑问，后边解释。
RootContext.getBranchType()的返回值怎么会是AT？

StatementProxy.execute()的处理逻辑

  • 当调用io.seata.rm.datasource.StatementProxy.execute()会将sql交给io.seata.rm.datasource.exec.ExecuteTemplate.execute(...)处理。

    1. package io.seata.rm.datasource;
    3. public class PreparedStatementProxy extends AbstractPreparedStatementProxy
    4.     implements PreparedStatement, ParametersHolder {
    6.     @Override
    7.     public boolean execute() throws SQLException {
    8.         return ExecuteTemplate.execute(this, (statement, args) -> statement.execute());
    9.     }
    11. }

    • ExecuteTemplate.execute(...)方法中，Seata根据不同dbType和sql语句类型使用不同的Executer，调用io.seata.rm.datasource.exec.Executer类的execute(Object... args)。

      ``` package io.seata.rm.datasource.exec;

  1.     public class ExecuteTemplate {
  3.         public static <T, S extends Statement> T execute(StatementProxy<S> statementProxy,
  4.                                                  StatementCallback<T, S> statementCallback,
  5.                                                  Object... args) throws SQLException {
  6.             return execute(null, statementProxy, statementCallback, args);
  7.         }
  9.         public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecognizers,
  10.                                              StatementProxy<S> statementProxy,
  11.                                              StatementCallback<T, S> statementCallback,
  12.                                              Object... args) throws SQLException {
  13.             if (!RootContext.requireGlobalLock() && BranchType.AT != RootContext.getBranchType()) {
  14.                 // Just work as original statement
  15.                 return statementCallback.execute(statementProxy.getTargetStatement(), args);
  16.             }
  18.             String dbType = statementProxy.getConnectionProxy().getDbType();
  19.             if (CollectionUtils.isEmpty(sqlRecognizers)) {
  20.                 sqlRecognizers = SQLVisitorFactory.get(
  21.                         statementProxy.getTargetSQL(),
  22.                         dbType);
  23.             }
  24.             Executor<T> executor;
  25.             if (CollectionUtils.isEmpty(sqlRecognizers)) {
  26.                 executor = new PlainExecutor<>(statementProxy, statementCallback);
  27.             } else {
  28.                 if (sqlRecognizers.size() == 1) {
  29.                     SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
  30.                     switch (sqlRecognizer.getSQLType()) {
  31.                         case INSERT:
  32.                             executor = EnhancedServiceLoader.load(InsertExecutor.class, dbType,
  33.                                     new Class[]{StatementProxy.class, StatementCallback.class, SQLRecognizer.class},
  34.                                     new Object[]{statementProxy, statementCallback, sqlRecognizer});
  35.                             break;
  36.                         case UPDATE:
  37.                             executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
  38.                             break;
  39.                         case DELETE:
  40.                             executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer);
  41.                             break;
  42.                         case SELECT_FOR_UPDATE:
  43.                             executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
  44.                             break;
  45.                         default:
  46.                             executor = new PlainExecutor<>(statementProxy, statementCallback);
  47.                             break;
  48.                     }
  49.                 } else {
  50.                     executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers);
  51.                 }
  52.             }
  53.             T rs;
  54.             try {
  55.                 rs = executor.execute(args);
  56.             } catch (Throwable ex) {
  57.                 if (!(ex instanceof SQLException)) {
  58.                     // Turn other exception into SQLException
  59.                     ex = new SQLException(ex);
  60.                 }
  61.                 throw (SQLException) ex;
  62.             }
  63.             return rs;
  64.         }
  66.     }
  67.     ```
  69.     > 也在这打下个疑问，后边解释。  
  70.     > **`RootContext.requireGlobalLock()`怎么判断当前是否需要全局锁？**
  72.     先以`io.seata.rm.datasource.exec.UpdateExecutor`举例，`UpdateExecutor` extends `AbstractDMLBaseExecutor` extends `BaseTransactionalExecutor`。 观察`execute()`方法的做了什么
  74.     ```
  75.     package io.seata.rm.datasource.exec;
  77.     public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor<T> {
  80.         protected StatementProxy<S> statementProxy;
  82.         protected StatementCallback<T, S> statementCallback;
  84.         protected SQLRecognizer sqlRecognizer;
  86.         public BaseTransactionalExecutor(StatementProxy<S> statementProxy, StatementCallback<T, S> statementCallback,
  87.             SQLRecognizer sqlRecognizer) {
  88.             this.statementProxy = statementProxy;
  89.             this.statementCallback = statementCallback;
  90.             this.sqlRecognizer = sqlRecognizer;
  91.         }
  93.         @Override
  94.         public T execute(Object... args) throws Throwable {
  95.             String xid = RootContext.getXID();
  96.             if (xid != null) {
  97.                 statementProxy.getConnectionProxy().bind(xid);
  98.             }
  100.             statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
  101.             return doExecute(args);
  102.         }
  104.     }
  105.     ```
  107.     ```
  108.     public abstract class AbstractDMLBaseExecutor<T, S extends Statement> extends BaseTransactionalExecutor<T, S> {
  110.         public AbstractDMLBaseExecutor(StatementProxy<S> statementProxy, StatementCallback<T, S> statementCallback,
  111.                                SQLRecognizer sqlRecognizer) {
  112.             super(statementProxy, statementCallback, sqlRecognizer);
  113.         }
  115.         @Override
  116.         public T doExecute(Object... args) throws Throwable {
  117.             AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
  118.             if (connectionProxy.getAutoCommit()) {
  119.                 return executeAutoCommitTrue(args);
  120.             } else {
  121.                 return executeAutoCommitFalse(args);
  122.             }
  123.         }
  125.         protected T executeAutoCommitTrue(Object[] args) throws Throwable {
  126.             ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
  127.             try {
  128.                 connectionProxy.changeAutoCommit(); // 注意，你如果没开启事务，seata帮你开启
  129.                 return new LockRetryPolicy(connectionProxy).execute(() -> {
  130.                     T result = executeAutoCommitFalse(args);
  131.                     connectionProxy.commit(); // 帮你开启事务后，通过connectionProxy来提交
  132.                     return result;
  133.                 });
  134.             } catch (Exception e) {
  135.                 // when exception occur in finally,this exception will lost, so just print it here
  136.                 LOGGER.error("execute executeAutoCommitTrue error:{}", e.getMessage(), e);
  137.                 if (!LockRetryPolicy.isLockRetryPolicyBranchRollbackOnConflict()) {
  138.                     connectionProxy.getTargetConnection().rollback();
  139.                 }
  140.                 throw e;
  141.             } finally {
  142.                 connectionProxy.getContext().reset();
  143.                 connectionProxy.setAutoCommit(true);
  144.             }
  145.         }
  147.         protected T executeAutoCommitFalse(Object[] args) throws Exception {
  148.             if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && isMultiPk()) {
  149.                 throw new NotSupportYetException("multi pk only support mysql!");
  150.             }
  151.             TableRecords beforeImage = beforeImage();
  152.             T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
  153.             TableRecords afterImage = afterImage(beforeImage);
  154.             prepareUndoLog(beforeImage, afterImage);
  155.             return result;
  156.         }
  157.     }
  158.     ```
  160.     ```
  161.     package io.seata.rm.datasource.exec;
  163.     public class UpdateExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
  165.         public UpdateExecutor(StatementProxy<S> statementProxy, StatementCallback<T, S> statementCallback,
  166.                             SQLRecognizer sqlRecognizer) {
  167.             super(statementProxy, statementCallback, sqlRecognizer);
  168.         }
  170.     }
  172.     ```
  174. -   如果选了DML类型Executer，可以在上面的executeAutoCommitFalse()中看到，主要做了以下事情：
  176.     -   查询前镜像（select for update，因此此时获得本地锁）
  178.         ```
  179.         package io.seata.rm.datasource.exec;
  181.         public class UpdateExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
  183.             private static final boolean ONLY_CARE_UPDATE_COLUMNS = CONFIG.getBoolean(
  184.                     ConfigurationKeys.TRANSACTION_UNDO_ONLY_CARE_UPDATE_COLUMNS, DefaultValues.DEFAULT_ONLY_CARE_UPDATE_COLUMNS); // 默认为true
  186.             @Override
  187.             protected TableRecords beforeImage() throws SQLException {
  188.                 ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
  189.                 TableMeta tmeta = getTableMeta();
  190.                 String selectSQL = buildBeforeImageSQL(tmeta, paramAppenderList);
  191.                 // SELECT id, count FROM storage_tbl WHERE id = ? FOR UPDATE
  192.                 return buildTableRecords(tmeta, selectSQL, paramAppenderList);
  193.             }
  195.             private String buildBeforeImageSQL(TableMeta tableMeta, ArrayList<List<Object>> paramAppenderList) {
  196.                 SQLUpdateRecognizer recognizer = (SQLUpdateRecognizer) sqlRecognizer;
  197.                 List<String> updateColumns = recognizer.getUpdateColumns();
  198.                 StringBuilder prefix = new StringBuilder("SELECT ");
  199.                 StringBuilder suffix = new StringBuilder(" FROM ").append(getFromTableInSQL());
  200.                 String whereCondition = buildWhereCondition(recognizer, paramAppenderList);
  201.                 if (StringUtils.isNotBlank(whereCondition)) {
  202.                     suffix.append(WHERE).append(whereCondition);
  203.                 }
  204.                 String orderBy = recognizer.getOrderBy();
  205.                 if (StringUtils.isNotBlank(orderBy)) {
  206.                     suffix.append(orderBy);
  207.                 }
  208.                 ParametersHolder parametersHolder = statementProxy instanceof ParametersHolder ? (ParametersHolder)statementProxy : null;
  209.                 String limit = recognizer.getLimit(parametersHolder, paramAppenderList);
  210.                 if (StringUtils.isNotBlank(limit)) {
  211.                     suffix.append(limit);
  212.                 }
  213.                 suffix.append(" FOR UPDATE");
  214.                 StringJoiner selectSQLJoin = new StringJoiner(", ", prefix.toString(), suffix.toString());
  215.                 if (ONLY_CARE_UPDATE_COLUMNS) {
  216.                     if (!containsPK(updateColumns)) {// 如果本次更新的行不包含主键，那select for update的时候加上主键
  217.                         selectSQLJoin.add(getColumnNamesInSQL(tableMeta.getEscapePkNameList(getDbType())));
  218.                     }
  219.                     for (String columnName : updateColumns) {
  220.                         selectSQLJoin.add(columnName);
  221.                     }
  222.                 } else {
  223.                     for (String columnName : tableMeta.getAllColumns().keySet()) {
  224.                         selectSQLJoin.add(ColumnUtils.addEscape(columnName, getDbType()));
  225.                     }
  226.                 }
  227.                 return selectSQLJoin.toString();
  228.             }
  231.             protected TableRecords buildTableRecords(TableMeta tableMeta, String selectSQL, ArrayList<List<Object>> paramAppenderList) throws SQLException {
  232.                 ResultSet rs = null;
  233.                 try (PreparedStatement ps = statementProxy.getConnection().prepareStatement(selectSQL)) { // 执行select for update，然后就拿到了本地锁
  234.                     if (CollectionUtils.isNotEmpty(paramAppenderList)) {
  235.                         for (int i = 0, ts = paramAppenderList.size(); i < ts; i++) {
  236.                             List<Object> paramAppender = paramAppenderList.get(i);
  237.                             for (int j = 0, ds = paramAppender.size(); j < ds; j++) {
  238.                                 ps.setObject(i * ds + j + 1, paramAppender.get(j));
  239.                             }
  240.                         }
  241.                     }
  242.                     rs = ps.executeQuery();
  243.                     return TableRecords.buildRecords(tableMeta, rs);
  244.                 } finally {
  245.                     IOUtil.close(rs);
  246.                 }
  247.             }
  248.         }
  250.         ```
  252.     -   执行业务sql
  254.     -   查询后镜像
  256.         ```
  257.           package io.seata.rm.datasource.exec;
  259.           public class UpdateExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
  261.               @Override
  262.               protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
  263.                   TableMeta tmeta = getTableMeta();
  264.                   if (beforeImage == null || beforeImage.size() == 0) {
  265.                       return TableRecords.empty(getTableMeta());
  266.                   }
  267.                   String selectSQL = buildAfterImageSQL(tmeta, beforeImage);
  268.                   //SELECT id, count FROM storage_tbl WHERE (id) in ( (?) )
  269.                   ResultSet rs = null;
  270.                   try (PreparedStatement pst = statementProxy.getConnection().prepareStatement(selectSQL)) {
  271.                       SqlGenerateUtils.setParamForPk(beforeImage.pkRows(), getTableMeta().getPrimaryKeyOnlyName(), pst);
  272.                       rs = pst.executeQuery();
  273.                       return TableRecords.buildRecords(tmeta, rs);
  274.                   } finally {
  275.                       IOUtil.close(rs);
  276.                   }
  277.               }
  278.           }
  279.         ```
  281.     -   准备undoLog
  283.         ```
  284.         public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor<T> {
  286.             protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
  287.                 if (beforeImage.getRows().isEmpty() && afterImage.getRows().isEmpty()) {
  288.                     return;
  289.                 }
  290.                 if (SQLType.UPDATE == sqlRecognizer.getSQLType()) {
  291.                     if (beforeImage.getRows().size() != afterImage.getRows().size()) {
  292.                         throw new ShouldNeverHappenException("Before image size is not equaled to after image size, probably because you updated the primary keys.");
  293.                     }
  294.                 }
  295.                 ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
  297.                 TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
  298.                 String lockKeys = buildLockKey(lockKeyRecords);
  299.                 if (null != lockKeys) {
  300.                     connectionProxy.appendLockKey(lockKeys);
  302.                     SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
  303.                     connectionProxy.appendUndoLog(sqlUndoLog); // 把undoLog存到connectionProxy中，具体怎么回事上面有提过
  304.                 }
  305.             }
  306.         }
  307.         ```
  309. -   如果你的sql是select for update则会使用`SelectForUpdateExecutor`（Seata代理了select for update），代理后处理的逻辑是这样的：
  311.     -   先执行 select for update（获取数据库本地锁）
  312.     -   如果处于`@GlobalTransactional` or `@GlobalLock`，**检查**是否有全局锁
  313.     -   如果有全局锁，则未开启本地事务下会rollback本地事务，再重新争抢本地锁和查询全局锁，直到全局锁释放
  315.     ```
  316.        package io.seata.rm.datasource.exec;
  318.        public class SelectForUpdateExecutor<T, S extends Statement> extends BaseTransactionalExecutor<T, S> {
  319.                @Override
  320.                 public T doExecute(Object... args) throws Throwable {
  321.                     Connection conn = statementProxy.getConnection();
  322.                     DatabaseMetaData dbmd = conn.getMetaData();
  323.                     T rs;
  324.                     Savepoint sp = null;
  325.                     boolean originalAutoCommit = conn.getAutoCommit();
  326.                     try {
  327.                         if (originalAutoCommit) {
  328.                             /*
  329.                              * In order to hold the local db lock during global lock checking
  330.                              * set auto commit value to false first if original auto commit was true
  331.                              */
  332.                             conn.setAutoCommit(false);
  333.                         } else if (dbmd.supportsSavepoints()) {
  334.                             /*
  335.                              * In order to release the local db lock when global lock conflict
  336.                              * create a save point if original auto commit was false, then use the save point here to release db
  337.                              * lock during global lock checking if necessary
  338.                              */
  339.                             sp = conn.setSavepoint();
  340.                         } else {
  341.                             throw new SQLException("not support savepoint. please check your db version");
  342.                         }
  344.                         LockRetryController lockRetryController = new LockRetryController();
  345.                         ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
  346.                         String selectPKSQL = buildSelectSQL(paramAppenderList);
  347.                         while (true) {
  348.                             try {
  349.                                 // #870
  350.                                 // execute return Boolean
  351.                                 // executeQuery return ResultSet
  352.                                 rs = statementCallback.execute(statementProxy.getTargetStatement(), args); //执行 select for update（获取数据库本地锁）
  354.                                 // Try to get global lock of those rows selected
  355.                                 TableRecords selectPKRows = buildTableRecords(getTableMeta(), selectPKSQL, paramAppenderList);
  356.                                 String lockKeys = buildLockKey(selectPKRows);
  357.                                 if (StringUtils.isNullOrEmpty(lockKeys)) {
  358.                                     break;
  359.                                 }
  361.                                 if (RootContext.inGlobalTransaction() || RootContext.requireGlobalLock()) {
  362.                                     // Do the same thing under either @GlobalTransactional or @GlobalLock, 
  363.                                     // that only check the global lock  here.
  364.                                     statementProxy.getConnectionProxy().checkLock(lockKeys);
  365.                                 } else {
  366.                                     throw new RuntimeException("Unknown situation!");
  367.                                 }
  368.                                 break;
  369.                             } catch (LockConflictException lce) {
  370.                                 if (sp != null) {
  371.                                     conn.rollback(sp);
  372.                                 } else {
  373.                                     conn.rollback();// 回滚，释放本地锁
  374.                                 }
  375.                                 // trigger retry
  376.                                 lockRetryController.sleep(lce);
  377.                             }
  378.                         }
  379.                     } finally {
  380.                         if (sp != null) {
  381.                             try {
  382.                                 if (!JdbcConstants.ORACLE.equalsIgnoreCase(getDbType())) {
  383.                                     conn.releaseSavepoint(sp);
  384.                                 }
  385.                             } catch (SQLException e) {
  386.                                 LOGGER.error("{} release save point error.", getDbType(), e);
  387.                             }
  388.                         }
  389.                         if (originalAutoCommit) {
  390.                             conn.setAutoCommit(true);
  391.                         }
  392.                     }
  393.                     return rs;
  394.                 }
  398.        }
  399.     ```

ConnectionProxy.commit()的处理逻辑

  1. public class ConnectionProxy extends AbstractConnectionProxy {
  3.     private final static LockRetryPolicy LOCK_RETRY_POLICY = new LockRetryPolicy();
  5.     private ConnectionContext context = new ConnectionContext();
  7.     @Override
  8.     public void commit() throws SQLException {
  9.         try {
  10.             LOCK_RETRY_POLICY.execute(() -> {
  11.                 doCommit();
  12.                 return null;
  13.             });
  14.         } catch (SQLException e) {
  15.             if (targetConnection != null && !getAutoCommit() && !getContext().isAutoCommitChanged()) {
  16.                 rollback();
  17.             }
  18.             throw e;
  19.         } catch (Exception e) {
  20.             throw new SQLException(e);
  21.         }
  22.     }
  24.     private void doCommit() throws SQLException {
  25.         if (context.inGlobalTransaction()) {
  26.             processGlobalTransactionCommit();
  27.         } else if (context.isGlobalLockRequire()) {
  28.             processLocalCommitWithGlobalLocks();
  29.         } else {
  30.             targetConnection.commit();
  31.         }
  32.     }
  34. }

也在这打下个疑问，后边解释。
ConnectionProxy里的ConnectionContext是如何判断inGlobalTransaction() or isGlobalLockRequire()的呢？

  • 处于全局事务中（即，数据持久化方法带有@GlobalTransactional）

    • 注册分支事务，获取全局锁
    • undoLog数据入库
    • 让数据库commit本次事务

    ```

    1. public class ConnectionProxy extends AbstractConnectionProxy {
    3.     private final static LockRetryPolicy LOCK_RETRY_POLICY = new LockRetryPolicy();
    5.     private ConnectionContext context = new ConnectionContext();
    7.     private void processGlobalTransactionCommit() throws SQLException {
    8.         try {
    9.             register(); // 注册分支，争全局锁
    10.         } catch (TransactionException e) {
    11.             recognizeLockKeyConflictException(e, context.buildLockKeys());
    12.         }
    13.         try {
    14.             UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this); // undolog入库
    15.             targetConnection.commit(); // 分支事务提交
    16.         } catch (Throwable ex) {
    17.             LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
    18.             report(false);
    19.             throw new SQLException(ex);
    20.         }
    21.         if (IS_REPORT_SUCCESS_ENABLE) {
    22.             report(true);
    23.         }
    24.         context.reset();
    25.     }
    27.     private void register() throws TransactionException {
    28.         if (!context.hasUndoLog() || !context.hasLockKey()) {
    29.             return;
    30.         }
    31.         Long branchId = DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(),
    32.             null, context.getXid(), null, context.buildLockKeys());
    33.         context.setBranchId(branchId);
    34.     }
  1.     }
  2.     ```
  4. ```

  • 处于@GlobalLock中（即，数据持久化方法带有@GlobalLock）

    • 向tc查询是否有全局锁存在
    • 让数据库commit本次事务
    1.    public class ConnectionProxy extends AbstractConnectionProxy {
    3.        private final static LockRetryPolicy LOCK_RETRY_POLICY = new LockRetryPolicy();
    5.        private ConnectionContext context = new ConnectionContext();
    7.        private void processLocalCommitWithGlobalLocks() throws SQLException {
    8.            checkLock(context.buildLockKeys());
    9.            try {
    10.                targetConnection.commit();
    11.            } catch (Throwable ex) {
    12.                throw new SQLException(ex);
    13.            }
    14.            context.reset();
    15.        }
    17.        public void checkLock(String lockKeys) throws SQLException {
    18.            if (StringUtils.isBlank(lockKeys)) {
    19.                return;
    20.            }
    21.            // Just check lock without requiring lock by now.
    22.            try {
    23.                boolean lockable = DefaultResourceManager.get().lockQuery(BranchType.AT,
    24.                    getDataSourceProxy().getResourceId(), context.getXid(), lockKeys);
    25.                if (!lockable) {
    26.                    throw new LockConflictException();
    27.                }
    28.            } catch (TransactionException e) {
    29.                recognizeLockKeyConflictException(e, lockKeys);
    30.            }
    31.        }
    33.    }

    ```

  • 除了以上情况（else分支）

    • 让数据库commit本次事务

介绍RootContext

我们在上面留下了3个“扣儿”，现在到了结合RootContext源码来解答的时候。

  1. RootContext.getBranchType()的返回值怎么会是AT？
    该方法的判断逻辑是：只要判断出当前处于全局事务中（即，只要有地方调用过RootContext.bind(xid)）, 就会返回默认BranchType.AT

    1. public class RootContext {
    3.     public static final String KEY_XID = "TX_XID";
    5.     private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load();
    7.     private static BranchType DEFAULT_BRANCH_TYPE;
    9.     @Nullable
    10.     public static BranchType getBranchType() {
    11.         if (inGlobalTransaction()) {
    12.             BranchType branchType = (BranchType) CONTEXT_HOLDER.get(KEY_BRANCH_TYPE);
    13.             if (branchType != null) {
    14.                 return branchType;
    15.             }
    16.             //Returns the default branch type.
    17.             return DEFAULT_BRANCH_TYPE != null ? DEFAULT_BRANCH_TYPE : BranchType.AT;
    18.         }
    19.         return null;
    20.     }
    22.     public static boolean inGlobalTransaction() {
    23.         return CONTEXT_HOLDER.get(KEY_XID) != null;
    24.     }
    26.     public static void bind(@Nonnull String xid) {
    27.         if (StringUtils.isBlank(xid)) {
    28.             if (LOGGER.isDebugEnabled()) {
    29.                 LOGGER.debug("xid is blank, switch to unbind operation!");
    30.             }
    31.             unbind();
    32.         } else {
    33.             MDC.put(MDC_KEY_XID, xid);
    34.             if (LOGGER.isDebugEnabled()) {
    35.                 LOGGER.debug("bind {}", xid);
    36.             }
    37.             CONTEXT_HOLDER.put(KEY_XID, xid);
    38.         }
    39.     }
    41. }

  2. RootContext.requireGlobalLock()怎么判断当前是否需要全局锁？
    需要有地方调用RootContext.bindGlobalLockFlag()

    1. public class RootContext {
    3.     public static final String KEY_GLOBAL_LOCK_FLAG = "TX_LOCK";
    4.     public static final Boolean VALUE_GLOBAL_LOCK_FLAG = true;
    6.     private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load();
    8.     public static boolean requireGlobalLock() {
    9.         return CONTEXT_HOLDER.get(KEY_GLOBAL_LOCK_FLAG) != null;
    10.     }
    12.     public static void bindGlobalLockFlag() {
    13.         if (LOGGER.isDebugEnabled()) {
    14.             LOGGER.debug("Local Transaction Global Lock support enabled");
    15.         }
    17.         //just put something not null
    18.         CONTEXT_HOLDER.put(KEY_GLOBAL_LOCK_FLAG, VALUE_GLOBAL_LOCK_FLAG);
    19.     }
    21. }

  3. ConnectionProxy.commit()会根据context的不同状态区分处理，那ConnectionContext是如何判断inGlobalTransaction() or isGlobalLockRequire()的呢？

    1.  public class ConnectionProxy extends AbstractConnectionProxy {
    3.     private ConnectionContext context = new ConnectionContext();
    5.     private void doCommit() throws SQLException {
    6.         if (context.inGlobalTransaction()) {
    7.             processGlobalTransactionCommit();
    8.         } else if (context.isGlobalLockRequire()) {
    9.             processLocalCommitWithGlobalLocks();
    10.         } else {
    11.             targetConnection.commit();
    12.         }
    13.     }
    15. }

    • 如何判断inGlobalTransaction()？（注意下，这里和上面提到的RootContext不是一个东西）

      1. public class ConnectionContext {
      3.     private String xid;
      5.     void setXid(String xid) {
      6.         this.xid = xid;
      7.     }
      9.     public boolean inGlobalTransaction() {
      10.         return xid != null;
      11.     }
      13.     void bind(String xid) {
      14.         if (xid == null) {
      15.             throw new IllegalArgumentException("xid should not be null");
      16.         }
      17.         if (!inGlobalTransaction()) {
      18.             setXid(xid);
      19.         } else {
      20.             if (!this.xid.equals(xid)) {
      21.                 throw new ShouldNeverHappenException();
      22.             }
      23.         }
      24.     }
      26. }

      哪里调用的ConnectionContext.bind(xid)?

      1. package io.seata.rm.datasource.exec;
      3. public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor<T> {
      5.     @Override
      6.     public T execute(Object... args) throws Throwable {
      7.         // 那么，这里的XID哪来的呢？往后看就知道，是来自开启全局事务的时候获得的，和@GlobalTransactional有关
      8.         String xid = RootContext.getXID(); 
      9.         if (xid != null) {
      10.             statementProxy.getConnectionProxy().bind(xid);
      11.         }
      13.         // 这里就是设置 isGlobalLockRequire的位置，和 @GlobalLock有关
      14.         statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
      15.         return doExecute(args);
      16.     }
      18. }
      1. public class ConnectionProxy extends AbstractConnectionProxy {
      3.    private ConnectionContext context = new ConnectionContext();
      5.     public void bind(String xid) {
      6.         context.bind(xid);
      7.     }
      9.     public void setGlobalLockRequire(boolean isLock) {
      10.         context.setGlobalLockRequire(isLock);
      11.     }
      13. }

    • 如何判断isGlobalLockRequire()？

      1. public class ConnectionContext {
      3.     private boolean isGlobalLockRequire;
      5.     boolean isGlobalLockRequire() {
      6.        return isGlobalLockRequire;
      7.     }
      9.     void setGlobalLockRequire(boolean isGlobalLockRequire) {
      10.         this.isGlobalLockRequire = isGlobalLockRequire;
      11.     }
      13. }

    在看过代码后，我们知道，只要有地方在RootContext中设置了xid，或bindGlobalLockFlag(),就会识别成不同的状态。 那么哪儿调用的呢？答案就在下方的GlobalTransactionalInterceptor中。

GlobalTransactionalInterceptor处理带有@GlobalTransactional或@GlobalLock的方法

带有@GlobalTransactional和@GlobalLock的方法会被代理，交给GlobalTransactionalInterceptor处理

  1. public class GlobalTransactionalInterceptor implements ConfigurationChangeListener, MethodInterceptor {
  3.     @Override
  4.     public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
  5.         Class<?> targetClass =
  6.             methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null;
  7.         Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
  8.         if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
  9.             final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
  10.             final GlobalTransactional globalTransactionalAnnotation =
  11.                 getAnnotation(method, targetClass, GlobalTransactional.class);
  12.             final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);
  13.             boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes);
  14.             if (!localDisable) {
  15.                 if (globalTransactionalAnnotation != null) {
  16.                     return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);// 处理 @GlobalTransactional
  17.                 } else if (globalLockAnnotation != null) {
  18.                     return handleGlobalLock(methodInvocation, globalLockAnnotation); // 处理 @GlobalLock
  19.                 }
  20.             }
  21.         }
  22.         return methodInvocation.proceed();
  23.     }
  25. }

先看处理@GlobalTransactional

  1. public class GlobalTransactionalInterceptor implements ConfigurationChangeListener, MethodInterceptor {
  3.     private final TransactionalTemplate transactionalTemplate = new TransactionalTemplate();
  5.     Object handleGlobalTransaction(final MethodInvocation methodInvocation,
  6.         final GlobalTransactional globalTrxAnno) throws Throwable {
  8.         //...
  9.         try {
  10.             return transactionalTemplate.execute(...);
  11.         } catch (TransactionalExecutor.ExecutionException e) {
  12.           // ...
  13.         } finally {
  14.             //...
  15.         }
  16.     }
  17. }

来到了经典的seata事务模板方法，我们要关注开启事务的部分

  1. public class TransactionalTemplate {
  3.     public Object execute(TransactionalExecutor business) throws Throwable {
  4.         // 1. Get transactionInfo
  5.         //...
  6.         // 1.1 Get current transaction, if not null, the tx role is 'GlobalTransactionRole.Participant'.
  7.         GlobalTransaction tx = GlobalTransactionContext.getCurrent();
  9.         // 1.2 Handle the transaction propagation.
  10.         // ...
  12.             // 1.3 If null, create new transaction with role 'GlobalTransactionRole.Launcher'.
  13.             if (tx == null) {
  14.                 tx = GlobalTransactionContext.createNew();
  15.             }
  17.            //...
  19.             try {
  20.                 // 2. If the tx role is 'GlobalTransactionRole.Launcher', send the request of beginTransaction to TC,
  21.                 //    else do nothing. Of course, the hooks will still be triggered.
  22.                 beginTransaction(txInfo, tx);
  24.                 Object rs;
  25.                 try {
  26.                     // Do Your Business
  27.                     rs = business.execute();
  28.                 } catch (Throwable ex) {
  29.                     // 3. The needed business exception to rollback.
  30.                     completeTransactionAfterThrowing(txInfo, tx, ex);
  31.                     throw ex;
  32.                 }
  34.                 // 4. everything is fine, commit.
  35.                 commitTransaction(tx);
  37.                 return rs;
  38.             } finally {
  39.                 //5. clear
  40.                 //...
  41.             }
  42.         } finally {
  43.             // If the transaction is suspended, resume it.
  44.            // ...
  45.         }
  46.     }
  49.     private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
  50.         try {
  51.             triggerBeforeBegin();
  52.             tx.begin(txInfo.getTimeOut(), txInfo.getName());
  53.             triggerAfterBegin();
  54.         } catch (TransactionException txe) {
  55.             throw new TransactionalExecutor.ExecutionException(tx, txe,
  56.                 TransactionalExecutor.Code.BeginFailure);
  58.         }
  59.     }
  62. }
  1. public class DefaultGlobalTransaction implements GlobalTransaction {
  3.     @Override
  4.     public void begin(int timeout, String name) throws TransactionException {
  5.         if (role != GlobalTransactionRole.Launcher) {
  6.             assertXIDNotNull();
  7.             if (LOGGER.isDebugEnabled()) {
  8.                 LOGGER.debug("Ignore Begin(): just involved in global transaction [{}]", xid);
  9.             }
  10.             return;
  11.         }
  12.         assertXIDNull();
  13.         String currentXid = RootContext.getXID();
  14.         if (currentXid != null) {
  15.             throw new IllegalStateException("Global transaction already exists," +
  16.                 " can't begin a new global transaction, currentXid = " + currentXid);
  17.         }
  18.         xid = transactionManager.begin(null, null, name, timeout);
  19.         status = GlobalStatus.Begin;
  20.         RootContext.bind(xid); // 绑定xid
  21.         if (LOGGER.isInfoEnabled()) {
  22.             LOGGER.info("Begin new global transaction [{}]", xid);
  23.         }
  24.     }
  25. }

看到了吗？RootContext.bind(xid);

接着看处理@GlobalLock

  1. public class GlobalTransactionalInterceptor implements ConfigurationChangeListener, MethodInterceptor {
  3.     private final GlobalLockTemplate globalLockTemplate = new GlobalLockTemplate();
  5.     Object handleGlobalLock(final MethodInvocation methodInvocation,
  6.         final GlobalLock globalLockAnno) throws Throwable {
  8.         return globalLockTemplate.execute(new GlobalLockExecutor() {...});
  9.     }
  10. }

也使用了模板方法来处理GlobalLock

  1. public class GlobalLockTemplate {
  3.     public Object execute(GlobalLockExecutor executor) throws Throwable {
  4.         boolean alreadyInGlobalLock = RootContext.requireGlobalLock();
  5.         if (!alreadyInGlobalLock) {
  6.             RootContext.bindGlobalLockFlag();
  7.         }
  9.         // set my config to config holder so that it can be access in further execution
  10.         // for example, LockRetryController can access it with config holder
  11.         GlobalLockConfig myConfig = executor.getGlobalLockConfig();
  12.         GlobalLockConfig previousConfig = GlobalLockConfigHolder.setAndReturnPrevious(myConfig);
  14.         try {
  15.             return executor.execute();
  16.         } finally {
  17.             // only unbind when this is the root caller.
  18.             // otherwise, the outer caller would lose global lock flag
  19.             if (!alreadyInGlobalLock) {
  20.                 RootContext.unbindGlobalLockFlag();
  21.             }
  23.             // if previous config is not null, we need to set it back
  24.             // so that the outer logic can still use their config
  25.             if (previousConfig != null) {
  26.                 GlobalLockConfigHolder.setAndReturnPrevious(previousConfig);
  27.             } else {
  28.                 GlobalLockConfigHolder.remove();
  29.             }
  30.         }
  31.     }
  32. }

看到吗，一进模板方法就RootContext.bindGlobalLockFlag();