Seata事务隔离

标签：事务 return Seata statementProxy class new null public 隔离

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

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

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

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

@Service
public class StorageService {

    @Autowired
    private DataSource dataSource;

    @GlobalTransactional
    public void batchUpdate() throws SQLException {
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        try {
            connection = dataSource.getConnection();
            connection.setAutoCommit(false);
            String sql = "update storage_tbl set count = ?" +
                "    where id = ? and commodity_code = ?";
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setInt(1, 100);
            preparedStatement.setLong(2, 1);
            preparedStatement.setString(3, "2001");
            preparedStatement.executeUpdate();
            connection.commit();
        } catch (Exception e) {
            throw e;
        } finally {
            IOutils.close(preparedStatement);
            IOutils.close(connection);
        }
    }

}

从代理数据源说起

使用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

class YourBussinessService {

    DbServiceA serviceA;
    DbServiceB serviceB;

    @GlobalTransactional
    public boolean updateAll(DTO dto) {
        serviceA.update(dto.getA());
        serviceB.update(dto.getB());
    }

    public boolean updateA(DTO dto) {
        serviceA.update(dto.getA());
    }

}

class DbServiceA {
    @Transactional
    public boolean update(A a) {
    
    }
}

怎么用Seata防止脏写？

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

class DbServiceA {

    @GlobalTransactional
    @Transactional
    public boolean updateA(DTO dto) {

        serviceA.update(dto.getA());

    }
}

updateAll()先被调用（未完成），updateA()后被调用

办法二： @GlobalLock + select for update

class DbServiceA {
    
    @GlobalLock
    @Transactional
    public boolean updateA(DTO dto) {

        serviceA.selectForUpdate(dto.getA());

        serviceA.update(dto.getA());

    }
}

updateAll()先被调用（未完成），updateA()后被调用
那如果是updateA()先被调用（未完成），updateAll()后被调用呢？
由于2个业务都是要先获得本地锁，因此同样不会发生脏写
一定有人会问，“这里为什么要加上select for update? 只用@GlobalLock能不能防止脏写？” 能。但请再回看下上面的图，select for update能带来这么几个好处：
- 锁冲突更“温柔”些。如果只有@GlobalLock，检查到全局锁，则立刻抛出异常，也许再“坚持”那么一下，全局锁就释放了，抛出异常岂不可惜了。
- 在updateA()中可以通过select for update获得最新的A，接着再做更新。

如何防止脏读？

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

源码展示

@Service
public class StorageService {

    @Autowired
    private DataSource dataSource;

    @GlobalTransactional
    public void update() throws SQLException {
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        try {
            connection = dataSource.getConnection();
            connection.setAutoCommit(false);
            String sql = "update storage_tbl set count = ?" +
                "    where id = ? and commodity_code = ?";
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setInt(1, 100);
            preparedStatement.setLong(2, 1);
            preparedStatement.setString(3, "2001");
            preparedStatement.execute();
            connection.commit();
        } catch (Exception e) {
            throw e;
        } finally {
            IOutils.close(preparedStatement);
            IOutils.close(connection);
        }
    }

}

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

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

代理数据源的用途
- 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

package io.seata.rm.datasource;

import java.sql.Connection;

public class DataSourceProxy extends AbstractDataSourceProxy implements Resource {
    
    @Override
    public ConnectionProxy getConnection() throws SQLException {
        Connection targetConnection = targetDataSource.getConnection();
        return new ConnectionProxy(this, targetConnection);
    }
}

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

package io.seata.rm.datasource;

import io.seata.rm.datasource.undo.SQLUndoLog;

public class ConnectionProxy extends AbstractConnectionProxy {

    private ConnectionContext context = new ConnectionContext();

    public void appendUndoLog(SQLUndoLog sqlUndoLog) {
        context.appendUndoItem(sqlUndoLog);
    }

}

package io.seata.rm.datasource;

public class ConnectionContext {

    private static final Savepoint DEFAULT_SAVEPOINT = new Savepoint() {
        @Override
        public int getSavepointId() throws SQLException {
            return 0;
        }

        @Override
        public String getSavepointName() throws SQLException {
            return "DEFAULT_SEATA_SAVEPOINT";
        }
    };
    
    private final Map<Savepoint, List<SQLUndoLog>> sqlUndoItemsBuffer = new LinkedHashMap<>();

