Seata解析

文章23 |   阅读 16660 |   点赞0

来源：https://blog.csdn.net/weixin_38308374/category_10261111.html

Seata解析-RM执行SQL语句原理详解

转x33g5p2x  于2021-12-21 转载在 其他  
字(14.4k)|赞(0)|评价(0)|浏览(817)

本文基于seata 1.3.0版本

seata使用DataSourceProxy对数据源进行代理，程序中执行分支事务相关的操作都是基于该代理数据源完成的。本文将详细分析RM如何基于代理数据源完成一条SQL语句的执行。
使用JDBC操作数据库的伪代码一般为：

Connection conn=dataSource.getConnection();
Statement state=conn.createStatement();
state.execute(sql);
或者
Connection conn=dataSource.getConnection();
PrepareStatement state=conn.prepareStatement(sql);
state.execute();

conn.commit();
conn.rollback();
conn.close();

接下来也将按照上述代码流程分析RM如何执行SQL。

一、分支事务获取数据库连接

数据库连接对象是通过DataSourceProxy得到的。

public ConnectionProxy getConnection() throws SQLException {
		//targetDataSource是被代理的数据源
        Connection targetConnection = targetDataSource.getConnection();
        //创建一个连接代理对象
        return new ConnectionProxy(this, targetConnection);
    }

ConnectionProxy代理了真实的数据库连接。该类的继承结构如下：

二、创建Statement或者PrepareStatement

创建完连接对象后，就要创建语句对象，下面代码是ConnectionProxy中createStatement和prepareStatement方法，当然还有其他重载的方法，这里不再展示，基本和这两个类似。

public Statement createStatement() throws SQLException {
		//调用真实连接对象获得Statement对象
        Statement targetStatement = getTargetConnection().createStatement();
        //创建Statement的代理
        return new StatementProxy(this, targetStatement);
    }
    public PreparedStatement prepareStatement(String sql) throws SQLException {
    	//数据库类型，比如mysql、oracle等
        String dbType = getDbType();
        PreparedStatement targetPreparedStatement = null;
        //如果是AT模式且开启全局事务，那么就会进入if分支
        //详细可以参见类ApacheDubboTransactionPropagationFilter
        if (StringUtils.equals(BranchType.AT.name(), RootContext.getBranchType())) {
            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);
                    //使用下面代码创建PreparedStatement原因是要数据库返回插入数据的主键值
                    targetPreparedStatement = getTargetConnection().prepareStatement(sql,pkNameArray);
                }
            }
        }
        if (targetPreparedStatement == null) {
            targetPreparedStatement = getTargetConnection().prepareStatement(sql);
        }
        return new PreparedStatementProxy(this, targetPreparedStatement, sql);
    }

从上面的代码可以看出seata对Statement和PreparedStatement均提供了对应的代理对象。

三、分支事务执行SQL

创建完Statement和PreparedStatement，接下来要执行SQL语句了。
先来看一下StatementProxy如何执行SQL。

1、StatementProxy执行

下面代码是StatementProxy中部分执行SQL语句的方法：

@Override
    public ResultSet executeQuery(String sql) throws SQLException {
        this.targetSQL = sql;
        return ExecuteTemplate.execute(this, (statement, args) -> statement.executeQuery((String) args[0]), sql);
    }

    @Override
    public int executeUpdate(String sql) throws SQLException {
        this.targetSQL = sql;
        return ExecuteTemplate.execute(this, (statement, args) -> statement.executeUpdate((String) args[0]), sql);
    }

    @Override
    public boolean execute(String sql) throws SQLException {
        this.targetSQL = sql;
        return ExecuteTemplate.execute(this, (statement, args) -> statement.execute((String) args[0]), sql);
    }

