Seata 源码篇之AT模式启动流程 - 下 - 04
- 全局事务提交
- 分支事务全局提交
- 全局事务回滚
- 分支事务全局回滚
- 小结
本系列文章:
- Seata 源码篇之核心思想 - 01
- Seata 源码篇之AT模式启动流程 - 上 - 02
- Seata 源码篇之AT模式启动流程 - 中 - 03
上一篇文章，我们看了Seata AT模式一阶段提交流程，本文我们来看看AT模式的二阶段流程和全局事务提交回滚逻辑的实现。
全局事务提交
当某个分支事务执行完本地业务SQL语句后，下一步就进入全局事务提交环节了，此处我们可以回顾TransactionalTemplate模版类的execute方法，如下所示:
public Object execute(TransactionalExecutor business) throws Throwable {
// 1. 获取当前全局事务的上下文信息
TransactionInfo txInfo = business.getTransactionInfo();
...
// 1.1 判断当前是否已经存在一个全局事务
GlobalTransaction tx = GlobalTransactionContext.getCurrent();
// 1.2 根据不同的全局事务传播行为进行处理
Propagation propagation = txInfo.getPropagation();
SuspendedResourcesHolder suspendedResourcesHolder = null;
try {
switch (propagation) {
...
}
// 1.3 将当前全局锁配置设置到本地线程缓存中,然后返回先前的配置
GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);
try {
// 2. 如果当前线程是全局事务的发起者,即TM,则给TC发送一个开启全局事务的请求,否则只是简单回调相关钩子方法
beginTransaction(txInfo, tx);
Object rs;
try {
// 3. 执行当前分支事务对应的本地事务
rs = business.execute();
} catch (Throwable ex) {
// 4. 分支事务执行发生异常,判断对应异常是否需要回滚,如果需要则回滚当前全局事务
completeTransactionAfterThrowing(txInfo, tx, ex);
throw ex;
}
// 5. 当前分支事务执行正常,由TM发送提交全局事务的请求
commitTransaction(tx, txInfo);
return rs;
} finally {
// 6. 资源清理和恢复,同时触发钩子回调
resumeGlobalLockConfig(previousConfig);
triggerAfterCompletion();
cleanUp();
}
} finally {
// 7. 如果存在被挂起的全局事务,则进行恢复
if (suspendedResourcesHolder != null) {
tx.resume(suspendedResourcesHolder);
}
}
}
本节我们来看一下全局事务提交的commitTransaction方法实现:
private void commitTransaction(GlobalTransaction tx, TransactionInfo txInfo)
throws TransactionalExecutor.ExecutionException, TransactionException {
// 1. 判断全局事务是否超时
if (isTimeout(tx.getCreateTime(), txInfo)) {
// business execution timeout
Exception exx = new TmTransactionException(TransactionExceptionCode.TransactionTimeout,
String.format("client detected transaction timeout before commit, so change to rollback, xid = %s", tx.getXid()));
rollbackTransaction(tx, exx);
return;
}
try {
// 2. 触发回调埋点,同时指向事务提交动作
triggerBeforeCommit();
tx.commit();
// 3. 记录事务执行提交动作后的状态
GlobalStatus afterCommitStatus = tx.getLocalStatus();
TransactionalExecutor.Code code = TransactionalExecutor.Code.Unknown;
switch (afterCommitStatus) {
case TimeoutRollbacking:
code = TransactionalExecutor.Code.Rollbacking;
break;
case TimeoutRollbacked:
code = TransactionalExecutor.Code.RollbackDone;
break;
case Finished:
code = TransactionalExecutor.Code.CommitFailure;
break;
default:
}
// 4. 如果事务提交失败或者超时,则抛出对应的异常信息
Exception statusException = null;
if (GlobalStatus.isTwoPhaseHeuristic(afterCommitStatus)) {
statusException = new TmTransactionException(TransactionExceptionCode.CommitHeuristic,
String.format("Global transaction[%s] not found, may be rollbacked.", tx.getXid()));
} else if (GlobalStatus.isOnePhaseTimeout(afterCommitStatus)) {
statusException = new TmTransactionException(TransactionExceptionCode.TransactionTimeout,
String.format("Global transaction[%s] is timeout and will be rollback[TC].", tx.getXid()));
}
if (null != statusException) {
throw new TransactionalExecutor.ExecutionException(tx, statusException, code);
}
// 5. 正常提交后,触发对应的回调埋点
triggerAfterCommit();
} catch (TransactionException txe) {
// 4.1 Failed to commit
throw new TransactionalExecutor.ExecutionException(tx, txe,
TransactionalExecutor.Code.CommitFailure);
}
}
DefaultGlobalTransaction 的 commit 方法负责完成全局事务的提交，当然全局事务提交由TM执行，如果当前分支事务角色是RM，这里直接返回，啥也不干:
@Override
public void commit() throws TransactionException {
// 1. 如果是RM角色,直接返回
if (role == GlobalTransactionRole.Participant) {
return;
}
// 2. 如果是TM角色,则尝试执行全局事务提交,如果提交失败了,则进行多轮尝试
int retry = COMMIT_RETRY_COUNT <= 0 ? DEFAULT_TM_COMMIT_RETRY_COUNT : COMMIT_RETRY_COUNT;
try {
while (retry > 0) {
try {
retry--;
status = transactionManager.commit(xid);
break;
} catch (Throwable ex) {
if (retry == 0) {
throw new TransactionException("Failed to report global commit", ex);
}
}
}
} finally {
// 3. TM全局事务提交成功后,执行XID解绑
if (xid.equals(RootContext.getXID())) {
suspend(true);
}
}
...
}
DefaultTransactionManager 在Seata中的职责主要作为防腐层存在，负责屏蔽与TC的通信过程，下面我们看看其commit方法实现:
@Override
public GlobalStatus commit(String xid) throws TransactionException {
GlobalCommitRequest globalCommit = new GlobalCommitRequest();
globalCommit.setXid(xid);
GlobalCommitResponse response = (GlobalCommitResponse) syncCall(globalCommit);
return response.getGlobalStatus();
}
当TM完成全局事务提交后，下面便是由TC通知其他分支事务执行全局提交了，也就是二阶段提交，二阶段提交的主要任务就是异步删除本地的undo日志。
分支事务全局提交
各个分支事务的全局提交由TC异步回调通知完成，如下图所示:
具体代码存在于DefaultRMHandler的handle方法中:
@Override
public BranchCommitResponse handle(BranchCommitRequest request) {
MDC.put(RootContext.MDC_KEY_XID, request.getXid());
MDC.put(RootContext.MDC_KEY_BRANCH_ID, String.valueOf(request.getBranchId()));
// 利用分支事务类型,获取对应的处理器,然后调用处理器的handle方法,处理分支事务提交请求
return getRMHandler(request.getBranchType()).handle(request);
}
这里获取的处理器类型为RMHandlerAT，所以最终会调用RMHandlerAT的handle方法处理分支事务提交请求：
public abstract class AbstractRMHandler extends AbstractExceptionHandler
implements RMInboundHandler, TransactionMessageHandler {
@Override
public BranchCommitResponse handle(BranchCommitRequest request) {
BranchCommitResponse response = new BranchCommitResponse();
exceptionHandleTemplate(new AbstractCallback<BranchCommitRequest, BranchCommitResponse>() {
@Override
public void execute(BranchCommitRequest request, BranchCommitResponse response)
throws TransactionException {
// 真正执行分支事务提交的方法
doBranchCommit(request, response);
}
}, request, response);
return response;
}
...
}
protected void doBranchCommit(BranchCommitRequest request, BranchCommitResponse response)
throws TransactionException {
String xid = request.getXid();
long branchId = request.getBranchId();
String resourceId = request.getResourceId();
String applicationData = request.getApplicationData();
// 调用资源管理器的branchCommit方法,完成分支事务提交
BranchStatus status = getResourceManager().branchCommit(request.getBranchType(), xid, branchId, resourceId,
applicationData);
response.setXid(xid);
response.setBranchId(branchId);
response.setBranchStatus(status);
}
由于这里我们使用的是AT模式，所以最终会调用DataSourceManager的branchCommit方法完成分支事务的提交:
@Override
public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId,
String applicationData) throws TransactionException {
return asyncWorker.branchCommit(xid, branchId, resourceId);
}
asyncWorker 通过名字可以猜到，此处采用的是异步提交方式，所以下面我们来看看asyncWorker是如何进行异步提交的:
public BranchStatus branchCommit(String xid, long branchId, String resourceId) {
// 准备两阶段提交上下文信息,然后加入分支事务提交队列中
Phase2Context context = new Phase2Context(xid, branchId, resourceId);
addToCommitQueue(context);
return BranchStatus.PhaseTwo_Committed;
}
不难看出，此处采用的是典型的生产者-消费者模式，下面看看具体是谁会从提交队列中取出任务执行:
private void addToCommitQueue(Phase2Context context) {
// 直接将任务加入队列中去,然后返回
if (commitQueue.offer(context)) {
return;
}
// 如果队列满了,则立即让线程池处理一波任务,然后再尝试将当前任务加入队列
// 此处的thenRun方法是异步执行的
CompletableFuture.runAsync(this::doBranchCommitSafely, scheduledExecutor)
.thenRun(() -> addToCommitQueue(context));
}
队列和定时任务线程池初始化过程可以在AsyncWorker类的构造函数中寻见:
public AsyncWorker(DataSourceManager dataSourceManager) {
this.dataSourceManager = dataSourceManager;
// 默认队列大小为10000
commitQueue = new LinkedBlockingQueue<>(ASYNC_COMMIT_BUFFER_LIMIT);
// 启动定时任务线程池,每秒执行一次任务
ThreadFactory threadFactory = new NamedThreadFactory("AsyncWorker", 2, true);
scheduledExecutor = new ScheduledThreadPoolExecutor(2, threadFactory);
scheduledExecutor.scheduleAtFixedRate(this::doBranchCommitSafely, 10, 1000, TimeUnit.MILLISECONDS);
}
下面可以来看看具体执行的是什么样的任务:
void doBranchCommitSafely() {
doBranchCommit();
}
private void doBranchCommit() {
if (commitQueue.isEmpty()) {
return;
}
// 1. 取出队列中所有任务
List<Phase2Context> allContexts = new LinkedList<>();
commitQueue.drainTo(allContexts);
// 2. 按照资源ID对分支事务提交任务进行分组
Map<String, List<Phase2Context>> groupedContexts = groupedByResourceId(allContexts);
// 3. 依次处理每个任务
groupedContexts.forEach(this::dealWithGroupedContexts);
}
private void dealWithGroupedContexts(String resourceId, List<Phase2Context> contexts) {
...
// 1. 通过资源ID获取对应的数据源代理器
DataSourceProxy dataSourceProxy = dataSourceManager.get(resourceId);
// 2. 删除undo日志
Connection conn = null;
conn = dataSourceProxy.getPlainConnection();
...
UndoLogManager undoLogManager = UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType());
List<List<Phase2Context>> splitByLimit = Lists.partition(contexts, UNDOLOG_DELETE_LIMIT_SIZE);
for (List<Phase2Context> partition : splitByLimit) {
deleteUndoLog(conn, undoLogManager, partition);
}
...
}
private void deleteUndoLog(final Connection conn, UndoLogManager undoLogManager, List<Phase2Context> contexts) {
Set<String> xids = new LinkedHashSet<>(contexts.size());
Set<Long> branchIds = new LinkedHashSet<>(contexts.size());
contexts.forEach(context -> {
xids.add(context.xid);
branchIds.add(context.branchId);
});
...
// 删除undo_log表中的undo_log日志
// 这里提交的事务是为了确保批量删除undo_log日志这一过程的原子性
undoLogManager.batchDeleteUndoLog(xids, branchIds, conn);
if (!conn.getAutoCommit()) {
conn.commit();
}
...
}
可以看到分支事务全局提交逻辑很简单，就是借助asyncWorker完成undo日志的异步删除。
全局事务回滚
全局事务回滚逻辑存在于TransactionalTemplate的completeTransactionAfterThrowing方法中:
private void completeTransactionAfterThrowing(TransactionInfo txInfo, GlobalTransaction tx, Throwable originalException)
throws TransactionalExecutor.ExecutionException, TransactionException {
// 如果需要对当前异常执行回滚,则执行全局事务回滚操作,否则还是执行全局事务提交
if (txInfo != null && txInfo.rollbackOn(originalException)) {
rollbackTransaction(tx, originalException);
} else {
// not roll back on this exception, so commit
commitTransaction(tx, txInfo);
}
}
private void rollbackTransaction(GlobalTransaction tx, Throwable originalException) throws TransactionException, TransactionalExecutor.ExecutionException {
try {
// 执行全局事务回滚逻辑和前后回调埋点
triggerBeforeRollback();
tx.rollback();
triggerAfterRollback();
}
...
}
DefaultGlobalTransaction 的 rollback 方法负责完成全局事务的回滚，当然全局事务回滚也由TM执行，如果当前分支事务角色是RM，这里直接返回，啥也不干:
@Override
public void rollback() throws TransactionException {
// 1. 如果当前分支事务的角色是RM,则直接返回
if (role == GlobalTransactionRole.Participant) {
...
return;
}
...
// 2. 如果当前分支事务角色是TM,则通知TC负责发起全局事务回滚请求
int retry = ROLLBACK_RETRY_COUNT <= 0 ? DEFAULT_TM_ROLLBACK_RETRY_COUNT : ROLLBACK_RETRY_COUNT;
try {
while (retry > 0) {
try {
retry--;
status = transactionManager.rollback(xid);
break;
} catch (Throwable ex) {
...
}
}
} finally {
if (xid.equals(RootContext.getXID())) {
suspend(true);
}
}
...
}
DefaultTransactionManager 在Seata中的职责主要作为防腐层存在，负责屏蔽与TC的通信过程，下面我们看看其rollback方法实现:
@Override
public GlobalStatus rollback(String xid) throws TransactionException {
GlobalRollbackRequest globalRollback = new GlobalRollbackRequest();
globalRollback.setXid(xid);
GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback);
return response.getGlobalStatus();
}
分支事务全局回滚
各个分支事务的本地由TC异步回调通知完成，如下图所示:
具体代码存在于DefaultRMHandler的handle方法中:
@Override
public BranchRollbackResponse handle(BranchRollbackRequest request) {
MDC.put(RootContext.MDC_KEY_XID, request.getXid());
MDC.put(RootContext.MDC_KEY_BRANCH_ID, String.valueOf(request.getBranchId()));
// 利用分支事务类型,获取对应的处理器,然后调用处理器的handle方法,处理分支事务提交请求
return getRMHandler(request.getBranchType()).handle(request);
}
这里获取的处理器类型为RMHandlerAT，所以最终会调用RMHandlerAT的handle方法处理分支事务提交请求：
public abstract class AbstractRMHandler extends AbstractExceptionHandler
implements RMInboundHandler, TransactionMessageHandler {
@Override
public BranchRollbackResponse handle(BranchRollbackRequest request) {
BranchRollbackResponse response = new BranchRollbackResponse();
exceptionHandleTemplate(new AbstractCallback<BranchRollbackRequest, BranchRollbackResponse>() {
@Override
public void execute(BranchRollbackRequest request, BranchRollbackResponse response)
throws TransactionException {
// 执行全局回滚
doBranchRollback(request, response);
}
}, request, response);
return response;
}
...
}
protected void doBranchRollback(BranchRollbackRequest request, BranchRollbackResponse response)
throws TransactionException {
String xid = request.getXid();
long branchId = request.getBranchId();
String resourceId = request.getResourceId();
String applicationData = request.getApplicationData();
// 调用资源管理器的branchRollback方法,完成分支事务回滚
BranchStatus status = getResourceManager().branchRollback(request.getBranchType(), xid, branchId, resourceId,
applicationData);
response.setXid(xid);
response.setBranchId(branchId);
response.setBranchStatus(status);
}
由于这里我们使用的是AT模式，所以最终会调用DataSourceManager的branchRollback方法完成分支事务的回滚:
@Override
public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId,
String applicationData) throws TransactionException {
DataSourceProxy dataSourceProxy = get(resourceId);
...
// 利用undo日志完成回滚
UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).undo(dataSourceProxy, xid, branchId);
...
return BranchStatus.PhaseTwo_Rollbacked;
}
关于如何利用undo日志完成回滚，这块内容将在本系列后面进行讲解，本文暂时不做深究。
小结
到目前为止，我们大体浏览了Seata AT模式整体实现流程。后面，我们将深入Server模块进行研究，以及Seata的RPC模块，感兴趣的童鞋可以持续关注。
![[NISACTF 2022]join-us - 报错注入无列名注入](https://img-blog.csdnimg.cn/f250de7824494bfebf962d7d8c67ab8f.png)