    private Savepoint currentSavepoint = DEFAULT_SAVEPOINT;

    void appendUndoItem(SQLUndoLog sqlUndoLog) {
        sqlUndoItemsBuffer.computeIfAbsent(currentSavepoint, k -> new ArrayList<>()).add(sqlUndoLog);
    }

}

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

package io.seata.rm.datasource;

public class ConnectionProxy extends AbstractConnectionProxy {

    public ConnectionProxy(DataSourceProxy dataSourceProxy, Connection targetConnection) {
        super(dataSourceProxy, targetConnection);
    }

}

package io.seata.rm.datasource;

import java.sql.Connection;

public abstract class AbstractConnectionProxy implements Connection {

    protected Connection targetConnection;

    public AbstractConnectionProxy(DataSourceProxy dataSourceProxy, Connection targetConnection) {
        this.dataSourceProxy = dataSourceProxy;
        this.targetConnection = targetConnection;
    }

    @Override
    public PreparedStatement prepareStatement(String sql) throws SQLException {
        String dbType = getDbType();
        // support oracle 10.2+
        PreparedStatement targetPreparedStatement = null;
        if (BranchType.AT == RootContext.getBranchType()) { //为什么这里会返回AT？
            List<SQLRecognizer> sqlRecognizers = SQLVisitorFactory.get(sql, dbType);
            if (sqlRecognizers != null && sqlRecognizers.size() == 1) {
                SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
                if (sqlRecognizer != null && sqlRecognizer.getSQLType() == SQLType.INSERT) {
                    TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dbType).getTableMeta(getTargetConnection(),
                            sqlRecognizer.getTableName(), getDataSourceProxy().getResourceId());
                    String[] pkNameArray = new String[tableMeta.getPrimaryKeyOnlyName().size()];
                    tableMeta.getPrimaryKeyOnlyName().toArray(pkNameArray);
                    // 如果是insert语句，这里创建的PreparedStatement需要可以返回自动生成的主键，因此使用这个prepareStatement()
                    targetPreparedStatement = getTargetConnection().prepareStatement(sql,pkNameArray);

                }
            }
        }
        if (targetPreparedStatement == null) {
            targetPreparedStatement = getTargetConnection().prepareStatement(sql);
        }
        return new PreparedStatementProxy(this, targetPreparedStatement, sql);
    }


    public Connection getTargetConnection() {
        return targetConnection;
    }

}

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

`StatementProxy.execute()`的处理逻辑

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

package io.seata.rm.datasource;

public class PreparedStatementProxy extends AbstractPreparedStatementProxy
    implements PreparedStatement, ParametersHolder {

    @Override
    public boolean execute() throws SQLException {
        return ExecuteTemplate.execute(this, (statement, args) -> statement.execute());
    }

}

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

package io.seata.rm.datasource.exec;


public class ExecuteTemplate {

    public static <T, S extends Statement> T execute(StatementProxy<S> statementProxy,
                                             StatementCallback<T, S> statementCallback,
                                             Object... args) throws SQLException {
        return execute(null, statementProxy, statementCallback, args);
    }

    public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecognizers,
                                         StatementProxy<S> statementProxy,
                                         StatementCallback<T, S> statementCallback,
                                         Object... args) throws SQLException {
        if (!RootContext.requireGlobalLock() && BranchType.AT != RootContext.getBranchType()) {
            // Just work as original statement
            return statementCallback.execute(statementProxy.getTargetStatement(), args);
        }

        String dbType = statementProxy.getConnectionProxy().getDbType();
        if (CollectionUtils.isEmpty(sqlRecognizers)) {
            sqlRecognizers = SQLVisitorFactory.get(
                    statementProxy.getTargetSQL(),
                    dbType);
        }
        Executor<T> executor;
        if (CollectionUtils.isEmpty(sqlRecognizers)) {
            executor = new PlainExecutor<>(statementProxy, statementCallback);
        } else {
            if (sqlRecognizers.size() == 1) {
                SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
                switch (sqlRecognizer.getSQLType()) {
                    case INSERT:
                        executor = EnhancedServiceLoader.load(InsertExecutor.class, dbType,
                                new Class[]{StatementProxy.class, StatementCallback.class, SQLRecognizer.class},
                                new Object[]{statementProxy, statementCallback, sqlRecognizer});
                        break;
                    case UPDATE:
                        executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                        break;
                    case DELETE:
                        executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                        break;
                    case SELECT_FOR_UPDATE:
                        executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                        break;
                    default:
                        executor = new PlainExecutor<>(statementProxy, statementCallback);
                        break;
                }
            } else {
                executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers);
            }
        }
        T rs;
        try {
            rs = executor.execute(args);
        } catch (Throwable ex) {
            if (!(ex instanceof SQLException)) {
                // Turn other exception into SQLException
                ex = new SQLException(ex);
            }
            throw (SQLException) ex;
        }
        return rs;
    }

}

