Seata 源码篇之AT模式启动流程 - 下 - 04

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 timeoutException 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 committhrow new TransactionalExecutor.ExecutionException(tx, txe,TransactionalExecutor.Code.CommitFailure);}}

DefaultGlobalTransaction 的 commit 方法负责完成全局事务的提交，当然全局事务提交由TM执行，如果当前分支事务角色是RM，这里直接返回，啥也不干:

    @Overridepublic 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方法实现:

    @Overridepublic 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方法中:

    @Overridepublic 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 AbstractExceptionHandlerimplements RMInboundHandler, TransactionMessageHandler {@Overridepublic BranchCommitResponse handle(BranchCommitRequest request) {BranchCommitResponse response = new BranchCommitResponse();exceptionHandleTemplate(new AbstractCallback<BranchCommitRequest, BranchCommitResponse>() {@Overridepublic 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方法完成分支事务的提交:

    @Overridepublic 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 commitcommitTransaction(tx, txInfo);}}

    private void rollbackTransaction(GlobalTransaction tx, Throwable originalException) throws TransactionException, TransactionalExecutor.ExecutionException {try {// 执行全局事务回滚逻辑和前后回调埋点triggerBeforeRollback();tx.rollback();triggerAfterRollback();}...}

DefaultGlobalTransaction 的 rollback 方法负责完成全局事务的回滚，当然全局事务回滚也由TM执行，如果当前分支事务角色是RM，这里直接返回，啥也不干:

   @Overridepublic 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方法实现:

    @Overridepublic GlobalStatus rollback(String xid) throws TransactionException {GlobalRollbackRequest globalRollback = new GlobalRollbackRequest();globalRollback.setXid(xid);GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback);return response.getGlobalStatus();}

分支事务全局回滚

各个分支事务的本地由TC异步回调通知完成，如下图所示:

在这里插入图片描述

具体代码存在于DefaultRMHandler的handle方法中:

    @Overridepublic 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 AbstractExceptionHandlerimplements RMInboundHandler, TransactionMessageHandler {@Overridepublic BranchRollbackResponse handle(BranchRollbackRequest request) {BranchRollbackResponse response = new BranchRollbackResponse();exceptionHandleTemplate(new AbstractCallback<BranchRollbackRequest, BranchRollbackResponse>() {@Overridepublic 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方法完成分支事务的回滚:

    @Overridepublic 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模块，感兴趣的童鞋可以持续关注。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/125590.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

前端JavaScript入门到精通，javascript核心进阶ES6语法、API、js高级等基础知识和实战 —— Web APIs（四）

前端JavaScript入门到精通，javascript核心进阶ES6语法、API、js高级等基础知识和实战 —— Web APIs（四）

思维导图 一、日期对象 1.1 实例化 实例化&#xff0c;默认得到当前时间&#xff0c;也可以指定时间 1.2 日期对象方法 <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8"><meta http-equiv"X-UA-Compatible&q…
阅读更多...
GPX可视化工具 GPX航迹预览工具

GPX可视化工具 GPX航迹预览工具

背景 当我们收到别人分享的航迹文档&#xff0c;即gpx文档时&#xff0c;如何快速的进行浏览呢&#xff1f;我们可以使用GIS软件来打开gpx文档并显示gpx中所记录的航迹&#xff0c;例如常用的GIS软件有googleEarth&#xff0c; Basecamp&#xff0c; GPXsee&#xff0c; GPX E…
阅读更多...
ffmpeg之去除视频水印

ffmpeg之去除视频水印

ffmpeg去除水印使用delogo视频滤镜。 delogo参数: x,y,w,h分别表示logo区域的左上角位置及宽度和高度&#xff1b; show:0表示不显示logo区域&#xff0c;1表示显示logo区域。 执行下面的命令&#xff1a; ffmpeg -i 1.mp4 -vf delogox300:y10:w80:h30:show0 out.mp4 效果…
阅读更多...
华为云云耀云服务器L实例评测｜基于canal缓存自动更新流程 SpringBoot项目应用案例和源码

华为云云耀云服务器L实例评测｜基于canal缓存自动更新流程 SpringBoot项目应用案例和源码

前言 最近华为云云耀云服务器L实例上新&#xff0c;也搞了一台来玩&#xff0c;期间遇到各种问题&#xff0c;在解决问题的过程中学到不少和运维相关的知识。 在之前的博客中&#xff0c;介绍过canal的安装和配置&#xff0c;参考博客 拉取创建canal镜像配置相关参数 & …
阅读更多...
克服网络安全压力：如何掌控无限的云数据

克服网络安全压力：如何掌控无限的云数据

管理云中的数字风险比以往任何时候都更加重要。数字化转型引发的云数据呈指数级增长&#xff0c;为安全分析师创造了一个更大的威胁环境。随着威胁行为者继续危害组织最敏感的数据&#xff0c;这一挑战将会加剧。 预计未来五年全球网络犯罪成本将激增&#xff0c;从 2022 年的…
阅读更多...
3D 毛玻璃晶质见证卡

3D 毛玻璃晶质见证卡

效果展示 页面结构 从上面的效果展示来看&#xff0c;页面主要成员是一张卡片&#xff0c;并且卡片上有三个小矩形&#xff0c;而小矩形上会展示对应的内容。 当鼠标悬停在卡片上时&#xff0c;卡片会随着鼠标的移动而改变视角。 CSS3 知识点 transform-style 属性的 prese…
阅读更多...
linux入门---信号的理解

linux入门---信号的理解

目录标题 如何理解计算机中的信号如何查看计算机中的信号初步了解信号的保存和发送如何向目标进程发送信号情景一&#xff1a;使用键盘发送信号情景二&#xff1a;系统调用发送信号情景三&#xff1a;硬件异常产生信号情景四&#xff1a;软件条件产生信号 核心转储信号的两个问…
阅读更多...
Elasticsearch基础篇(二)：Elasticsearch在windows和liunx上的安装部署

Elasticsearch基础篇(二)：Elasticsearch在windows和liunx上的安装部署

Elasticsearch简介 前言1. Windows环境部署Elasticsearch1.1 下载并解压Elasticsearch压缩包1.2 命令行启动elasticsearch1.3 验证是否成功启动elasticsearch1.4 关闭Elasticsearch1.5 在Windows上安装Elasticsearch作为服务 2. Liunx环境部署Elasticsearch安装 Elasticsearch …
阅读更多...
逆变器下垂控制单机

逆变器下垂控制单机

仿真控制参数 主电路 坐标变换 功率计算 下垂控制 电压电流双闭环控制 结果变量监控 断路器闭合&#xff0c;负载突增 负载突增&#xff0c;有功和无功突增 有功增加&#xff0c;频率减小 无功增加&#xff0c;参考电压减小
阅读更多...
基于JavaWeb技术的在线考试系统设计与实现

基于JavaWeb技术的在线考试系统设计与实现

目录 前言 一、技术栈 二、系统功能介绍 用户信息管理 考试统计管理 专业列表管理 忘记密码人员登记管理 修改密码 试卷信息 考试信息管理 三、核心代码 1、登录模块 2、文件上传模块 3、代码封装 前言 随着信息技术在管理上越来越深入而广泛的应用&#xff0c;管理…
阅读更多...
Linux Vi编辑器基础操作指南

Linux Vi编辑器基础操作指南

Linux Vi编辑器基础操作指南 Linux中的Vi是一个强大的文本编辑器&#xff0c;虽然它有一些陡峭的学习曲线&#xff0c;但一旦掌握了基本操作&#xff0c;它就变得非常高效。以下是Vi编辑器的一些基本用法&#xff1a; 打开Vi编辑器&#xff1a; vi 文件名退出Vi编辑器&#xff…
阅读更多...
pygame - 贪吃蛇小游戏

pygame - 贪吃蛇小游戏

蛇每吃掉一个身体块&#xff0c;蛇身就增加一个长度。为了统一计算&#xff0c;界面的尺寸和游戏元素的位置都是身体块长度的倍数 1. 上下左右方向键&#xff08;或者ASDW键&#xff09;控制蛇的移动方向 2. 空格键暂停和继续蛇的身体图片文件&#xff0c;复制到项目的asset\im…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023