# seata源码解析：seata AT模式是如何实现的？
# AT模式
AT模式是seata主推的一种模式，对业务无侵入，用户只需要关系自己的业务sql即可，用户的业务sql就是全局事务的一阶段，Seata会自动生成事务的二阶段提交和回滚操作。
那么它是如何做到对业务无侵入的？
seata对业务无侵入是通过数据源代理实现的。
一阶段：
在at模式中需要在客户端的数据库中建如下的表，用来保存回滚的数据
CREATE TABLE IF NOT EXISTS `undo_log`
(
`branch_id` BIGINT NOT NULL COMMENT 'branch transaction id',
`xid` VARCHAR(128) NOT NULL COMMENT 'global transaction id',
`context` VARCHAR(128) NOT NULL COMMENT 'undo_log context,such as serialization',
`rollback_info` LONGBLOB NOT NULL COMMENT 'rollback info',
`log_status` INT(11) NOT NULL COMMENT '0:normal status,1:defense status',
`log_created` DATETIME(6) NOT NULL COMMENT 'create datetime',
`log_modified` DATETIME(6) NOT NULL COMMENT 'modify datetime',
UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`)
) ENGINE = InnoDB
AUTO_INCREMENT = 1
DEFAULT CHARSET = utf8 COMMENT ='AT transaction mode undo table';
seata的数据源代理通过对业务sql的解析，在业务sql执行前后生成前置镜像和后置镜像保存在undo_log中，业务sql的执行和undo_log的插入在一个本地事务中，这样就可以保证任何对业务数据的更新都有相应的回滚日志存在。
在本地事务提交之前还会构建lockKey（本次事务提交影响的数据，lockKey的构建规则之前的文章已经说过了哈），向TC对涉及到的数据加锁，当加锁成功后，才会执行本地事务的commit过程，如果加锁失败，RM会根据重试策略进行重试，如果重试也失败，那么回滚本地事务
二阶段提交
当TM向TC发起全局提交请求时，此时分支事务实际上以及提交了，TC立即释放该全局事务的锁，然后异步调用RM清理回滚日志
二阶段回滚
当RM收到TC发来的回滚请求时，根据xid和branchid找到对应的回滚记录，通过回滚记录生成反向的SQL并执行，完成分支的回滚。当分支回滚结束时，通过TC回滚完成，当所有分支都回滚完成时，才会释放全局事务的锁
RM在回滚的时候会有如下的校验
- beforeImmage等于afterImage则不用回滚（数据没有发生变化），否则执行下一步
- 如果当前的记录等于afterImage，则回滚，否则执行下一步
- 如果当前的记录等于beforeImage，则不用会滚了，否则抛出异常（当前记录和beforeImage和afterImage都不相等），说明发生了脏写
为什么会出现回滚失败呢？
- 有人绕过系统直接操作数据库
- 没有正确配置RM，比如RM执行单独的本地事务，此时可以在相应的方法上加上@GlobalLock注解，让RM在本地事务提交前也需要获取全局锁
# AT模式是如何保证隔离性的？
# 写隔离
seata at 模式的读写隔离是通过全局锁来实现的。官网的例子讲的很清楚，我们引用一下
两个全局事务 tx1 和 tx2，分别对 a 表的 m 字段进行更新操作，m 的初始值 1000。
- tx1 先开始，开启本地事务，拿到本地锁，更新操作 m = 1000 - 100 = 900。本地事务提交前，先拿到该记录的全局锁 ，本地提交释放本地锁。
- tx2 后开始，开启本地事务，拿到本地锁，更新操作 m = 900 - 100 = 800。本地事务提交前，尝试拿该记录的全局锁，tx1 全局提交前，该记录的全局锁被 tx1 持有，tx2 需要重试等待全局锁。
- tx1 二阶段全局提交，释放全局锁。tx2 拿到全局锁提交本地事务。
如果 tx1 的二阶段全局回滚，则 tx1 需要重新获取该数据的本地锁，进行反向补偿的更新操作，实现分支的回滚。
此时，如果 tx2 仍在等待该数据的全局锁，同时持有本地锁，则 tx1 的分支回滚会失败。分支的回滚会一直重试，直到 tx2 的全局锁等锁超时，放弃全局锁并回滚本地事务释放本地锁，tx1 的分支回滚最终成功。
因为整个过程全局锁在 tx1 结束前一直是被 tx1 持有的，所以不会发生脏写的问题。
# 读隔离
seata at 模式默认的隔离级别为读未提交（因为已经提交的sql有可能会回滚）。如果要实现读已提交，select语句需要更改为 SELECT FOR UPDATE 语句。
SELECT FOR UPDATE 语句的执行会申请全局锁，如果全局锁被其他事务持有，则释放本地锁（回滚 SELECT FOR UPDATE 语句的本地执行）并重试。这个过程中，查询是被 block 住的，直到 全局锁 拿到，即读取的相关数据是已提交的，才返回
出于总体性能上的考虑，Seata 目前的方案并没有对所有 SELECT 语句都进行代理，仅针对 FOR UPDATE 的 SELECT 语句
# 源码分析
在执行sql的过程中，各个代理对象起到的作用如下
# ExecuteTemplate#execute执行本地sql
分支事务的执行过程是在StatementProxy，PreparedStatementProxy的execute，executeQuery，executeUpdate等方法中的，而这些方法最终都会执行到ExecuteTemplate#execute方法
public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecognizers,
StatementProxy<S> statementProxy,
StatementCallback<T, S> statementCallback,
Object... args) throws SQLException {
// 方法上加了 GlobalLock 注解，是at模式，都得执行代理后的逻辑
// 否则不用执行
if (!RootContext.requireGlobalLock() && BranchType.AT != RootContext.getBranchType()) {
// Just work as original statement
return statementCallback.execute(statementProxy.getTargetStatement(), args);
}
// 获取数据库类型
String dbType = statementProxy.getConnectionProxy().getDbType();
if (CollectionUtils.isEmpty(sqlRecognizers)) {
// 获取到sql语句解析器
sqlRecognizers = SQLVisitorFactory.get(
statementProxy.getTargetSQL(),
dbType);
}
Executor<T> executor;
if (CollectionUtils.isEmpty(sqlRecognizers)) {
// 没有找到合适的sql语句解析器，则执行使用原生的Statement执行sql
executor = new PlainExecutor<>(statementProxy, statementCallback);
} else {
if (sqlRecognizers.size() == 1) {
SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
switch (sqlRecognizer.getSQLType()) {
case INSERT:
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包含多个update语句等
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;
}
根据执行的sql语句的不同选择不同的执行器，然后执行execute方法。只有2个Executor重写了execute方法，即PlainExecutor和BaseTransactionalExecutor。
没有找到对应sql分析器的语句（比如select语句），才会执行PlainExecutor#execute方法，而PlainExecutor#execute方法只是单纯调用原生的Statement来执行sql
其他的语句则会通过BaseTransactionalExecutor#execute方法来执行
// BaseTransactionalExecutor
public T execute(Object... args) throws Throwable {
String xid = RootContext.getXID();
if (xid != null) {
statementProxy.getConnectionProxy().bind(xid);
}
// 方法被 @GlobalLock 修饰，将值设置为true
statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
return doExecute(args);
}
除了select for update语句，其他语句基本上会执行到AbstractDMLBaseExecutor#executeAutoCommitTrue方法
// AbstractDMLBaseExecutor
protected T executeAutoCommitTrue(Object[] args) throws Throwable {
ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
try {
connectionProxy.changeAutoCommit();
// 提交事务时会获取锁，当获锁失败时，按照锁重试策略进行重试
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 {
connectionProxy.getContext().reset();
connectionProxy.setAutoCommit(true);
}
}
protected T executeAutoCommitFalse(Object[] args) throws Exception {
if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && isMultiPk()) {
throw new NotSupportYetException("multi pk only support mysql!");
}
// 查询前置镜像
TableRecords beforeImage = beforeImage();
// 执行sql
T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
// 查询后置镜像
TableRecords afterImage = afterImage(beforeImage);
// 构建undoLog
prepareUndoLog(beforeImage, afterImage);
return result;
}
protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
if (beforeImage.getRows().isEmpty() && afterImage.getRows().isEmpty()) {
return;
}
// 修改主键时抛出异常
if (SQLType.UPDATE == sqlRecognizer.getSQLType()) {
if (beforeImage.getRows().size() != afterImage.getRows().size()) {
throw new ShouldNeverHappenException("Before image size is not equaled to after image size, probably because you updated the primary keys.");
}
}
ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
// 构建lockKeys
String lockKeys = buildLockKey(lockKeyRecords);
if (null != lockKeys) {
connectionProxy.appendLockKey(lockKeys);
SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
connectionProxy.appendUndoLog(sqlUndoLog);
}
}
主要流程就是生成前置镜像，执行业务sql，生成后置镜像，构建lockKey和undoLog，接着就是调用ConnectionProxy来提交事务
构建lockKeys的规则如下，根据执行的sql构建出对应的select语句，找到影响数据的主键值（用seata表中必须有主键）
如果主键是一列，则形式如下
// 表名:主键值1,主键值2
account_info:1,2
如果主键是多列，则形式如下
// 表名:主键值1_主键值a,主键值1_主键值b
account_info:1_a,2_b
# ConnectionProxy#commit提交本地事务
private void doCommit() throws SQLException {
if (context.inGlobalTransaction()) {
// 在全局事务中
processGlobalTransactionCommit();
} else if (context.isGlobalLockRequire()) {
// 处理被 GlobalLock 修饰的方法
processLocalCommitWithGlobalLocks();
} else {
targetConnection.commit();
}
}
本次sql提交在一个全局事务中，则注册分支事务，插入undoLog
private void processGlobalTransactionCommit() throws SQLException {
try {
// 注册分支事务，向tc发一个分支事务注册请求，然后tc在branch_table表里插入一条记录
// 注册的时候tc会看事务提交所需要的资源是否能被加锁
// 如果能，则注册成功
// 如果不能，说明资源正在被别的事务使用，注册失败
register();
} catch (TransactionException e) {
recognizeLockKeyConflictException(e, context.buildLockKeys());
}
try {
// 插入undolog
UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
// 将业务修改 和 undolog 一起插入
targetConnection.commit();
} catch (Throwable ex) {
LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
// 向tc汇报分支状态为失败
report(false);
throw new SQLException(ex);
}
if (IS_REPORT_SUCCESS_ENABLE) {
// 向tc汇报分支状态为成功
report(true);
}
context.reset();
}
本次sql提交在一个本地事务中，只不过方法被加了@GlobalLock，需要尝试获取锁，避免脏写
private void processLocalCommitWithGlobalLocks() throws SQLException {
// 查询是否能获取到锁
checkLock(context.buildLockKeys());
try {
targetConnection.commit();
} catch (Throwable ex) {
throw new SQLException(ex);
}
context.reset();
}
# SelectForUpdateExecutor
前面的文章我们说过用select for update语句来保证隔离级别为读已提交。SelectForUpdateExecutor就是用来执行select for update语句的。
流程比较简单就不放源码了，先构建出lockKey，然后向TC发送请求看这些记录是否已经被其他事务加锁了，如果被加锁了，则根据重试策略不断重试，如果没被加锁，则正常返回查询的结果