也在这打下个疑问，后边解释。
RootContext.requireGlobalLock()怎么判断当前是否需要全局锁？

先以io.seata.rm.datasource.exec.UpdateExecutor举例，UpdateExecutor extends AbstractDMLBaseExecutor extends BaseTransactionalExecutor。观察execute()方法的做了什么

package io.seata.rm.datasource.exec;

public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor<T> {
    

    protected StatementProxy<S> statementProxy;

    protected StatementCallback<T, S> statementCallback;

    protected SQLRecognizer sqlRecognizer;

    public BaseTransactionalExecutor(StatementProxy<S> statementProxy, StatementCallback<T, S> statementCallback,
        SQLRecognizer sqlRecognizer) {
        this.statementProxy = statementProxy;
        this.statementCallback = statementCallback;
        this.sqlRecognizer = sqlRecognizer;
    }

    @Override
    public T execute(Object... args) throws Throwable {
        String xid = RootContext.getXID();
        if (xid != null) {
            statementProxy.getConnectionProxy().bind(xid);
        }

        statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
        return doExecute(args);
    }

}

public abstract class AbstractDMLBaseExecutor<T, S extends Statement> extends BaseTransactionalExecutor<T, S> {
    
    public AbstractDMLBaseExecutor(StatementProxy<S> statementProxy, StatementCallback<T, S> statementCallback,
                           SQLRecognizer sqlRecognizer) {
        super(statementProxy, statementCallback, sqlRecognizer);
    }

    @Override
    public T doExecute(Object... args) throws Throwable {
        AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        if (connectionProxy.getAutoCommit()) {
            return executeAutoCommitTrue(args);
        } else {
            return executeAutoCommitFalse(args);
        }
    }

    protected T executeAutoCommitTrue(Object[] args) throws Throwable {
        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        try {
            connectionProxy.changeAutoCommit(); // 注意，你如果没开启事务，seata帮你开启
            return new LockRetryPolicy(connectionProxy).execute(() -> {
                T result = executeAutoCommitFalse(args);
                connectionProxy.commit(); // 帮你开启事务后，通过connectionProxy来提交
                return result;
            });
        } catch (Exception e) {
            // when exception occur in finally,this exception will lost, so just print it here
            LOGGER.error("execute executeAutoCommitTrue error:{}", e.getMessage(), e);
            if (!LockRetryPolicy.isLockRetryPolicyBranchRollbackOnConflict()) {
                connectionProxy.getTargetConnection().rollback();
            }
            throw e;
        } finally {
            connectionProxy.getContext().reset();
            connectionProxy.setAutoCommit(true);
        }
    }

    protected T executeAutoCommitFalse(Object[] args) throws Exception {
        if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && isMultiPk()) {
            throw new NotSupportYetException("multi pk only support mysql!");
        }
        TableRecords beforeImage = beforeImage();
        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
        TableRecords afterImage = afterImage(beforeImage);
        prepareUndoLog(beforeImage, afterImage);
        return result;
    }
}

package io.seata.rm.datasource.exec;

public class UpdateExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
    
    public UpdateExecutor(StatementProxy<S> statementProxy, StatementCallback<T, S> statementCallback,
                        SQLRecognizer sqlRecognizer) {
        super(statementProxy, statementCallback, sqlRecognizer);
    }

}

如果选了DML类型Executer，可以在上面的executeAutoCommitFalse()中看到，主要做了以下事情：

查询前镜像（select for update，因此此时获得本地锁）

package io.seata.rm.datasource.exec;

