Fescar源码阅读-神奇的UndoLog(一)

本文梳理了Fescar生成undoLog的流程和源码，项目不停迭代，本文源码仅供参考。

Fescar处理分布式事务采用是二段提交的方案，但是Fescar对二段提交的流程，尤其是锁的流程进行了改进，大幅度提升了性能。
参考官网图示：

XA-来源官网

Fescar-来源官网

Fescar尽可能的缩短了锁的持有时间，本地事务在phase1结束时就可以提交，并释放本地锁；
如果TC决议全局commit，在phase2开始阶段，全局锁就被释放。（当然这又导致全局事务的隔离级别实际上是READ_UNCOMMITED,这方面的问题我们后面再谈，按FESCAR的理念：微服务场景产生的分布式事务，绝大部分应用在读已提交的隔离级别下工作是没有问题的。而实际上，这当中又有绝大多数的应用场景，实际上工作在读未提交的隔离级别下同样没有问题。）

这一做法的好处是极大的降低了对数据库资源的占用时间，提升了系统的吞吐量。
但是也引入了一个问题：Phase1 即提交的情况下，Phase2 如何回滚？

从上文梳理的事务流程中可以看到，Fescar通过DataSourceRM为本地事务自动生成undolog的方式，解决了这一问题。
听起来真的很神奇！开发人员什么都不用管，方向大胆的commit，除了问题Fescar自动执行数据的反向操作，将数据回滚到事务提交前的状态，这也太爽了！
刨根究底，正本清源，作为码农的我们刚刚老本行，从它的源码上看看，这一神奇的事情是如何做到的。

SQL的执行

JDBC中SQL的执行是通过Statment/PreparedStatement来实现的。Fescar也一样，通过提供了对应的Proxy代理了具体的Statement实现。在Statement执行的时，搞一点小动作~
下图为PreparedStatementProxy的继承关系，可以看到Statment和PreparedStatement都有对应的Proxy实现，

PreparedStatementProxy

代码很简单，PreparedStatementProxy所有操作都委托给了ExecuteTemplate去执行。

    @Override
    public boolean execute() throws SQLException {
        return ExecuteTemplate.execute(this, new StatementCallback<Boolean, PreparedStatement>() {
            @Override
            public Boolean execute(PreparedStatement statement, Object... args) throws SQLException {
                return statement.execute();
            }
        });
    }

    @Override
    public ResultSet executeQuery() throws SQLException {
        return ExecuteTemplate.execute(this, new StatementCallback<ResultSet, PreparedStatement>() {
            @Override
            public ResultSet execute(PreparedStatement statement, Object... args) throws SQLException {
                return statement.executeQuery();
            }
        });
    }

ExecuteTemplate

再看看ExecuteTemplate，也很简单滴，识别当前statement需要执行的SQL是什么类型（insert\update\delete等等），然后继续委托给具体的Executor去执行。

//ExecuteTemplate#execute(SQLRecognizer, StatementProxy<S>, StatementCallback<T,S>, java.lang.Object...)

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(statementProxy, statementCallback, sqlRecognizer);
                break;
            default:
                executor = new PlainExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                break;
        }

XXXXExecutor

看看InsertExecutor的继承关系，其他的Executor都类似（除了PlainExecutor不谈）。
关系也非常清晰：事务处理->DML处理->Insert处理

InsertExecutor

BaseTransactionalExecutor
主要工作：在当前事务的Context中绑定了全局事务ID：XID。

    // BaseTransactionalExecutor
    @Override
    public Object execute(Object... args) throws Throwable {
        String xid = RootContext.getXID();
        statementProxy.getConnectionProxy().bind(xid);
        return doExecute(args);
    }

    protected abstract Object doExecute(Object... args) throws Throwable;

AbstractDMLBaseExecutor
在statement执行前后分别生成beforeImage、afterImage，并调用prepareUndoLog()生成undoLog。
beforeImage、afterImage为abstract方法，由具体的executor去实现。

// AbstractDMLBaseExecutor
 @Override
    public T doExecute(Object... args) throws Throwable {
        AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        if (connectionProxy.getAutoCommit()) {
        //  executeAutoCommitTrue和executeAutoCommitFalse类似，最终还是会调用executeAutoCommitFalse。
            return executeAutoCommitTrue(args);
        } else {
            return executeAutoCommitFalse(args);
        }
    }

    protected T executeAutoCommitFalse(Object[] args) throws Throwable {
        TableRecords beforeImage = beforeImage();
        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
        TableRecords afterImage = afterImage(beforeImage);
        // 很语义化的接口，根据sql执行前和执行后的数据快照，定然能够生成对应的undolog！！！ 
        super.prepareUndoLog(beforeImage, afterImage);
        return result;
    }

   protected abstract TableRecords beforeImage() throws SQLException;

   protected abstract TableRecords afterImage(TableRecords beforeImage) throws SQLException;

