本文梳理了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
所有操作都委托给了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
已经生成,但是它何时发挥作用,怎么发挥作用,这写问题后续结合全局事务流程再来细看。
最后附上官方的一张图,这时候看就清楚多了。
网友评论