public class UpdateExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
    
    private static final boolean ONLY_CARE_UPDATE_COLUMNS = CONFIG.getBoolean(
            ConfigurationKeys.TRANSACTION_UNDO_ONLY_CARE_UPDATE_COLUMNS, DefaultValues.DEFAULT_ONLY_CARE_UPDATE_COLUMNS); // 默认为true

    @Override
    protected TableRecords beforeImage() throws SQLException {
        ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
        TableMeta tmeta = getTableMeta();
        String selectSQL = buildBeforeImageSQL(tmeta, paramAppenderList);
        // SELECT id, count FROM storage_tbl WHERE id = ? FOR UPDATE
        return buildTableRecords(tmeta, selectSQL, paramAppenderList);
    }

    private String buildBeforeImageSQL(TableMeta tableMeta, ArrayList<List<Object>> paramAppenderList) {
        SQLUpdateRecognizer recognizer = (SQLUpdateRecognizer) sqlRecognizer;
        List<String> updateColumns = recognizer.getUpdateColumns();
        StringBuilder prefix = new StringBuilder("SELECT ");
        StringBuilder suffix = new StringBuilder(" FROM ").append(getFromTableInSQL());
        String whereCondition = buildWhereCondition(recognizer, paramAppenderList);
        if (StringUtils.isNotBlank(whereCondition)) {
            suffix.append(WHERE).append(whereCondition);
        }
        String orderBy = recognizer.getOrderBy();
        if (StringUtils.isNotBlank(orderBy)) {
            suffix.append(orderBy);
        }
        ParametersHolder parametersHolder = statementProxy instanceof ParametersHolder ? (ParametersHolder)statementProxy : null;
        String limit = recognizer.getLimit(parametersHolder, paramAppenderList);
        if (StringUtils.isNotBlank(limit)) {
            suffix.append(limit);
        }
        suffix.append(" FOR UPDATE");
        StringJoiner selectSQLJoin = new StringJoiner(", ", prefix.toString(), suffix.toString());
        if (ONLY_CARE_UPDATE_COLUMNS) {
            if (!containsPK(updateColumns)) {// 如果本次更新的行不包含主键，那select for update的时候加上主键
                selectSQLJoin.add(getColumnNamesInSQL(tableMeta.getEscapePkNameList(getDbType())));
            }
            for (String columnName : updateColumns) {
                selectSQLJoin.add(columnName);
            }
        } else {
            for (String columnName : tableMeta.getAllColumns().keySet()) {
                selectSQLJoin.add(ColumnUtils.addEscape(columnName, getDbType()));
            }
        }
        return selectSQLJoin.toString();
    }


    protected TableRecords buildTableRecords(TableMeta tableMeta, String selectSQL, ArrayList<List<Object>> paramAppenderList) throws SQLException {
        ResultSet rs = null;
        try (PreparedStatement ps = statementProxy.getConnection().prepareStatement(selectSQL)) { // 执行select for update，然后就拿到了本地锁
            if (CollectionUtils.isNotEmpty(paramAppenderList)) {
                for (int i = 0, ts = paramAppenderList.size(); i < ts; i++) {
                    List<Object> paramAppender = paramAppenderList.get(i);
                    for (int j = 0, ds = paramAppender.size(); j < ds; j++) {
                        ps.setObject(i * ds + j + 1, paramAppender.get(j));
                    }
                }
            }
            rs = ps.executeQuery();
            return TableRecords.buildRecords(tableMeta, rs);
        } finally {
            IOUtil.close(rs);
        }
    }
}

执行业务sql

查询后镜像

  package io.seata.rm.datasource.exec;

  public class UpdateExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
      
      @Override
      protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
          TableMeta tmeta = getTableMeta();
          if (beforeImage == null || beforeImage.size() == 0) {
              return TableRecords.empty(getTableMeta());
          }
          String selectSQL = buildAfterImageSQL(tmeta, beforeImage);
          //SELECT id, count FROM storage_tbl WHERE (id) in ( (?) )
          ResultSet rs = null;
          try (PreparedStatement pst = statementProxy.getConnection().prepareStatement(selectSQL)) {
              SqlGenerateUtils.setParamForPk(beforeImage.pkRows(), getTableMeta().getPrimaryKeyOnlyName(), pst);
              rs = pst.executeQuery();
              return TableRecords.buildRecords(tmeta, rs);
          } finally {
              IOUtil.close(rs);
          }
      }
  }

