seata源码-全局事务提交 服务端源码

前面的博客中，我们介绍了，发起全局事务时，是如何进行全局事务提交的，这篇博客，主要记录，在seata分布式事务中，全局事务提交的时候，服务端是如何进行处理的

发起全局事务提交操作

事务发起者，在所有分支事务执行完毕之后，如果没有发生异常，会进行全局事务提交

这里就不做过多解释了，前面seata入门和全局事务begin的博客中，有介绍过这里入参的request对象的重要性

服务端接收到请求

前面全局事务begin的源码，介绍过，netty服务端接收到请求之后，是如何执行到这里的，在这里会根据request请求的类型，交给不同的handler来处理

io.seata.server.AbstractTCInboundHandler#handle(io.seata.core.protocol.transaction.GlobalCommitRequest, io.seata.core.rpc.RpcContext)

io.seata.server.coordinator.DefaultCoordinator#doGlobalCommit

上面也没有太多的业务逻辑，没什么好说的，我们直接来看core.commit()方法的逻辑

io.seata.server.coordinator.DefaultCore#commit

@Override
public GlobalStatus commit(String xid) throws TransactionException {/*** 1.获取全局session：globalSession*/GlobalSession globalSession = SessionHolder.findGlobalSession(xid);if (globalSession == null) {return GlobalStatus.Finished;}/*** 2.给globalSession添加监听*/globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager());// just lock changeStatus/*** 3.对globalSession进行一些处理*/boolean shouldCommit = SessionHolder.lockAndExecute(globalSession, () -> {// Highlight: Firstly, close the session, then no more branch can be registered./*** 3.1 globalSession.setActive(false);将全局session的active设置为false* 在clean()方法中会把lockTable中本次加锁的记录（分支事务相关锁信息）删除*/globalSession.closeAndClean();if (globalSession.getStatus() == GlobalStatus.Begin) {/*** 对于AT模式，这里永远返回false* 对于AT模式，这里会执行if的逻辑，将globalSession的status设置为AsyncCommitting*/if (globalSession.canBeCommittedAsync()) {globalSession.asyncCommit();return false;} else {/*** 将globalSession的状态设置为committing状态*/globalSession.changeStatus(GlobalStatus.Committing);return true;}}return false;});/*** 4.通知分支事务进行提交，如果是AT模式，不会进入到这里执行，因为shouldCommit是false* 是通过一个异步线程来进行调用doGlobalCommit()方法的*/if (shouldCommit) {boolean success = doGlobalCommit(globalSession, false);//If successful and all remaining branches can be committed asynchronously, do async commit.if (success && globalSession.hasBranch() && globalSession.canBeCommittedAsync()) {globalSession.asyncCommit();return GlobalStatus.Committed;} else {return globalSession.getStatus();}} else {return globalSession.getStatus() == GlobalStatus.AsyncCommitting ? GlobalStatus.Committed : globalSession.getStatus();}
}

上面贴的这段代码，是netty服务端接收到客户端全局事务请求之后，最核心的一个入口，我们拆开来看

commit

closeAndClean()

上面代码中注释1、2、3就不细看了，是一些不重要的逻辑，我们直接来看3.1这个注释对应的方法
在这个方法中，一个close()方法，一个clean()方法

close()方法中的逻辑比较简单，就是把globalSession的active属性设置为了false

我们接着来看clean()方法：

clean()内部调用的这个方法，如果看过前面全局事务回滚的代码，就会发现这个代码很眼熟，就是把分支事务对应的lockTable中指定的加锁的资源进行释放

shouldCommit()

我们接着来看shouldCommit参数的赋值逻辑，可以着重看下第三点的注释，这里的shouldCommit参数，如果是AT模式，永远false，原因是在这里

在canBeCommittedAsync方法中，下面两张截图我们结合起来看，如果是AT模式，canBeCommitedAsync()返回的一定是true；如果这里返回true，那上面截图，一定会进入到if()的逻辑中

所以，对于AT模式，shouldCommit一定是false，并且会调用globalSession.asyncCommit();

这段代码的整体逻辑，就是说，如果当前事务是允许异步处理的，那就给shouldCommit赋值为false，同时把globalSession的status修该为AsyncCommitting；这个状态很重要，这里的意思，我认为是说，当前事务是需要异步处理的，当前代码中，就不同步处理了，接来下的逻辑，可以证明

可以发现，当是AT模式的时候，直接执行了else的逻辑，那我们接下来看下，对于netty服务端，真正去处理分支事务的代码

init()

这段初始化的代码，和上面commit()有点关联，我们截止到这里，需要知道上面commit()方法，如果是AT模式的时候，只是把当前globalSession的状态改成了AsyncCommitting状态

io.seata.server.coordinator.DefaultCoordinator#init

这个方法，是在服务端启动的时候，会在这里初始化一批异步线程，其中有一个和本篇博客有关系

查找所有AsyncCommitting状态的globalSession

io.seata.server.storage.db.session.DataBaseSessionManager#allSessions

这里就不再继续往下贴代码了，逻辑比较简单，可以自己看下，简单来说，就是根据当前入参中的AsyncCommitting，从globalTable中，根据状态进行查询，然后找到所有待异步处理的globalSession

