标签:初始 prope 建立 角色 等于 file 服务端 dog factor
维护全局和分支事务的状态,驱动全局事务提交或回滚,即Seata服务端。
定义全局事务的范围:开始全局事务、提交或回滚全局事务,在事务发起的客户端。
管理分支事务处理的资源,与TC交谈以注册分支事务和报告分支事务的状态,并驱动分支事务提交或回滚,在分支事务执行的客户端。
通过代理数据源DataSourceProxy对业务SQL进行解析,转换成undolog,并与业务SQL在一个事务内入库,然后注册分支事务、提交、上报状态。
分布式事务操作成功,则TC通知RM异步删除undolog。
分布式事务操作失败,TM向TC发送回滚请求,RM 收到协调器TC发来的回滚请求,通过 XID 和 Branch ID 找到相应的回滚日志记录,通过回滚记录生成反向的更新 SQL 并执行,以完成分支的回滚。
以一个示例来说明整个 AT 分支的工作过程。
业务表:product
Field |
Type |
Key |
id |
bigint(20) |
PRI |
name |
varchar(100) |
|
since |
varchar(100) |
AT 分支事务的业务逻辑:
update product set name = ‘GTS‘ where name = ‘TXC‘;
过程:
select id, name, since from product where name = ‘TXC‘;
得到前镜像:
id |
name |
since |
1 |
TXC |
2014 |
select id, name, since from product where id = 1;
得到后镜像:
id |
name |
since |
1 |
GTS |
2014 |
UNDO_LOG
表中。{ "branchId": 641789253, "undoItems": [{ "afterImage": { "rows": [{ "fields": [{ "name": "id", "type": 4, "value": 1 }, { "name": "name", "type": 12, "value": "GTS" }, { "name": "since", "type": 12, "value": "2014" }] }], "tableName": "product" }, "beforeImage": { "rows": [{ "fields": [{ "name": "id", "type": 4, "value": 1 }, { "name": "name", "type": 12, "value": "TXC" }, { "name": "since", "type": 12, "value": "2014" }] }], "tableName": "product" }, "sqlType": "UPDATE" }], "xid": "xid:xxx" }
product
表中,主键值等于 1 的记录的 全局锁 。update product set name = ‘TXC‘ where id = 1;
我们只需要使用一个 @GlobalTransactional
注解在业务方法上:
@GlobalTransactional public void purchase(String userId, String commodityCode, int orderCount) { ...... }
Spring自动配置类中配置了全局事务扫描器GlobalTransactionScanner。
GlobalTransactionScanner内容如下。
可以看到分别实现了Spring的3个接口InitializingBean,ApplicationContextAware,DisposableBean。
在afterPropertiesSet()中,调用了initClient方法:
initClient方法里面对TmClient,RmClient进行了初始化(参数就是配置文件bean里配置的applicationId和txServiceGroup),并注册了一个Spring的ShutdownHook。
TmClient的初始化方法。
其最终调用到的是AbstractNettyRemotingClient的init()方法,启动了一个定时器不断进行重连操作。
NettyClientChannelManager的reconnect方法内容如下:
方法getAvailServerList内从注册中心获取服务器列表。
RegistryFactory.getInstance().lookup(transactionServiceGroup)是针对不同注册中心做了适配的,默认看下File形式的实现。
进到FileRegistryServiceImpl#lookup方法,这里结合File.conf配置来说明。
FileRegistryServiceImpl的服务器查找方法如下:
1、现根据事务分组(key=vgroup_mapping.事务分组名称)找到分组所属的server集群名称,这里是default
2、然后根据集群名称(key=集群名称.grouplist)找到server对应ip端口地址
梳理下TmClient的初始化流程:
RmClient的初始化方法。
1、设置了资源管理器resourceManager
2、设置了消息回调监听器,rmHandler用于接收seata-server在二阶段发出的提交或者回滚请求
ResourceManager管理资源的注册和注销。
RMHandlerAT在收到TC二阶段回滚消息时执行回滚。
在需要加全局事务的方法中,会加上GlobalTransactional注解,注解往往对应着拦截器,Seata中拦截全局事务的拦截器是GlobalTransactionalInterceptor,看下其拦截方法。
判断:
l 如果方法上有全局事务注解,调用handleGlobalTransaction开启全局事务
l 如果没有,按普通方法执行,避免性能下降
看下handleGlobalTransaction()方法:
可以看到最终调用的是TransactionalTemplate的execute方法,execute方法如下:
分为几步:
l 开启全局事务beginTransaction
l 执行业务方法
l 提交事务commitTransaction(若没抛异常)
l 执行completeTransactionAfterThrowing回滚操作(抛异常)
beginTransaction最终调用到了io.seata.tm.api.DefaultGlobalTransaction#begin(int, java.lang.String)方法,代码如下:
看到这里,也就明确了一点,全局事务开启时,是由TM来发起的。
commitTransaction和rollbackTransaction方法类似,由TM发送事务commit或rollback信息给seata-server。
由于Seata对数据源做了代理,所以sql解析与undolog入库操作是在数据源代理中执行的。数据源代理类创建器的配置。
DataSourceProxy,ConnectionProxy,StatementProxy是Seata提供的代理封装类。
最终对Sql进行解析操作,发生在StatementProxy类中:
然后交给了ExecuteTemplate执行,跟到ExecuteTemplate中查看:
关键点在于特定类型执行器中的execute方法,挑选InsertExecutor为例说明,其execute方法调用的是父类BaseTransactionalExecutor中的execute方法,看下源码。
将ROOT上下文中的xid绑定到了connectionProxy中,并调用了doExecute方法,看下AbstractDMLBaseExecutor中的doExecute方法。
查看代码,生成undolog在executeAutoCommitFalse方法中:
executeAutoCommitTrue中先将autoCommit设置为false(因为要对sql进行解析,生成undolog在一个事务中入库,避免提前入库)。
再执行到executeAutoCommitFalse中,分为4步:
到此为止,红色框中几步已经完成。
业务sql执行以及undolog执行完后会在ConnectionProxy中执行commit操作,
看下代码。
1、如果处于全局事务中,则调用processGlobalTransactionCommit()处理全局事务提交
2、如果加了全局锁注解,加全局锁并提交
3、如果没有对应注释,按直接进行事务提交
主要看processGlobalTransactionCommit()方法,也是核心代码:
流程分为如下几步:
undolog入库和普通业务sql的执行用的一个connection,处于一个本地事务中,保证了业务数据变更时,一定会有对应undolog存在。
至此,第一阶段中undolog提交与本地事务提交,分支事务注册与汇报也已完成。
在前面分析RmClient.init()方法时,提到了Seata会使用SPI拓展机制找到RmClient的回调处理器RMHandlerAT,该类是负责接送二阶段seata-server发给RmClient的提交、回滚消息,并作出提交,回滚操作。
RMHandlerAT继承自AbstractRMHandler,AbstractRMHandler中两个handle方法对应,事务提交、回滚操作。
全局事务提交对应了doBranchCommit(request, response)方法。
调用的是getResourceManager(),上面提到SPI拓展提到的DataSourceManager类。
DataSourceManager中调用了asyncWorker来异步提交,看下AsyncWorker中branchCommit方法。
这边只是往一个ASYNC_COMMIT_BUFFER缓冲List中新增了一个二阶段提交的context。
但真正提交在哪呢?答案在AsyncWorker的init()方法中,其init()方法会在DataSourceManager中被调用,内部启动一个定时器不断进行全局事务提交操作。
真正的分支事务提交就是在doBranchCommits中完成的,主要工作是删除回滚日志。
主要分为几步:
回过头来看下设计原理图:
接下来我们看看全局事务回滚的方法,AbstractRMHandler#doBranchRollback 。
该方法调用了DataSourceManager的branchRollback方法。
最终回滚方法调用的是UndoLogManager.undo(dataSourceProxy, xid, branchId),大体是根据Undolog进行反解析并执行回滚操作。
然后进行回滚日志的清理和提交。
具体的Undolog反解析操作实现在AbstractUndoExecutor的子类中。
再回头看下回滚设计原理图:
TCC执行流程如下图所示:
大致流程如下:
1.全局事务拦截器拦截到@GlobalTransational注解,调用TM开启全局事务
2.执行TCC参与者的prepare方法时,被TCC拦截器拦截,在prepare方法执
行前注册分支事务到TC,在prepare方法执行后向TC报告分支事务的状态
3.如果执行发生异常则TM通知TC回滚事务,否则TM通知TC执行提交事务
4.TC收到TM的提交或回滚通知,遍历各TCC分支事务,逐个进行提交或回滚
跟AT模式一样,TCC模式也通过@GlobalTransactional注解开启全局事务,然后调用各个两阶段参与者的prepare方法即可。
两阶段的参与者格式如下:
l TwoPhaseBusinessAction注解标记这是个TCC接口,同时指定commitMethod,rollbackMethod的名称
l BusinessActionContext是TCC事务中的上下文对象
l BusinessActionContextParameter注解标记的参数会在上下文中传播,即能通过BusinessActionContext对象在commit方法及cancle方法中取到该参数值
GlobalTransactionScanner继承了AbstractAutoProxyCreator抽象类,并重新实现了wrapIfNecessary接口,该接口用来在spring启动时,生成代理类。
看下重写的wrapIfNecessary方法。
可以看到这段逻辑中,判断了bean如果是个TCC的接口实现,则将拦截器初始化为TccActionInterceptor,TccActionInterceptor是TCC方法的核心拦截器,后面会具体介绍,先跟到TCCBeanParserUtils.isTccAutoProxy()中看下源码。
isTccAutoProxy()中又会调用DefaultRemotingParser#parserRemotingServiceInfo来进行TCC资源注册。
可以看到,通过反射拿到了TwoPhaseBusinessAction注解中声明的Commit方法和Rollback方法并封装成TCCResource对象,最终调用ResourceManager的registerResource方法。
TCC模式下ResourceManger的实现为TCCResourceManager,AbstractRMHandler的实现为RMHandlerTCC。
跟到TCCResourceManager中查看registerResource方法。
看到将TCCResource对象存储在本地Map中,方便后续通过ResourceId找到对应Resource来进行提交,回滚操作。super.registerResource代码如下,通过RmNettyRemotingClient发送rpc请求给Seata-server进行资源注册。
至此,本地内存中会有个TCCResourceCache,注册完成后,seata-server端也会有个TCC的资源列表。
服务端接收RM注册信息的接口在DefaultServerMessageListenerImpl 的onRegRmMessage中,看下代码。
最终调用了ChannelManager.registerRMChannel方法。
服务端也会对RpcContext进行缓存,缓存Map嵌套层次较多,最外层key为resourceId,往内一次是applicationId,clientIIp,port。
至此,TCC资源管理器RM已完成注册,本地及服务端均有以resourceId为key的缓存Map。
TCC模式业务调用方和AT模式一样,需要使用GlobalTransactional注解来开启全局事务。
业务方法执行时,最终会被AT模式源码分析中提到过的拦截器GlobalTransactionalInterceptor拦截,开启一个全局事务,获得全局事务id,即xid。
具体代码是TransactionalTemplate的execute方法,execute方法如下:
分为一下几步:
1.开启全局事务beginTransaction(TM与TC通信并获得Xid)
服务端接收全局事务开启请求的方法在DefaultCore的begin方法中,可以看到创建了一个GlabalSession。
2.执行业务方法
3.提交事务commitTransaction(TM与TC通信,发起事务提交请求)
4.如果发生异常,执行completeTransactionAfterThrowing回滚操作(TM与TC通信,发起事务回滚请求)
TCC注册过程分析时,如果bean是个TCC的bean(即bean中方法包含TwoPhaseBusinessAction注解),会初始出TccActionInterceptor拦截器,其实现了MethodInterceptor,这也是TCC接口的方法级别核心拦截器。
看下源码中的invoke方法:
方法调用了actionInterceptorHandler.proceed方法:
接着看doTccActionLogStore方法:
服务端接收分支注册的代码也在DefaultCore(见Seata源码)中,代码如下:
至此,TCC分支事务注册完毕。
TransactionalTemplate的execute方法中,若业务执行无异常,则会调用commitTransaction方法。
最终调用的DefaultGlobalTransaction的commit方法。
其中调用TM的commit方法,来通知TC对全局事务进行提交。
TC收到commit消息的处理在DefaultCore(见Seata源码)的commit方法中,查看代码:
分为几步:
看下同步提交代码doGlobalCommit:
遍历每个branchSession,对每个分支事务进行提交,失败会无限重试。
resourceManagerInbound.branchCommit方法会调用DefaultCoordinator中branchCommit方法来与TCC资源管理器通信,发送分支事务提交消息,这里sendSyncRequest方法中就会根据resourceId去找到第一步(TCC资源管理器注册)中RpcContext的缓存,并得到对应Channel来建立Netty通信。
各TCC资源管理器接收到分支事务提交请求后,会调用TCCResourceManager的branchCommit方法实际对事务进行提交。
自此客户端收到seata-server提交信息后,完成了对分支事务的提交。
总结一下全局事务提交的大致流程:
全局事务回滚思路与全局事务提交过程基本一致。
全局事务回滚的大致流程:
1. 业务方调用微服务发生异常,通过TM发起事务回滚请求
2. TC接收到事务回滚请求后,通过Xid找到全局事务,再取出所有分支事务
3. 遍历分支事务,发出分支事务回滚请求
4. TCC资源管理器RM接收到回滚请求后,从本地TCCResource缓存中根据resourceId取出对应方法bean,反射调用rollback方法
到此,我们完成了对Seata框架AT模式和TCC模式完整执行流程的分析。
作者:朝雨忆轻尘
出处:https://www.cnblogs.com/xifengxiaoma/
版权所有,欢迎转载,转载请注明原文作者及出处。
标签:初始 prope 建立 角色 等于 file 服务端 dog factor
原文地址:https://www.cnblogs.com/xifengxiaoma/p/13995416.html