准备undoLog

public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor<T> {
    
    protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
        if (beforeImage.getRows().isEmpty() && afterImage.getRows().isEmpty()) {
            return;
        }
        if (SQLType.UPDATE == sqlRecognizer.getSQLType()) {
            if (beforeImage.getRows().size() != afterImage.getRows().size()) {
                throw new ShouldNeverHappenException("Before image size is not equaled to after image size, probably because you updated the primary keys.");
            }
        }
        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();

        TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
        String lockKeys = buildLockKey(lockKeyRecords);
        if (null != lockKeys) {
            connectionProxy.appendLockKey(lockKeys);

            SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
            connectionProxy.appendUndoLog(sqlUndoLog); // 把undoLog存到connectionProxy中，具体怎么回事上面有提过
        }
    }
}

如果你的sql是select for update则会使用SelectForUpdateExecutor（Seata代理了select for update），代理后处理的逻辑是这样的：

先执行 select for update（获取数据库本地锁）
如果处于@GlobalTransactional or @GlobalLock，检查是否有全局锁
如果有全局锁，则未开启本地事务下会rollback本地事务，再重新争抢本地锁和查询全局锁，直到全局锁释放

   package io.seata.rm.datasource.exec;

   public class SelectForUpdateExecutor<T, S extends Statement> extends BaseTransactionalExecutor<T, S> {
           @Override
            public T doExecute(Object... args) throws Throwable {
                Connection conn = statementProxy.getConnection();
                DatabaseMetaData dbmd = conn.getMetaData();
                T rs;
                Savepoint sp = null;
                boolean originalAutoCommit = conn.getAutoCommit();
                try {
                    if (originalAutoCommit) {
                        /*
                         * In order to hold the local db lock during global lock checking
                         * set auto commit value to false first if original auto commit was true
                         */
                        conn.setAutoCommit(false);
                    } else if (dbmd.supportsSavepoints()) {
                        /*
                         * In order to release the local db lock when global lock conflict
                         * create a save point if original auto commit was false, then use the save point here to release db
                         * lock during global lock checking if necessary
                         */
                        sp = conn.setSavepoint();
                    } else {
                        throw new SQLException("not support savepoint. please check your db version");
                    }

                    LockRetryController lockRetryController = new LockRetryController();
                    ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
                    String selectPKSQL = buildSelectSQL(paramAppenderList);
                    while (true) {
                        try {
                            // #870
                            // execute return Boolean
                            // executeQuery return ResultSet
                            rs = statementCallback.execute(statementProxy.getTargetStatement(), args); //执行 select for update（获取数据库本地锁）

                            // Try to get global lock of those rows selected
                            TableRecords selectPKRows = buildTableRecords(getTableMeta(), selectPKSQL, paramAppenderList);
                            String lockKeys = buildLockKey(selectPKRows);
                            if (StringUtils.isNullOrEmpty(lockKeys)) {
                                break;
                            }

                            if (RootContext.inGlobalTransaction() || RootContext.requireGlobalLock()) {
                                // Do the same thing under either @GlobalTransactional or @GlobalLock, 
                                // that only check the global lock  here.
                                statementProxy.getConnectionProxy().checkLock(lockKeys);
                            } else {
                                throw new RuntimeException("Unknown situation!");
                            }
                            break;
                        } catch (LockConflictException lce) {
                            if (sp != null) {
                                conn.rollback(sp);
                            } else {
                                conn.rollback();// 回滚，释放本地锁
                            }
                            // trigger retry
                            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;
            }



   }

`ConnectionProxy.commit()`的处理逻辑

public class ConnectionProxy extends AbstractConnectionProxy {

    private final static LockRetryPolicy LOCK_RETRY_POLICY = new LockRetryPolicy();

    private ConnectionContext context = new ConnectionContext();

    @Override
    public void commit() throws SQLException {
        try {
            LOCK_RETRY_POLICY.execute(() -> {
                doCommit();
                return null;
            });
        } catch (SQLException e) {
            if (targetConnection != null && !getAutoCommit() && !getContext().isAutoCommitChanged()) {
                rollback();
            }
            throw e;
        } catch (Exception e) {
            throw new SQLException(e);
        }
    }

    private void doCommit() throws SQLException {
        if (context.inGlobalTransaction()) {
            processGlobalTransactionCommit();
        } else if (context.isGlobalLockRequire()) {
            processLocalCommitWithGlobalLocks();
        } else {
            targetConnection.commit();
        }
    }

}

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

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

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

    public class ConnectionProxy extends AbstractConnectionProxy {

        private final static LockRetryPolicy LOCK_RETRY_POLICY = new LockRetryPolicy();

        private ConnectionContext context = new ConnectionContext();
        
        private void processGlobalTransactionCommit() throws SQLException {
            try {
                register(); // 注册分支，争全局锁
            } catch (TransactionException e) {
                recognizeLockKeyConflictException(e, context.buildLockKeys());
            }
            try {
                UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this); // undolog入库
                targetConnection.commit(); // 分支事务提交
            } catch (Throwable ex) {
                LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
                report(false);
                throw new SQLException(ex);
            }
            if (IS_REPORT_SUCCESS_ENABLE) {
                report(true);
            }
            context.reset();
        }

        private void register() throws TransactionException {
            if (!context.hasUndoLog() || !context.hasLockKey()) {
                return;
            }
            Long branchId = DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(),
                null, context.getXid(), null, context.buildLockKeys());
            context.setBranchId(branchId);
        }


    }

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

向tc查询是否有全局锁存在
让数据库commit本次事务

   public class ConnectionProxy extends AbstractConnectionProxy {

       private final static LockRetryPolicy LOCK_RETRY_POLICY = new LockRetryPolicy();

       private ConnectionContext context = new ConnectionContext();
       
       private void processLocalCommitWithGlobalLocks() throws SQLException {
           checkLock(context.buildLockKeys());
           try {
               targetConnection.commit();
           } catch (Throwable ex) {
               throw new SQLException(ex);
           }
           context.reset();
       }

       public void checkLock(String lockKeys) throws SQLException {
           if (StringUtils.isBlank(lockKeys)) {
               return;
           }
           // Just check lock without requiring lock by now.
           try {
               boolean lockable = DefaultResourceManager.get().lockQuery(BranchType.AT,
                   getDataSourceProxy().getResourceId(), context.getXid(), lockKeys);
               if (!lockable) {
                   throw new LockConflictException();
               }
           } catch (TransactionException e) {
               recognizeLockKeyConflictException(e, lockKeys);
           }
       }


   }

除了以上情况（else分支）
- 让数据库commit本次事务

介绍`RootContext`

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

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

public class RootContext {

    public static final String KEY_XID = "TX_XID";

    private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load();

    private static BranchType DEFAULT_BRANCH_TYPE;

    @Nullable
    public static BranchType getBranchType() {
        if (inGlobalTransaction()) {
            BranchType branchType = (BranchType) CONTEXT_HOLDER.get(KEY_BRANCH_TYPE);
            if (branchType != null) {
                return branchType;
            }
            //Returns the default branch type.
            return DEFAULT_BRANCH_TYPE != null ? DEFAULT_BRANCH_TYPE : BranchType.AT;
        }
        return null;
    }

    public static boolean inGlobalTransaction() {
        return CONTEXT_HOLDER.get(KEY_XID) != null;
    }

    public static void bind(@Nonnull String xid) {
        if (StringUtils.isBlank(xid)) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("xid is blank, switch to unbind operation!");
            }
            unbind();
        } else {
            MDC.put(MDC_KEY_XID, xid);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("bind {}", xid);
            }
            CONTEXT_HOLDER.put(KEY_XID, xid);
        }
    }

}

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

