标签：begin oba glob hang 通知 表示 global tar 线程
https://www.cnblogs.com/lay2017/p/12485081.html
上一篇文章中，我们看了看DefaultCoordinator作为分布式事务的协调者，关于全局事务begin的流程。
DefaultCoordinator把begin的核心实现交付给了DefaultCore，DefaultCore将会构造处一个GlobalSession，一份放到内存里面，一份持久化（默认持久化到 ~/sessionStore/root.data)。
总结下来就是begin的时候将会在Server端构造一个GlobalSession。
那么，本文将看看关于全局事务的commit是怎么样一个流程。
和begin一样，DefaultCoordinator把commit全局事务的实现交付给了DefaultCore来做。我们打开DefaultCore的commit方法
@Override public GlobalStatus commit(String xid) throws TransactionException { // 通过XID找到GlobalSession GlobalSession globalSession = SessionHolder.findGlobalSession(xid); if (globalSession == null) { return GlobalStatus.Finished; } // 添加FileTransactionStoreManager globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager()); // 加锁 boolean shouldCommit = globalSession.lockAndExcute(() -> { // 关闭session，防止还有分支事务注册 globalSession.closeAndClean(); if (globalSession.getStatus() == GlobalStatus.Begin) { // 状态变化 globalSession.changeStatus(GlobalStatus.Committing); return true; } return false; }); // 如果原本不是begin的状态，返回原本状态 if (!shouldCommit) { return globalSession.getStatus(); } // 判断是否可以异步提交 if (globalSession.canBeCommittedAsync()) { // 异步提交 asyncCommit(globalSession); return GlobalStatus.Committed; } else { // 同步提交 doGlobalCommit(globalSession, false); } // 返回状态 return globalSession.getStatus(); }
commit方法比较长，但是逻辑并不复杂
1）首先，先根据XID找到对应的GlobalSession。
2）其次，将GlobalSession的状态从begin到committing表示正在执行提交
3）然后，进行提交操作。
findGlobalSession方法将从file或者db中查询GlobalSession，changeStatus的时候先close了当前全局事务，这样就不会有新的分支事务注册进来了。
canBeCommittedAsync将判断是否可以进行异步提交，判断标准是什么呢？我们看看该方法
public boolean canBeCommittedAsync() { // 遍历分支session for (BranchSession branchSession : branchSessions) { // 如果有一个TCC分支返回false if (branchSession.getBranchType() == BranchType.TCC) { return false; } } return true; }
这里判断的是分支事务的session是TCC类型。
也就是说，如果当前GlobalSession代表的是AT自动事务模式。那么只需要将GlobalSession的状态改为AsyncCommitting即可，而通知RM端的事情交给DefaultCoordinator的异步线程来做。如果包含TCC分支，那么直接调用doGlobalCommit同步提交。
我们再看看DefaultCoordinator的异步线程怎么处理AT模式的的。
asyncCommitting.scheduleAtFixedRate(() -> { try { handleAsyncCommitting(); } catch (Exception e) { LOGGER.info("Exception async committing ... ", e); } }, 0, ASYNC_COMMITTING_RETRY_PERIOD, TimeUnit.MILLISECONDS); protected void handleAsyncCommitting() { // 获取所有AsyncCommitting状态的GlobalSession Collection<GlobalSession> asyncCommittingSessions = SessionHolder.getAsyncCommittingSessionManager().allSessions(); if (CollectionUtils.isEmpty(asyncCommittingSessions)) { return; } // 遍历session for (GlobalSession asyncCommittingSession : asyncCommittingSessions) { try { if (GlobalStatus.AsyncCommitting != asyncCommittingSession.getStatus()) { continue; } asyncCommittingSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager()); // 调用core的doGlobalCommit通知客户端 core.doGlobalCommit(asyncCommittingSession, true); } catch (TransactionException ex) { // ... } } }
异步线程默认1秒中执行一次，将从SessionManager中获取所有AsyncCommitting状态的GlobalSession。
然后逐个调用DefaultCore的doGlobalCommit方法通知客户端删除undoLog。
上面看了全局事务的commit，那么rollback呢？我们跟进DefaultCore的rollback代码
@Override public GlobalStatus rollback(String xid) throws TransactionException { // 找到GlobalSession GlobalSession globalSession = SessionHolder.findGlobalSession(xid); if (globalSession == null) { return GlobalStatus.Finished; } globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager()); // 加锁 boolean shouldRollBack = globalSession.lockAndExcute(() -> { // 先关闭 globalSession.close(); // 状态变更 if (globalSession.getStatus() == GlobalStatus.Begin) { globalSession.changeStatus(GlobalStatus.Rollbacking); return true; } return false; }); if (!shouldRollBack) { return globalSession.getStatus(); } // 直接同步处理 doGlobalRollback(globalSession, false); return globalSession.getStatus(); }
rollback比commit逻辑简单多了，也不用同步异步的区分。直接都是同步遍历所有分支事务的session，通知RM进行rollback操作。
全局事务的提交，无非是从file或者db中拿到GlobalSession。不论是同步提交还是异步提交，最终都是拿到所有分支事务的session，然后通知RM去做相应的commit操作。
全局事务的回滚，一样是拿到GlobalSession。然后通知RM做分支事务的rollback操作。
总得来说，Server端主动推送commit或者rollback通知到RM，如果失败那么就进行retry。seata在一致性方面选择了最终一致性。
标签：begin oba glob hang 通知 表示 global tar 线程
原文地址：https://www.cnblogs.com/lay2017/p/12507557.html
