开篇

 这篇文章的目的主要是讲解RM的执行回滚的Executor对象即undoExecutor，执行回滚日志就是由undoExecutor去执行的。

undoExecutor源码分析

public class UndoExecutorFactory {    public static AbstractUndoExecutor getUndoExecutor(         String dbType, SQLUndoLog sqlUndoLog) {        if (!dbType.equals(JdbcConstants.MYSQL)) {            throw new NotSupportYetException(dbType);        }        switch (sqlUndoLog.getSqlType()) {            case INSERT:                return new MySQLUndoInsertExecutor(sqlUndoLog);            case UPDATE:                return new MySQLUndoUpdateExecutor(sqlUndoLog);            case DELETE:                return new MySQLUndoDeleteExecutor(sqlUndoLog);            default:                throw new ShouldNeverHappenException();        }    }}

说明：

• UndoExecutorFactory负责根据不同的回滚日志返回对应的undoExecutor对象。

说明：

• AbstractUndoExecutor作为回滚类的抽象基类。
• MySQLUndoDeleteExecutor负责回滚delete操作。
• MySQLUndoInsertExecutor负责回滚insert操作。
• MySQLUndoUpdateExecutor负责回滚update操作。

AbstractUndoExecutor

public abstract class AbstractUndoExecutor {    protected SQLUndoLog sqlUndoLog;    protected abstract String buildUndoSQL();    public AbstractUndoExecutor(SQLUndoLog sqlUndoLog) {        this.sqlUndoLog = sqlUndoLog;    }    public void executeOn(Connection conn) throws SQLException {        dataValidation(conn);        try {            // 拼接undoSql的模板            String undoSQL = buildUndoSQL();            // 获取PreparedStatement对象            PreparedStatement undoPST = conn.prepareStatement(undoSQL);            // 获取回滚的记录            TableRecords undoRows = getUndoRows();            // 遍历所有待回滚的记录然后一条条的拼接字段            for (Row undoRow : undoRows.getRows()) {                ArrayList
    
      undoValues = new ArrayList<>();                Field pkValue = null;                for (Field field : undoRow.getFields()) {                    if (field.getKeyType() == KeyType.PrimaryKey) {                        pkValue = field;                    } else {                        undoValues.add(field);                    }                }                // 针对每一条回滚记录进行准备                undoPrepare(undoPST, undoValues, pkValue);               // 执行回滚操作                undoPST.executeUpdate();            }        } catch (Exception ex) {            if (ex instanceof SQLException) {                throw (SQLException) ex;            } else {                throw new SQLException(ex);            }        }    }    protected void undoPrepare(PreparedStatement undoPST,               ArrayList
     