其他执行SQL语句的方法与上面三个方法都是类似的，都是调用ExecuteTemplate.execute方法，下面来看一下ExecuteTemplate类：

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 {
        //如果事务不需要全局事务锁，而且不是在AT模式下，那么直接使用原生的Statement对象执行SQL
        if (!RootContext.requireGlobalLock() && !StringUtils.equals(BranchType.AT.name(), RootContext.getBranchType())) {
            return statementCallback.execute(statementProxy.getTargetStatement(), args);
        }
        //得到数据库类型
        String dbType = statementProxy.getConnectionProxy().getDbType();
        if (CollectionUtils.isEmpty(sqlRecognizers)) {
            //sqlRecognizers可以理解为SQL语句的解析器，通过它可以获得SQL语句表名、相关的列名、类型等信息
            sqlRecognizers = SQLVisitorFactory.get(
                    statementProxy.getTargetSQL(),
                    dbType);
        }
        Executor<T> executor;
        if (CollectionUtils.isEmpty(sqlRecognizers)) {
            //如果seata没有找到合适的SQL语句解析器，那么便创建简单执行器PlainExecutor，
            //PlainExecutor直接使用原生的Statement对象执行SQL
            executor = new PlainExecutor<>(statementProxy, statementCallback);
        } else {
            if (sqlRecognizers.size() == 1) {
                SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
                switch (sqlRecognizer.getSQLType()) {
                    //下面根据是增、删、改、加锁查询、普通查询分别创建对应的处理器
                    case INSERT:
                        //本文使用的是MySQL，下面代码加载的执行器是MySQLInsertExecutor
                        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 {
                //下面这个执行器可以处理一条SQL语句包含多个delete、update语句，
                //比如使用Spring的jdbcTemplate执行下面的的SQL语句：
                //jdbcTemplate.update("update account_tbl set money = money - ? where user_id = ?;update account_tbl set money = money - ? where user_id = ?", new Object[] {money, userId,"U10000",money,"U1000"});
                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;
    }

}

从ExecuteTemplate中可以看到，seata将SQL语句的执行委托给了不同的执行器。seata提供了6个执行器。

执行器作用
UpdateExecutor执行update语句
InsertExecutor执行insert语句
DeleteExecutor执行delete语句
SelectForUpdateExecutor执行select for update语句
PlainExecutor执行普通查询语句
MultiExecutor复合执行器，在一条SQL语句中执行多条语句

执行器的继承结构如下：

下面分别介绍这些执行器。

1.1、PlainExecutor

PlainExecutor是最简单的执行器，它是直接调用原生Statement对象执行SQL语句。

public T execute(Object... args) throws Throwable {
        //statementCallback是在StatementProxy中以lamda表达式定义的
        return statementCallback.execute(statementProxy.getTargetStatement(), args);
    }

1.2、InsertExecutor

InsertExecutor根据数据库的不同，它有多个实现类。本文以MySQLInsertExecutor为例介绍。
MySQLInsertExecutor.executor方法继承自父类BaseTransactionalExecutor：

public T execute(Object... args) throws Throwable {
        if (RootContext.inGlobalTransaction()) {
            String xid = RootContext.getXID();
            //绑定XID，将XID设置到连接上下文中ConnectionContext
            statementProxy.getConnectionProxy().bind(xid);
        }
        //对于分支事务来说，RootContext.requireGlobalLock()返回的是false，因为分支事务不需要全局锁
        //是否需要全局锁标示也是设置到连接上下文中
        statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
        return doExecute(args);
    }

最后的doExecute方法调用的是AbstractDMLBaseExecutor类的：

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 {
            //设置连接非自动提交，接下来事务由seata控制，如果执行过程没有异常，seata会自动将事务提交
            connectionProxy.setAutoCommit(false);
            return new LockRetryPolicy(connectionProxy).execute(() -> {
                T result = executeAutoCommitFalse(args);
                connectionProxy.commit();
                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 {
            //事务执行完，将连接上下文的XID字段置空，并恢复自动提交设置
            connectionProxy.getContext().reset();
            connectionProxy.setAutoCommit(true);
        }
    }
    protected T executeAutoCommitFalse(Object[] args) throws Exception {
        if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && getTableMeta().getPrimaryKeyOnlyName().size() > 1)
        {
            throw new NotSupportYetException("multi pk only support mysql!");
        }
        //保存事务执行前的快照，beforeImage由子类实现
        TableRecords beforeImage = beforeImage();
        //调用原生的Statement对象执行SQL语句
        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
        //保存事务执行后的快照，afterImage由子类实现
        TableRecords afterImage = afterImage(beforeImage);
        prepareUndoLog(beforeImage, afterImage);
        return result;
    }
    //生成回滚日志，回滚日志保存事务上下文ConnectionContext中
    protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
        if (beforeImage.getRows().isEmpty() && afterImage.getRows().isEmpty()) {
            return;
        }

        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();

        TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
        String lockKeys = buildLockKey(lockKeyRecords);
        //记录加锁信息，将加锁的数据保存到连接上下文中
        connectionProxy.appendLockKey(lockKeys);

        SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
        //将回滚记录保存到连接上下文中
        connectionProxy.appendUndoLog(sqlUndoLog);
    }
    protected SQLUndoLog buildUndoItem(TableRecords beforeImage, TableRecords afterImage) {
        SQLType sqlType = sqlRecognizer.getSQLType();
        String tableName = sqlRecognizer.getTableName();

        SQLUndoLog sqlUndoLog = new SQLUndoLog();
        sqlUndoLog.setSqlType(sqlType);
        sqlUndoLog.setTableName(tableName);
        sqlUndoLog.setBeforeImage(beforeImage);
        sqlUndoLog.setAfterImage(afterImage);
        return sqlUndoLog;
    }

在方法executeAutoCommitTrue中，创建了对象LockRetryPolicy，该对象定义当出现锁冲突时，事务的重试策略，如果设置client.rm.lock.retryPolicyBranchRollbackOnConflict=true，那么当出现锁冲突时，seata会不断的重试，直到记录加锁成功，否则直接抛出加锁异常。
在executeAutoCommitTrue中创建了对象LockRetryPolicy，但是executeAutoCommitFalse方法没有创建，这是为什么？
这是因为锁冲突是在事务提交的时候判断的，也就是提交分支事务时，首先要发送分支事务注册请求到TC，TC根据分支事务加锁的记录判断是否已经有其他事务加锁了，如果有则返回LockConflictException，禁止事务提交。
事务提交的内容本文下面介绍。先看一下executeAutoCommitFalse中调用的beforeImage、afterImage方法。这两个方法需要由子类来实现。
MySQLInsertExecutor的这两个方法是在父类BaseInsertExecutor中实现的。

protected TableRecords beforeImage() throws SQLException {
        //在数据插入前，记录是空的，所以创建一个空的TableRecords对象
        //getTableMeta()用于获得表的元数据，也就是表结构。该方法在后文介绍
        return TableRecords.empty(getTableMeta());
    }
    protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
        //获得插入数据的主键值，如果插入语句中有主键，那么可以根据insert语句分析出来，
        // 如果插入语句中没有包含主键，那么调用数据库返回自动生成的主键值
        Map<String,List<Object>> pkValues = getPkValues();
        //构建select语句，将插入的数据从表里面查询出来，根据查询结果创建TableRecords对象
        //TableRecords对象包含了每条记录的每个字段值及其类型
        TableRecords afterImage = buildTableRecords(pkValues);
        if (afterImage == null) {
            throw new SQLException("Failed to build after-image for insert");
        }
        return afterImage;
    }

beforeImage和afterImage生成事务执行前后的快照，然后将他们保存在事务上下文ConnectionContext中。

1.3、UpdateExecutor

UpdateExecutor执行器在执行execut方法时，与InsertExecutor基本是一致的，所不同的是beforeImage和afterImage这两个方法的实现。下面看一下这两个方法在UpdateExecutor中如何实现。

protected TableRecords beforeImage() throws SQLException {
        ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
        //获得表结构
        TableMeta tmeta = getTableMeta();
        //根据update语句构建查询SQL语句，将被更新的记录查询出来，然后构建TableRecords对象
        String selectSQL = buildBeforeImageSQL(tmeta, paramAppenderList);
        return buildTableRecords(tmeta, selectSQL, paramAppenderList);
    }
    protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
        TableMeta tmeta = getTableMeta();
        if (beforeImage == null || beforeImage.size() == 0) {
            //如果事务执行前的快照为null，表示没有要更新的记录
            return TableRecords.empty(getTableMeta());
        }
        //构建查询SQL语句，查询条件为主键，获取所有的字段值
        String selectSQL = buildAfterImageSQL(tmeta, beforeImage);
        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);
        }
    }