core.doGlobalCommit

io.seata.server.coordinator.DefaultCore#doGlobalCommit

这个方法，是服务端进行全局事务提交的处理逻辑，中间绕了这么一大圈，现在逻辑应该有点清晰明了了，其实就是在netty服务端接收到同步请求的时候，只会先把lockTable中加锁的数据删除，然后修改globalSession的状态，最后通过异步定时执行的线程池去执行全局事务提交的逻辑

public boolean doGlobalCommit(GlobalSession globalSession, boolean retrying) throws TransactionException {boolean success = true;// start committing eventeventBus.post(new GlobalTransactionEvent(globalSession.getTransactionId(), GlobalTransactionEvent.ROLE_TC,globalSession.getTransactionName(), globalSession.getBeginTime(), null, globalSession.getStatus()));if (globalSession.isSaga()) {success = getCore(BranchType.SAGA).doGlobalCommit(globalSession, retrying);} else {for (BranchSession branchSession : globalSession.getSortedBranches()) {// if not retrying, skip the canBeCommittedAsync branchesif (!retrying && branchSession.canBeCommittedAsync()) {continue;}BranchStatus currentStatus = branchSession.getStatus();/*** 如果二阶段分支事务状态是失败，就无需执行下面的逻辑，直接remove即可*/if (currentStatus == BranchStatus.PhaseOne_Failed) {globalSession.removeBranch(branchSession);continue;}try {/*** 这里是服务端在全局事务提交的时候，会通知RM去对本地的branch事务进行处理，是通过netty去交互的* 1.如果RM删除成功，就将branchSession移除，并释放锁* 2.如果删除失败*/BranchStatus branchStatus = getCore(branchSession.getBranchType()).branchCommit(globalSession, branchSession);switch (branchStatus) {case PhaseTwo_Committed:globalSession.removeBranch(branchSession);continue;case PhaseTwo_CommitFailed_Unretryable:if (globalSession.canBeCommittedAsync()) {LOGGER.error("Committing branch transaction[{}], status: PhaseTwo_CommitFailed_Unretryable, please check the business log.", branchSession.getBranchId());continue;} else {SessionHelper.endCommitFailed(globalSession);LOGGER.error("Committing global transaction[{}] finally failed, caused by branch transaction[{}] commit failed.", globalSession.getXid(), branchSession.getBranchId());return false;}default:if (!retrying) {globalSession.queueToRetryCommit();return false;}if (globalSession.canBeCommittedAsync()) {LOGGER.error("Committing branch transaction[{}], status:{} and will retry later",branchSession.getBranchId(), branchStatus);continue;} else {LOGGER.error("Committing global transaction[{}] failed, caused by branch transaction[{}] commit failed, will retry later.", globalSession.getXid(), branchSession.getBranchId());return false;}}} catch (Exception ex) {StackTraceLogger.error(LOGGER, ex, "Committing branch transaction exception: {}",new String[] {branchSession.toString()});if (!retrying) {globalSession.queueToRetryCommit();throw new TransactionException(ex);}}}//If has branch and not all remaining branches can be committed asynchronously,//do print log and return falseif (globalSession.hasBranch() && !globalSession.canBeCommittedAsync()) {LOGGER.info("Committing global transaction is NOT done, xid = {}.", globalSession.getXid());return false;}}//If success and there is no branch, end the global transaction./*** 这里的sessionHelper.encCommitted会将globalSession中的记录删除*/if (success && globalSession.getBranchSessions().isEmpty()) {SessionHelper.endCommitted(globalSession);// committed eventeventBus.post(new GlobalTransactionEvent(globalSession.getTransactionId(), GlobalTransactionEvent.ROLE_TC,globalSession.getTransactionName(), globalSession.getBeginTime(), System.currentTimeMillis(),globalSession.getStatus()));LOGGER.info("Committing global transaction is successfully done, xid = {}.", globalSession.getXid());}return success;
}

上面这段代码，是全局事务提交的逻辑，本质上，和全局事务回滚的逻辑，没太大的区别，只是底层一个调用的是commit，一个调用的是rollback；针对上面这段代码，我们需要着重关注的是：branchCommit这个方法，这个方法，我们下面单独说，先来看下这段代码的逻辑

1. getCore(branchSession.getBranchType()).branchCommit(globalSession, branchSession); 通过这个方法，发起netty请求，通知各个分支事务，进行全局事务的提交
2. 根据分支事务的返回状态，进行不同的处理；如果分支事务处理成功，在seata服务端这里，会把分支事务删除，并且把内存中的分支事务id删除（在seata服务端，每个全局事务都会维护一个集合，存储当前全局事务对应的所有分支事务）
3. 在所有的分支事务，处理完毕之后，SessionHelper.endCommitted(globalSession); 通过这个方法，结束所有的逻辑，其实就是把globalSession从mysql数据库中删除

针对第二点和第三点，就不点进去细看了，其实本质上和全局事务回滚时，做的逻辑几乎上是一致的，我们主要来看下第一点，分支事务是如何处理分支事务提交的逻辑，因为这段逻辑和分支事务回滚的逻辑不一样