       undoValues,               Field pkValue) throws SQLException {        int undoIndex = 0;        for (Field undoValue : undoValues) {            undoIndex++;            undoPST.setObject(undoIndex, undoValue.getValue(), undoValue.getType());        }        // PK is at last one.        // INSERT INTO a (x, y, z, pk) VALUES (?, ?, ?, ?)        // UPDATE a SET x=?, y=?, z=? WHERE pk = ?        // DELETE FROM a WHERE pk = ?        undoIndex++;        undoPST.setObject(undoIndex, pkValue.getValue(), pkValue.getType());    }    protected abstract TableRecords getUndoRows();    protected void dataValidation(Connection conn) throws SQLException {        // Validate if data is dirty.    }}
     
    

说明：

• AbstractUndoExecutor定义了回滚操作的整个命令行模板流程。
• 拼接undoSql的模板，buildUndoSQL()。
• 获取PreparedStatement对象，conn.prepareStatement(undoSQL)。
• 遍历所有待回滚的记录然后一条条的拼接字段。
• 针对每一条回滚记录进行准备，undoPrepare(undoPST, undoValues, pkValue)。
• 执行回滚操作，undoPST.executeUpdate()。
• buildUndoSQL()和getUndoRows()由子类具体实现。

MySQLUndoInsertExecutor

public class MySQLUndoInsertExecutor extends AbstractUndoExecutor {    @Override    protected String buildUndoSQL() {        TableRecords afterImage = sqlUndoLog.getAfterImage();        List
    
      afterImageRows = afterImage.getRows();        if (afterImageRows == null || afterImageRows.size() == 0) {            throw new ShouldNeverHappenException("Invalid UNDO LOG");        }        Row row = afterImageRows.get(0);        StringBuffer mainSQL = new StringBuffer(        "DELETE FROM " + sqlUndoLog.getTableName());        StringBuffer where = new StringBuffer(" WHERE ");        boolean first = true;        for (Field field : row.getFields()) {            if (field.getKeyType() == KeyType.PrimaryKey) {                where.append(field.getName() + " = ? ");            }        }        return mainSQL.append(where).toString();    }    @Override    protected void undoPrepare(PreparedStatement undoPST,            ArrayList
     
       undoValues, Field pkValue)         throws SQLException {        undoPST.setObject(1, pkValue.getValue(), pkValue.getType());    }    public MySQLUndoInsertExecutor(SQLUndoLog sqlUndoLog) {        super(sqlUndoLog);    }    @Override    protected TableRecords getUndoRows() {        return sqlUndoLog.getAfterImage();    }}
     
    

说明：

• Insert的回滚操作在于逆向进行delete操作，MySQLUndoInsertExecutor负责拼接delete的SQL。
• delete的SQL的where条件就是insert生成的主键primary key。
• 整个回滚操作在父类AbstractUndoExecutor定义。

MySQLUndoDeleteExecutor

public class MySQLUndoDeleteExecutor extends AbstractUndoExecutor {    public MySQLUndoDeleteExecutor(SQLUndoLog sqlUndoLog) {        super(sqlUndoLog);    }    @Override    protected String buildUndoSQL() {        TableRecords beforeImage = sqlUndoLog.getBeforeImage();        List
    
      beforeImageRows = beforeImage.getRows();        if (beforeImageRows == null || beforeImageRows.size() == 0) {            throw new ShouldNeverHappenException("Invalid UNDO LOG");        }        Row row = beforeImageRows.get(0);        StringBuffer insertColumns = new StringBuffer();        StringBuffer insertValues = new StringBuffer();        Field pkField = null;        boolean first = true;        for (Field field : row.getFields()) {            if (field.getKeyType() == KeyType.PrimaryKey) {                pkField = field;                continue;            } else {                if (first) {                    first = false;                } else {                    insertColumns.append(", ");                    insertValues.append(", ");                }                insertColumns.append(field.getName());                insertValues.append("?");            }        }        if (first) {            first = false;        } else {            insertColumns.append(", ");            insertValues.append(", ");        }        insertColumns.append(pkField.getName());        insertValues.append("?");        return "INSERT INTO " + sqlUndoLog.getTableName()         + "(" + insertColumns.toString() + ")         VALUES (" + insertValues.toString() + ")";    }    @Override    protected TableRecords getUndoRows() {        return sqlUndoLog.getBeforeImage();    }}
    

说明：

• Delete的回滚操作在于逆向进行Insert操作，MySQLUndoDeleteExecutor负责拼接Insert的SQL。
• Insert的拼接的SQL是insert tableName (column1,column2) values (?,?).
• 整个回滚操作在父类AbstractUndoExecutor定义。

MySQLUndoUpdateExecutor

public class MySQLUndoUpdateExecutor extends AbstractUndoExecutor {    @Override    protected String buildUndoSQL() {        TableRecords beforeImage = sqlUndoLog.getBeforeImage();        List
    
      beforeImageRows = beforeImage.getRows();        if (beforeImageRows == null || beforeImageRows.size() == 0) {            throw new ShouldNeverHappenException("Invalid UNDO LOG"); // TODO        }        Row row = beforeImageRows.get(0);        StringBuffer mainSQL = new StringBuffer(          "UPDATE " + sqlUndoLog.getTableName() + " SET ");        StringBuffer where = new StringBuffer(" WHERE ");        boolean first = true;        for (Field field : row.getFields()) {            if (field.getKeyType() == KeyType.PrimaryKey) {                where.append(field.getName() + " = ?");            } else {                if (first) {                    first = false;                } else {                    mainSQL.append(", ");                }                mainSQL.append(field.getName() + " = ?");            }        }        return mainSQL.append(where).toString();    }    public MySQLUndoUpdateExecutor(SQLUndoLog sqlUndoLog) {        super(sqlUndoLog);    }    @Override    protected TableRecords getUndoRows() {        return sqlUndoLog.getBeforeImage();    }}
    

说明：

• update的回滚操作在于逆向进行update操作，MySQLUndoUpdateExecutor负责拼接update的SQL。
• Insert的拼接的SQL是update tableName set column1=? where column=?。
• 整个回滚操作在父类AbstractUndoExecutor定义。

Fescar源码分析连载