六、StatementProxy生成undoLog

所有文章

https://www.cnblogs.com/lay2017/p/12485081.html

正文

前两篇文章中 ，我们分别看到了DataSourceProxy被注册为Resource，ConnectionProxy注册分支事务到全局事务当中。

本文继续数据源代理的StatementProxy的内容

我们知道，Statement是由Connection生成的。那么StatementProxy将由ConnectionProxy生成

StatementProxy的uml类图

我们先看看类图

StatementProxy间接实现了Statement，所以也表示一个sql语句的执行对象。向下到最底，又被PreparedStatementProxy扩展。既然StatementProxy作为Statement的代理对象而存在，那么它组合了哪些原始的对象呢？

我们打开AbstractStatementProxy

public abstract class AbstractStatementProxy<T extends Statement> implements Statement {
    /**
     * ConnectionProxy对象
     */
    protected AbstractConnectionProxy connectionProxy;
    /**
     * 原始Statement对象
     */
    protected T targetStatement;
    /**
     * 原始的待执行sql
     */
    protected String targetSQL;

    public AbstractStatementProxy(AbstractConnectionProxy connectionProxy, T targetStatement, String targetSQL) throws SQLException {
        this.connectionProxy = connectionProxy;
        this.targetStatement = targetStatement;
        this.targetSQL = targetSQL;
    }
}

可以看到，除了被代理的Statement对象之外，在构造过程中还把ConnectionProxy和sql给包含进来了。

execute方法

Statement最核心的方法就是执行方法，我们以execute为例，看看该StatementProxy在代理Statement的时候做了哪些事

@Override
public boolean execute(String sql) throws SQLException {
    this.targetSQL = sql;
    return ExecuteTemplate.execute(this, new StatementCallback<Boolean, T>() {
        @Override
        public Boolean execute(T statement, Object... args) throws SQLException {
            // 回调，执行原始的Statement的execute方法
            return statement.execute((String) args[0]);
        }
    }, sql);
}

可见，ExecuteTemplate的execute方法包含了核心的代理逻辑。跟进ExecuteTemplate.execute()

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);
}

继续跟进execute方法，execute方法会解析当前待执行的sql语句，并判断sql语句的类型是insert、update、delete...通过类型来选择对应的Executor，然后执行。

public static <T, S extends Statement> T execute(SQLRecognizer sqlRecognizer,
                                                 StatementProxy<S> statementProxy,
                                                 StatementCallback<T, S> statementCallback,
                                                 Object... args) throws SQLException {
    // 不存在全局事务，不存在全局锁
    if (!RootContext.inGlobalTransaction() && !RootContext.requireGlobalLock()) {
        // 直接执行原始方法
        return statementCallback.execute(statementProxy.getTargetStatement(), args);
    }

    if (sqlRecognizer == null) {
        // sql解析
        sqlRecognizer = SQLVisitorFactory.get(statementProxy.getTargetSQL(), statementProxy.getConnectionProxy().getDbType());
    }

    Executor<T> executor = null;
    if (sqlRecognizer == null) {
        executor = new PlainExecutor<T, S>(statementProxy, statementCallback);
    } else {
        // 根据解析出来的类型选择Executor
        switch (sqlRecognizer.getSQLType()) {
            case INSERT:
                executor = new InsertExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                break;
            case UPDATE:
                executor = new UpdateExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                break;
            case DELETE:
                executor = new DeleteExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                break;
            case SELECT_FOR_UPDATE:
                executor = new SelectForUpdateExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                break;
            default:
                executor = new PlainExecutor<T, S>(statementProxy, statementCallback);
                break;
        }
    }
    T rs = null;
    try {
        // executor执行
        rs = executor.execute(args);
    } catch (Throwable ex) {
        if (!(ex instanceof SQLException)) {
            ex = new SQLException(ex);
        }
        throw (SQLException)ex;
    }
    return rs;
}

跟进BaseTransactionalExecutor的execute方法