public class RootContext {

    public static final String KEY_GLOBAL_LOCK_FLAG = "TX_LOCK";
    public static final Boolean VALUE_GLOBAL_LOCK_FLAG = true;

    private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load();

    public static boolean requireGlobalLock() {
        return CONTEXT_HOLDER.get(KEY_GLOBAL_LOCK_FLAG) != null;
    }

    public static void bindGlobalLockFlag() {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Local Transaction Global Lock support enabled");
        }

        //just put something not null
        CONTEXT_HOLDER.put(KEY_GLOBAL_LOCK_FLAG, VALUE_GLOBAL_LOCK_FLAG);
    }

}

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

 public class ConnectionProxy extends AbstractConnectionProxy {

    private ConnectionContext context = new ConnectionContext();

    private void doCommit() throws SQLException {
        if (context.inGlobalTransaction()) {
            processGlobalTransactionCommit();
        } else if (context.isGlobalLockRequire()) {
            processLocalCommitWithGlobalLocks();
        } else {
            targetConnection.commit();
        }
    }

}

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

public class ConnectionContext {

    private String xid;

    void setXid(String xid) {
        this.xid = xid;
    }

    public boolean inGlobalTransaction() {
        return xid != null;
    }

    void bind(String xid) {
        if (xid == null) {
            throw new IllegalArgumentException("xid should not be null");
        }
        if (!inGlobalTransaction()) {
            setXid(xid);
        } else {
            if (!this.xid.equals(xid)) {
                throw new ShouldNeverHappenException();
            }
        }
    }

}

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