1.4、DeleteExecutor

DeleteExecutor相当于InsertExecutor的逆过程，beforeImage将要删除的记录查询出来，然后生成TableRecords对象，afterImage则与MySQLInsertExecutor的afterImage完全一样。

1.5、SelectForUpdateExecutor

SelectForUpdateExecutor用于执行select for update语句，执行前后不改变数据，所以该执行器与前面不同的是它不用保存事务快照。
SelectForUpdateExecutor在执行的时候，首先将连接设置为非自动提交，然后执行select for update语句，根据查询结果找出要加锁的记录的主键值，之后调用ConnectionProxy.checkLock(主键值)方法向TC发送全局事务锁状态查询请求检查这些记录是否已经被其他事务加锁了，如果加锁了，则重新发起查询，直到成功为止。

1.6、MultiExecutor

MultiExecutor是要在一条SQL语句中执行多条delete、update语句，所以相当于多次重复调用DeleteExecutor和UpdateExecutor。

2、PreparedStatementProxy执行

PreparedStatementProxy执行SQL语句与StatementProxy是一样的，都是调用ExecuteTemplate.execute方法通过执行器完成快照的保存和SQL语句的执行。
下面是PreparedStatementProxy的部分方法。

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

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

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

四、分支事务提交

事务提交调用ConnectionProxy的commit方法。