分支事务提交

io.seata.rm.datasource.AsyncWorker#branchCommit

这是rm这一端接收到分支事务提交的处理逻辑，但是我们会发现，这段代码很简单：

这里可以看到，只是将当前请求信息，塞到了一个queue中

io.seata.rm.datasource.AsyncWorker#init

这里可以看到，在rm启动的时候，会初始化一个定时执行的线程池，在这个线程池中，会定时的调用doBranchCommit()方法

io.seata.rm.datasource.AsyncWorker#doBranchCommits

private void doBranchCommits() {if (ASYNC_COMMIT_BUFFER.isEmpty()) {return;}/*** 1.mappedContexts存储的是从阻塞队列中获取到的要处理的分支事务*/Map<String, List<Phase2Context>> mappedContexts = new HashMap<>(DEFAULT_RESOURCE_SIZE);List<Phase2Context> contextsGroupedByResourceId;while (!ASYNC_COMMIT_BUFFER.isEmpty()) {Phase2Context commitContext = ASYNC_COMMIT_BUFFER.poll();contextsGroupedByResourceId = CollectionUtils.computeIfAbsent(mappedContexts, commitContext.resourceId, key -> new ArrayList<>());contextsGroupedByResourceId.add(commitContext);}/*** 2.遍历mappedContexts*/for (Map.Entry<String, List<Phase2Context>> entry : mappedContexts.entrySet()) {Connection conn = null;DataSourceProxy dataSourceProxy;try {try {DataSourceManager resourceManager = (DataSourceManager) DefaultResourceManager.get().getResourceManager(BranchType.AT);dataSourceProxy = resourceManager.get(entry.getKey());if (dataSourceProxy == null) {throw new ShouldNeverHappenException("Failed to find resource on " + entry.getKey());}conn = dataSourceProxy.getPlainConnection();} catch (SQLException sqle) {LOGGER.warn("Failed to get connection for async committing on " + entry.getKey(), sqle);continue;}contextsGroupedByResourceId = entry.getValue();Set<String> xids = new LinkedHashSet<>(UNDOLOG_DELETE_LIMIT_SIZE);Set<Long> branchIds = new LinkedHashSet<>(UNDOLOG_DELETE_LIMIT_SIZE);/*** 3.判断要处理的分支事务数量是否达到了批量处理的阈值* 如果到了，就批量进行删除* 否则的话，就清空xids和branchIds 然后return*/for (Phase2Context commitContext : contextsGroupedByResourceId) {xids.add(commitContext.xid);branchIds.add(commitContext.branchId);int maxSize = Math.max(xids.size(), branchIds.size());/*** 并不是在每次全局事务提交的时候，就会执行下面的sql* 而是在达到一定的阈值的时候，才会批量执行，阈值默认是1000** 删除undoLog日志*/if (maxSize == UNDOLOG_DELETE_LIMIT_SIZE) {try {UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).batchDeleteUndoLog(xids, branchIds, conn);} catch (Exception ex) {LOGGER.warn("Failed to batch delete undo log [" + branchIds + "/" + xids + "]", ex);}xids.clear();branchIds.clear();}}if (CollectionUtils.isEmpty(xids) || CollectionUtils.isEmpty(branchIds)) {return;}try {UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).batchDeleteUndoLog(xids,branchIds, conn);} catch (Exception ex) {LOGGER.warn("Failed to batch delete undo log [" + branchIds + "/" + xids + "]", ex);}if (!conn.getAutoCommit()) {conn.commit();}} catch (Throwable e) {LOGGER.error(e.getMessage(), e);try {if (conn != null) {conn.rollback();}} catch (SQLException rollbackEx) {LOGGER.warn("Failed to rollback JDBC resource while deleting undo_log ", rollbackEx);}} finally {if (conn != null) {try {conn.close();} catch (SQLException closeEx) {LOGGER.warn("Failed to close JDBC resource while deleting undo_log ", closeEx);}}}}
}

上面这端代码，我们可以看下，没1000ms执行一次，如果这1000ms之内，queue中待指定分支事务commit的请求达到了一定的阈值(UNDOLOG_DELETE_LIMIT_SIZE)，就会执行commit请求（UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).batchDeleteUndoLog(
xids, branchIds, conn);）；如果没有达到阈值，也会去执行commit的操作

UndoLogManagerFactory.getUndoLogManager 这里执行提交的逻辑，其实就是把undoLog数据给删除，因为对于全局事务提交时，其实在第一阶段，mysql已经真正执行了commit的操作，在第二阶段，只需要把undoLog给删除即可

总结

以上，就是全局事务提交的逻辑，整体看下来，我们可以发现，对于全局事务提交的时候，分支事务在处理的时候，是异步来处理的，这是和回滚逻辑有很大的区别，因为上篇博客中，我们有看到，全局事务回滚时，分支事务在第二阶段，是同步处理的，在接收到请求之后，会根据undoLog生成回滚sql，并执行，然后删除undoLog数据，但是对于全局事务提交的第二阶段，会发现，接收到请求之后，直接塞到了队列中，通过异步的请求，没1000ms执行一次提交的逻辑