package io.seata.rm.datasource.exec;

public abstract class BaseTransactionalExecutor<T, S extends Statement> implements Executor<T> {

    @Override
    public T execute(Object... args) throws Throwable {
        // 那么，这里的XID哪来的呢？往后看就知道，是来自开启全局事务的时候获得的，和@GlobalTransactional有关
        String xid = RootContext.getXID(); 
        if (xid != null) {
            statementProxy.getConnectionProxy().bind(xid);
        }

        // 这里就是设置 isGlobalLockRequire的位置，和 @GlobalLock有关
        statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
        return doExecute(args);
    }

}

public class ConnectionProxy extends AbstractConnectionProxy {

   private ConnectionContext context = new ConnectionContext();

    public void bind(String xid) {
        context.bind(xid);
    }

    public void setGlobalLockRequire(boolean isLock) {
        context.setGlobalLockRequire(isLock);
    }

}

如何判断isGlobalLockRequire()？

public class ConnectionContext {

    private boolean isGlobalLockRequire;

    boolean isGlobalLockRequire() {
       return isGlobalLockRequire;
    }

    void setGlobalLockRequire(boolean isGlobalLockRequire) {
        this.isGlobalLockRequire = isGlobalLockRequire;
    }

}

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

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

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

public class GlobalTransactionalInterceptor implements ConfigurationChangeListener, MethodInterceptor {

    @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);
        if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
            final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
            final GlobalTransactional globalTransactionalAnnotation =
                getAnnotation(method, targetClass, GlobalTransactional.class);
            final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);
            boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes);
            if (!localDisable) {
                if (globalTransactionalAnnotation != null) {
                    return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);// 处理 @GlobalTransactional
                } else if (globalLockAnnotation != null) {
                    return handleGlobalLock(methodInvocation, globalLockAnnotation); // 处理 @GlobalLock
                }
            }
        }
        return methodInvocation.proceed();
    }

}

先看处理`@GlobalTransactional`

public class GlobalTransactionalInterceptor implements ConfigurationChangeListener, MethodInterceptor {

    private final TransactionalTemplate transactionalTemplate = new TransactionalTemplate();

    Object handleGlobalTransaction(final MethodInvocation methodInvocation,
        final GlobalTransactional globalTrxAnno) throws Throwable {
        
        //...
        try {
            return transactionalTemplate.execute(...);
        } catch (TransactionalExecutor.ExecutionException e) {
          // ...
        } finally {
            //...
        }
    }
}

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

public class TransactionalTemplate {

    public Object execute(TransactionalExecutor business) throws Throwable {
        // 1. Get transactionInfo
        //...
        // 1.1 Get current transaction, if not null, the tx role is 'GlobalTransactionRole.Participant'.
        GlobalTransaction tx = GlobalTransactionContext.getCurrent();

        // 1.2 Handle the transaction propagation.
        // ...

            // 1.3 If null, create new transaction with role 'GlobalTransactionRole.Launcher'.
            if (tx == null) {
                tx = GlobalTransactionContext.createNew();
            }

           //...

            try {
                // 2. If the tx role is 'GlobalTransactionRole.Launcher', send the request of beginTransaction to TC,
                //    else do nothing. Of course, the hooks will still be triggered.
                beginTransaction(txInfo, tx);

                Object rs;
                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
                //...
            }
        } finally {
            // If the transaction is suspended, resume it.
           // ...
        }
    }


    private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
        try {
            triggerBeforeBegin();
            tx.begin(txInfo.getTimeOut(), txInfo.getName());
            triggerAfterBegin();
        } catch (TransactionException txe) {
            throw new TransactionalExecutor.ExecutionException(tx, txe,
                TransactionalExecutor.Code.BeginFailure);

        }
    }


}