public void commit() throws SQLException {
        try {
            //使用锁重试策略LockRetryPolicy执行提交
            LOCK_RETRY_POLICY.execute(() -> {
                doCommit();
                return null;
            });
        } catch (SQLException e) {
            throw e;
        } catch (Exception e) {
            throw new SQLException(e);
        }
    }
    private void doCommit() throws SQLException {
        //如果是分支事务，执行第一个if分支
        if (context.inGlobalTransaction()) {
            processGlobalTransactionCommit();
        } else if (context.isGlobalLockRequire()) {
            processLocalCommitWithGlobalLocks();
        } else {
            targetConnection.commit();
        }
    }
    private void processGlobalTransactionCommit() throws SQLException {
        try {
            //向TC发送分支事务注册请求BranchRegisterRequest
            //并从TC获得分支事务ID，将事务ID设置到连接上下文ConnectionContext
            register();
        } catch (TransactionException e) {
            //如果分支事务要加锁的数据，已经被其他事务加了锁，那么会抛出LockConflictException异常
            recognizeLockKeyConflictException(e, context.buildLockKeys());
        }
        try {
            //将事务执行前快照和执行后快照保存到数据库undo_log表中
            UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
            //提交事务，下面的语句执行完成意味着更改写入了数据库
            targetConnection.commit();
        } catch (Throwable ex) {
            LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
            //向TC发送BranchReportRequest请求，报告分支事务一阶段成功
            report(false);
            throw new SQLException(ex);
        }
        if (IS_REPORT_SUCCESS_ENABLE) {
            //向TC发送BranchReportRequest请求，报告分支事务一阶段成功
            report(true);
        }
        //清除上下文中的XID
        context.reset();
    }

分支事务提交相对比较简单。首先向TC注册分支事务，对修改的记录加锁，之后保存快照数据到数据库的回滚表中，最后调用原生数据库连接对象提交数据库事务。

五、分支事务回滚

事务回滚首先调用原生连接回滚事务，之后向TC报告分支事务执行失败。

public void rollback() throws SQLException {
        //调用原生连接回滚事务
        targetConnection.rollback();
        if (context.inGlobalTransaction() && context.isBranchRegistered()) {
            //如果当前分支事务处于全局事务中且已经向TC注册过该分支事务，
            //那么就向TC发送BranchReportRequest请求报告分支事务执行失败
            report(false);
        }
        //清除连接上下文中的XID
        context.reset();
    }

六、分支事务关闭连接

连接关闭是直接调用原生连接对象的close方法。

public void close() throws SQLException {
        targetConnection.close();
    }

七、总结

seata对Connection、Statement、PreparedStatement等都提供了对应的代理对象。这样seata可以方便的将自己需要的逻辑嵌入到事务的执行过程中。
seata执行SQL语句前先保存数据库快照，执行之后再保存一次快照，这样就记录了SQL语句执行前后数据变化情况，以便于事务回滚。之后在事务提交的时候，检查本事务要加锁的数据是否已经被其他事务加锁，如果没有则本分支事务可以顺利提交。同时前面的两次数据库快照也会保存到undo_log表中。

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_38308374/article/details/108805933

内容来源于网络，如有侵权，请联系作者删除！

Javamysql分布式事务seata

相关文章

    查看更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com