很明显，为了生成undoLog，InsertExecutor、UpdateExecutor等之类的任务就是实现自己的beforeImage()和afterImage()，生成statement执行前后的数据快照。
最终依据数据Image生成undoLog。

undoLog

看一下如何生成undoLog

// BaseTransactionalExecutor
public void prepareUndoLog(SQLType sqlType, String tableName, TableRecords beforeImage, TableRecords afterImage) throws SQLException {
        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
        // lock key在这里生成了，由表名和主键组成
        String lockKeys = buildLockKey(lockKeyRecords);
        connectionProxy.appendLockKey(lockKeys);

        SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
        //生成的undolog最终都要交给connectionProxy,毕竟人家负责回滚呢
        connectionProxy.appendUndoLog(sqlUndoLog);
    }
//构建SQLUndoLog
private SQLUndoLog buildUndoItem(SQLType sqlType, String tableName, TableRecords beforeImage, TableRecords afterImage) {
        SQLUndoLog sqlUndoLog = new SQLUndoLog();
        sqlUndoLog.setSqlType(sqlType);
        sqlUndoLog.setTableName(tableName);
        sqlUndoLog.setBeforeImage(beforeImage);
        sqlUndoLog.setAfterImage(afterImage);
        return sqlUndoLog;
    }
// SQLUndoLog对象
public class SQLUndoLog {
    // select,insert,update,delete......
    private SQLType sqlType;

    private String tableName;

    private TableRecords beforeImage;

    private TableRecords afterImage;
}

至此，随着本地SQL执行完成，SQL对应的undoLog也同时生成完成。
我们还注意到，生成UndoLog的同时还生成了lock keys，还记得上文的流程中，lock keys会被发送给TC的吧。
对于undoLog，将跟随本地事务一起commit，插入undo_log表中，同时记录的还包括全局事务的XID和branchID

    @Override
    public void commit() throws SQLException {
        if (context.inGlobalTransaction()) {
                register();
                if (context.hasUndoLog()) {
                    UndoLogManager.flushUndoLogs(this);// 刷入undo_log表中
                }
                targetConnection.commit();
                report(true);
                context.reset();
        } else {
            targetConnection.commit();
        }
    }

undoLog生成完成!

AfterImage

以InsertExecutor为例看一下AfterImage如何生成

    @Override
    protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
        //解析SQL
        SQLInsertRecognizer recogizier = (SQLInsertRecognizer)sqlRecognizer;
        List<String> insertColumns = recogizier.getInsertColumns();
        TableMeta tmeta = getTableMeta();
        TableRecords afterImage = null;
        // insert是否已经包含pk
        if (tmeta.containsPK(insertColumns)) {
            // insert values including PK
            List<Object> pkValues = ...//
            //从SQL中将pk抽取，然后使用ID查询insert的数据，这就是afterImage了
            afterImage = getTableRecords(pkValues);

        } else {
            // PK 是自增长的
            Map<String, ColumnMeta> pkMetaMap = getTableMeta().getPrimaryKeyMap();
            if (pkMetaMap.size() != 1) {
                throw new NotSupportYetException();
            }
            ColumnMeta pkMeta = pkMetaMap.values().iterator().next();
            if (!pkMeta.isAutoincrement()) {
                throw new ShouldNeverHappenException();
            }
          //无需解释了。。。
            ResultSet genKeys = null;
            try {
                genKeys = statementProxy.getTargetStatement().getGeneratedKeys();
            } catch (SQLException e) {
                // java.sql.SQLException: Generated keys not requested. You need to
                // specify Statement.RETURN_GENERATED_KEYS to
                // Statement.executeUpdate() or Connection.prepareStatement().
                if ("S1009".equalsIgnoreCase(e.getSQLState())) {
                    genKeys = statementProxy.getTargetStatement().executeQuery("SELECT LAST_INSERT_ID()");
                } else {
                    throw e;
                }
            }
            List<Object> pkValues = new ArrayList<>();
            while (genKeys.next()) {
                Object v = genKeys.getObject(1);
                pkValues.add(v);
            }

            afterImage = getTableRecords(pkValues);
        }

流程很清晰（updateExecutor相对复杂，但是原理相同），根本目的都是获取insert数据的主键ID，然后把数据select出来，形成afterImage。
从代码中可以看出的约束：

必须有主键
必须自增长，或者insert时直接提供ID
不支持批量插入（至少不含主键的batch insert是不支持的）

至此，undoLog已经生成，但是它何时发挥作用，怎么发挥作用，这写问题后续结合全局事务流程再来细看。
最后附上官方的一张图，这时候看就清楚多了。

ATBranch