public Object execute(Object... args) throws Throwable {
    // 如果在全局事务当中
    if (RootContext.inGlobalTransaction()) {
        String xid = RootContext.getXID();
        // 将ConnectionProxy绑定XID
        statementProxy.getConnectionProxy().bind(xid);
    }
    // 设置全局锁标识
    if (RootContext.requireGlobalLock()) {
        statementProxy.getConnectionProxy().setGlobalLockRequire(true);
    } else {
        statementProxy.getConnectionProxy().setGlobalLockRequire(false);
    }
    
    return doExecute(args);
}

注意到，XID是在这里被绑定到ConnectionProxy的。

上一篇文章我们说过，ConnectionProxy在commit的时候会根据是否存在XID来判断要不要注册事务分支之类的操作。所以，这里的bind方法将决定是否当前Connection处于全局事务当中。

接下来继续跟进AbstractDMLBaseExecutor的doExecute方法

@Override
public T doExecute(Object... args) throws Throwable {
    AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
    // 判断自动提交标志位
    if (connectionProxy.getAutoCommit()) {
        return executeAutoCommitTrue(args);
    } else {
        return executeAutoCommitFalse(args);
    }
}

doExecute方法根据autoCommit标志位来选择执行方法，执行方法中包含了StatementProxy的生成undoLog的核心点。

我们选择相对比较清晰的executeAutoCommitFalse这个方法看看

protected T executeAutoCommitFalse(Object[] args) throws Exception {
    // 前置生成数据记录
    TableRecords beforeImage = beforeImage();
    // 执行sql语句
    T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
    // 生成后置数据记录
    TableRecords afterImage = afterImage(beforeImage);
    // 生成undoLog
    prepareUndoLog(beforeImage, afterImage);
    return result;
}

beforeImage和afterImage由子类实现，分别生成sql执行前后的数据记录。

prepareUndoLog将根据前后的数据记录生成一份SQLUndoLog，顾名思义undo_log就是用于在分支事务rollback的时候做补偿操作的。

flushUndoLog持久化到数据库

undoLog到这里还只是存在于内存当中，当ConnectionProxy的执行完register注册分支事务以后，在本地事务提交之前就会将undoLog给flush到数据库中。

我们跟进AbstractUndoLogManager的flushUndoLogs方法

@Override
public void flushUndoLogs(ConnectionProxy cp) throws SQLException {
    ConnectionContext connectionContext = cp.getContext();
    String xid = connectionContext.getXid();
    long branchID = connectionContext.getBranchId();

    // 构建一个分支事务undoLog对象
    BranchUndoLog branchUndoLog = new BranchUndoLog();
    branchUndoLog.setXid(xid);
    branchUndoLog.setBranchId(branchID);
    // 获取undoLog内存对象
    branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());

    UndoLogParser parser = UndoLogParserFactory.getInstance();
    // 解析器解析后生成字节码
    byte[] undoLogContent = parser.encode(branchUndoLog);
    // insert到数据库表当中
    insertUndoLogWithNormal(xid, branchID, buildContext(parser.getName()), undoLogContent, cp.getTargetConnection());
}

核心逻辑就是获取上文中的SqlUndoLog集合，然后持久化到数据库undo_log表当中，给分支事务的rollback预留。

总结

到这里，本文就结束了。可以看到，StatementProxy主要就是在sql执行后生成SqlUndoLog。而在ConnectionProxy的commit方法里，将分支事务register以后会把SqlUndoLog给flush到数据库当中。

undoLog的作用就是用于rollback操作补偿的，所以这里就是预留的一个undo的操作。

我们再回顾一下数据源代理的三个核心对象

1）DataSourceProxy注册了Resource

2）ConnectionProxy注册分支事务，持久化UndoLog，并先提交本地事务

3）StatementProxy执行sql，生成undoLog

我们可以暂时简单地把数据源代理这一过程理解为，主要是为rollback预留undo操作，如果执行顺利异步删除undoLog即可。