public class DefaultGlobalTransaction implements GlobalTransaction {

    @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();
        String currentXid = RootContext.getXID();
        if (currentXid != null) {
            throw new IllegalStateException("Global transaction already exists," +
                " can't begin a new global transaction, currentXid = " + currentXid);
        }
        xid = transactionManager.begin(null, null, name, timeout);
        status = GlobalStatus.Begin;
        RootContext.bind(xid); // 绑定xid
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Begin new global transaction [{}]", xid);
        }
    }
}

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

接着看处理`@GlobalLock`

public class GlobalTransactionalInterceptor implements ConfigurationChangeListener, MethodInterceptor {

    private final GlobalLockTemplate globalLockTemplate = new GlobalLockTemplate();

    Object handleGlobalLock(final MethodInvocation methodInvocation,
        final GlobalLock globalLockAnno) throws Throwable {

        return globalLockTemplate.execute(new GlobalLockExecutor() {...});
    }
}

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

public class GlobalLockTemplate {

    public Object execute(GlobalLockExecutor executor) throws Throwable {
        boolean alreadyInGlobalLock = RootContext.requireGlobalLock();
        if (!alreadyInGlobalLock) {
            RootContext.bindGlobalLockFlag();
        }

        // set my config to config holder so that it can be access in further execution
        // for example, LockRetryController can access it with config holder
        GlobalLockConfig myConfig = executor.getGlobalLockConfig();
        GlobalLockConfig previousConfig = GlobalLockConfigHolder.setAndReturnPrevious(myConfig);

        try {
            return executor.execute();
        } finally {
            // only unbind when this is the root caller.
            // otherwise, the outer caller would lose global lock flag
            if (!alreadyInGlobalLock) {
                RootContext.unbindGlobalLockFlag();
            }

            // if previous config is not null, we need to set it back
            // so that the outer logic can still use their config
            if (previousConfig != null) {
                GlobalLockConfigHolder.setAndReturnPrevious(previousConfig);
            } else {
                GlobalLockConfigHolder.remove();
            }
        }
    }
}

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

标签：事务,return,Seata,statementProxy,class,new,null,public,隔离
From： https://blog.csdn.net/qq_36070104/article/details/144011456

从代理数据源说起

`StatementProxy.executeXXX()`的处理逻辑

`ConnectionProxy.commit()`的处理逻辑

@GlobalTransactional的作用

@GlobalLock + select for update的作用

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

怎么用Seata防止脏写？

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

办法二： @GlobalLock + select for update

如何防止脏读？

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

源码展示

DataSourceProxy的作用

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

`StatementProxy.execute()`的处理逻辑

`ConnectionProxy.commit()`的处理逻辑

介绍`RootContext`

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

先看处理`@GlobalTransactional`

接着看处理`@GlobalLock`

相关文章

赞助商

阅读排行

Seata事务隔离

从代理数据源说起​

StatementProxy.executeXXX()的处理逻辑​

ConnectionProxy.commit()的处理逻辑​

@GlobalTransactional的作用​

@GlobalLock + select for update的作用​

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

怎么用Seata防止脏写？​

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

办法二： @GlobalLock + select for update​

如何防止脏读？​

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

源码展示

DataSourceProxy的作用​

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

StatementProxy.execute()的处理逻辑​

ConnectionProxy.commit()的处理逻辑​

介绍RootContext​

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

先看处理@GlobalTransactional​

接着看处理@GlobalLock​

相关文章

赞助商

阅读排行

从代理数据源说起

`StatementProxy.executeXXX()`的处理逻辑

`ConnectionProxy.commit()`的处理逻辑

@GlobalTransactional的作用

@GlobalLock + select for update的作用

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

怎么用Seata防止脏写？

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

办法二： @GlobalLock + select for update

如何防止脏读？

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

DataSourceProxy的作用

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

`StatementProxy.execute()`的处理逻辑

`ConnectionProxy.commit()`的处理逻辑

介绍`RootContext`

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

先看处理`@GlobalTransactional`

接着看处理`@GlobalLock`