用户体验 网站 外国,企业网站空间多大合适,支付网站建设费的会计分录,深圳企业网站建设多少钱前言
在分布式系统中#xff0c;分布式事务是一个必须要解决的问题#xff0c;目前使用较多的是最终一致性方案。自年初阿里开源了Fescar#xff08;四月初更名为Seata#xff09;后#xff0c;该项目受到了极大的关注#xff0c;目前已接近 8000 Star。Seata 以高性能和…前言
在分布式系统中分布式事务是一个必须要解决的问题目前使用较多的是最终一致性方案。自年初阿里开源了Fescar四月初更名为Seata后该项目受到了极大的关注目前已接近 8000 Star。Seata 以高性能和零侵入的特性为目标解决微服务领域的分布式事务难题目前正处于快速迭代中近期小目标是生产可用的 Mysql 版本。
本文主要基于 spring cloud spring jpa spring cloud alibaba fescar mysql seata 的结构搭建一个分布式系统的 demo通过 seata 的 debug 日志和源代码从 client 端RM、TM的角度分析其工作流程及原理。
为了更好地理解全文我们来熟悉一下相关概念
XID全局事务的唯一标识由 ip:port:sequence 组成Transaction Coordinator (TC)事务协调器维护全局事务的运行状态负责协调并驱动全局事务的提交或回滚Transaction Manager (TM )控制全局事务的边界负责开启一个全局事务并最终发起全局提交或全局回滚的决议Resource Manager (RM)控制分支事务负责分支注册、状态汇报并接收事务协调器的指令驱动分支本地事务的提交和回滚提示文中代码是基于 fescar-0.4.1 版本由于项目刚更名为 seata 不久其中一些包名、类名、jar包等名称还没统一更换过来故下文中仍使用 fescar 进行表述。 分布式框架支持
Fescar 使用 XID 表示一个分布式事务XID 需要在一次分布式事务请求所涉的系统中进行传递从而向 feacar-server 发送分支事务的处理情况以及接收 feacar-server 的 commit、rollback 指令。 Fescar 官方已支持全版本的 dubbo 协议而对于 spring cloudspring-boot的分布式项目社区也提供了相应的实现
dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-alibaba-fescar/artifactIdversion2.1.0.BUILD-SNAPSHOT/version
/dependency
该组件实现了基于 RestTemplate、Feign 通信时的 XID 传递功能。
业务逻辑
业务逻辑是经典的下订单、扣余额、减库存流程。 根据模块划分为三个独立的服务且分别连接对应的数据库
订单order-server账户account-server库存storage-server
另外还有发起分布式事务的业务系统
业务business-server
项目结构如下图 正常业务:
business发起购买请求storage扣减库存order创建订单account扣减余额
异常业务
business发起购买请求storage扣减库存order创建订单account扣减余额异常
正常流程下 2、3、4 步的数据正常更新全局 commit异常流程下的数据则由于第 4 步的异常报错全局回滚。
配置文件
fescar 的配置入口文件是 registry.conf, 查看代码 ConfigurationFactory 得知目前还不能指定该配置文件所以配置文件名称只能为 registry.conf。
private static final String REGISTRY_CONF registry.conf;
public static final Configuration FILE_INSTANCE new FileConfiguration(REGISTRY_CONF);
在 registry 中可以指定具体配置的形式默认使用 file 类型在 file.conf 中有 3 部分配置内容
transport transport 部分的配置对应 NettyServerConfig 类用于定义 Netty 相关的参数TM、RM 与 fescar-server 之间使用 Netty 进行通信。 service service {#vgroup-rgroupvgroup_mapping.my_test_tx_group default#配置Client连接TC的地址default.grouplist 127.0.0.1:8091#degrade current not supportenableDegrade false#disable是否启用seata的分布式事务disableGlobalTransaction false
} client client {#RM接收TC的commit通知后缓冲上限async.commit.buffer.limit 10000lock {retry.internal 10retry.times 30}
}
数据源 Proxy
除了前面的配置文件fescar 在 AT 模式下稍微有点代码量的地方就是对数据源的代理指定且目前只能基于DruidDataSource的代理。 注在最新发布的 0.4.2 版本中已支持任意数据源类型
Bean
ConfigurationProperties(prefix spring.datasource)
public DruidDataSource druidDataSource() {DruidDataSource druidDataSource new DruidDataSource();return druidDataSource;
}Primary
Bean(dataSource)
public DataSourceProxy dataSource(DruidDataSource druidDataSource) {return new DataSourceProxy(druidDataSource);
}
使用 DataSourceProxy 的目的是为了引入 ConnectionProxy fescar 无侵入的一方面就体现在 ConnectionProxy 的实现上即分支事务加入全局事务的切入点是在本地事务的 commit 阶段这样设计可以保证业务数据与 undo_log 是在一个本地事务中。
undo_log 是需要在业务库上创建的一个表fescar 依赖该表记录每笔分支事务的状态及二阶段 rollback 的回放数据。不用担心该表的数据量过大形成单点问题在全局事务 commit 的场景下事务对应的 undo_log 会异步删除。
CREATE TABLE undo_log (id bigint(20) NOT NULL AUTO_INCREMENT,branch_id bigint(20) NOT NULL,xid varchar(100) NOT NULL,rollback_info longblob NOT NULL,log_status int(11) NOT NULL,log_created datetime NOT NULL,log_modified datetime NOT NULL,ext varchar(100) DEFAULT NULL,PRIMARY KEY (id),UNIQUE KEY ux_undo_log (xid,branch_id)
) ENGINEInnoDB AUTO_INCREMENT1 DEFAULT CHARSETutf8;
启动 Server
前往https://github.com/seata/seata/releases 下载与 Client 版本对应的 fescar-server,避免由于版本的不同导致的协议不一致问题 进入解压之后的 bin 目录执行
./fescar-server.sh 8091 ../data
启动成功输出
2019-04-09 20:27:24.637 INFO [main]c.a.fescar.core.rpc.netty.AbstractRpcRemotingServer.start:152 -Server started ...
启动 Client
fescar 的加载入口类位于 GlobalTransactionAutoConfiguration对基于 spring boot 的项目能够自动加载当然也可以通过其他方式示例化 GlobalTransactionScanner。
Configuration
EnableConfigurationProperties({FescarProperties.class})
public class GlobalTransactionAutoConfiguration {private final ApplicationContext applicationContext;private final FescarProperties fescarProperties;public GlobalTransactionAutoConfiguration(ApplicationContext applicationContext, FescarProperties fescarProperties) {this.applicationContext applicationContext;this.fescarProperties fescarProperties;}/*** 示例化GlobalTransactionScanner* scanner为client初始化的发起类*/Beanpublic GlobalTransactionScanner globalTransactionScanner() {String applicationName this.applicationContext.getEnvironment().getProperty(spring.application.name);String txServiceGroup this.fescarProperties.getTxServiceGroup();if (StringUtils.isEmpty(txServiceGroup)) {txServiceGroup applicationName -fescar-service-group;this.fescarProperties.setTxServiceGroup(txServiceGroup);}return new GlobalTransactionScanner(applicationName, txServiceGroup);}
}
可以看到支持一个配置项FescarProperties用于配置事务分组名称
spring.cloud.alibaba.fescar.tx-service-groupmy_test_tx_group
如果不指定服务组则默认使用spring.application.name -fescar-service-group生成名称所以不指定spring.application.name启动会报错。
ConfigurationProperties(spring.cloud.alibaba.fescar)
public class FescarProperties {private String txServiceGroup;public FescarProperties() {}public String getTxServiceGroup() {return this.txServiceGroup;}public void setTxServiceGroup(String txServiceGroup) {this.txServiceGroup txServiceGroup;}
}
获取 applicationId 和 txServiceGroup 后创建 GlobalTransactionScanner 对象主要看类中 initClient 方法。
private void initClient() {if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {throw new IllegalArgumentException(applicationId: applicationId , txServiceGroup: txServiceGroup);}//init TMTMClient.init(applicationId, txServiceGroup);//init RMRMClient.init(applicationId, txServiceGroup);}
方法中可以看到初始化了 TMClient 和 RMClient对于一个服务既可以是TM角色也可以是RM角色至于什么时候是 TM 或者 RM 则要看在一次全局事务中 GlobalTransactional 注解标注在哪。 Client 创建的结果是与 TC 的一个 Netty 连接所以在启动日志中可以看到两个 Netty Channel其中标明了 transactionRole 分别为 TMROLE 和 RMROLE。
2019-04-09 13:42:57.417 INFO 93715 --- [imeoutChecker_1] c.a.f.c.rpc.netty.NettyPoolableFactory : NettyPool create channel to {address:127.0.0.1:8091,message:{applicationId:business-service,byteBuffer:{char:\u0000,direct:false,double:0.0,float:0.0,int:0,long:0,readOnly:false,short:0},transactionServiceGroup:my_test_tx_group,typeCode:101,version:0.4.1},transactionRole:TMROLE}
2019-04-09 13:42:57.505 INFO 93715 --- [imeoutChecker_1] c.a.f.c.rpc.netty.NettyPoolableFactory : NettyPool create channel to {address:127.0.0.1:8091,message:{applicationId:business-service,byteBuffer:{char:\u0000,direct:false,double:0.0,float:0.0,int:0,long:0,readOnly:false,short:0},transactionServiceGroup:my_test_tx_group,typeCode:103,version:0.4.1},transactionRole:RMROLE}
2019-04-09 13:42:57.629 DEBUG 93715 --- [lector_TMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:RegisterTMRequest{applicationIdbusiness-service, transactionServiceGroupmy_test_tx_group}
2019-04-09 13:42:57.629 DEBUG 93715 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:RegisterRMRequest{resourceIdsnull, applicationIdbusiness-service, transactionServiceGroupmy_test_tx_group}
2019-04-09 13:42:57.699 DEBUG 93715 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:version0.4.1,extraDatanull,identifiedtrue,resultCodenull,msgnull,messageId:1
2019-04-09 13:42:57.699 DEBUG 93715 --- [lector_TMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:version0.4.1,extraDatanull,identifiedtrue,resultCodenull,msgnull,messageId:2
2019-04-09 13:42:57.701 DEBUG 93715 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : com.alibaba.fescar.core.rpc.netty.RmRpcClient3b06d101 msgId:1, future :com.alibaba.fescar.core.protocol.MessageFuture28bb1abd, body:version0.4.1,extraDatanull,identifiedtrue,resultCodenull,msgnull
2019-04-09 13:42:57.701 DEBUG 93715 --- [lector_TMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : com.alibaba.fescar.core.rpc.netty.TmRpcClient65fc3fb7 msgId:2, future :com.alibaba.fescar.core.protocol.MessageFuture9a1e3df, body:version0.4.1,extraDatanull,identifiedtrue,resultCodenull,msgnull
2019-04-09 13:42:57.710 INFO 93715 --- [imeoutChecker_1] c.a.fescar.core.rpc.netty.RmRpcClient : register RM success. server version:0.4.1,channel:[id: 0xe6468995, L:/127.0.0.1:57397 - R:/127.0.0.1:8091]
2019-04-09 13:42:57.710 INFO 93715 --- [imeoutChecker_1] c.a.f.c.rpc.netty.NettyPoolableFactory : register success, cost 114 ms, version:0.4.1,role:TMROLE,channel:[id: 0xd22fe0c5, L:/127.0.0.1:57398 - R:/127.0.0.1:8091]
2019-04-09 13:42:57.711 INFO 93715 --- [imeoutChecker_1] c.a.f.c.rpc.netty.NettyPoolableFactory : register success, cost 125 ms, version:0.4.1,role:RMROLE,channel:[id: 0xe6468995, L:/127.0.0.1:57397 - R:/127.0.0.1:8091]
日志中可以看到
创建Netty连接发送注册请求得到响应结果RmRpcClient、TmRpcClient 成功实例化
TM 处理流程
在本例中TM 的角色是 business-service, BusinessService 的 purchase 方法标注了 GlobalTransactional 注解
Service
public class BusinessService {Autowiredprivate StorageFeignClient storageFeignClient;Autowiredprivate OrderFeignClient orderFeignClient;GlobalTransactionalpublic void purchase(String userId, String commodityCode, int orderCount){storageFeignClient.deduct(commodityCode, orderCount);orderFeignClient.create(userId, commodityCode, orderCount);}
}
方法调用后将会创建一个全局事务首先关注 GlobalTransactional 注解的作用在 GlobalTransactionalInterceptor 中被拦截处理。
/*** AOP拦截方法调用*/
Override
public Object invoke(final MethodInvocation methodInvocation) throws Throwable {Class? targetClass (methodInvocation.getThis() ! null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null);Method specificMethod ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);final Method method BridgeMethodResolver.findBridgedMethod(specificMethod);//获取方法GlobalTransactional注解final GlobalTransactional globalTransactionalAnnotation getAnnotation(method, GlobalTransactional.class);final GlobalLock globalLockAnnotation getAnnotation(method, GlobalLock.class);//如果方法有GlobalTransactional注解则拦截到相应方法处理if (globalTransactionalAnnotation ! null) {return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);} else if (globalLockAnnotation ! null) {return handleGlobalLock(methodInvocation);} else {return methodInvocation.proceed();}
}
handleGlobalTransaction 方法中对 TransactionalTemplate 的 execute 进行了调用从类名可以看到这是一个标准的模版方法它定义了 TM 对全局事务处理的标准步骤注释已经比较清楚了。
public Object execute(TransactionalExecutor business) throws TransactionalExecutor.ExecutionException {// 1. get or create a transactionGlobalTransaction tx GlobalTransactionContext.getCurrentOrCreate();try {// 2. begin transactiontry {triggerBeforeBegin();tx.begin(business.timeout(), business.name());triggerAfterBegin();} catch (TransactionException txe) {throw new TransactionalExecutor.ExecutionException(tx, txe,TransactionalExecutor.Code.BeginFailure);}Object rs null;try {// Do Your Businessrs business.execute();} catch (Throwable ex) {// 3. any business exception, rollback.try {triggerBeforeRollback();tx.rollback();triggerAfterRollback();// 3.1 Successfully rolled backthrow new TransactionalExecutor.ExecutionException(tx, TransactionalExecutor.Code.RollbackDone, ex);} catch (TransactionException txe) {// 3.2 Failed to rollbackthrow new TransactionalExecutor.ExecutionException(tx, txe,TransactionalExecutor.Code.RollbackFailure, ex);}}// 4. everything is fine, commit.try {triggerBeforeCommit();tx.commit();triggerAfterCommit();} catch (TransactionException txe) {// 4.1 Failed to committhrow new TransactionalExecutor.ExecutionException(tx, txe,TransactionalExecutor.Code.CommitFailure);}return rs;} finally {//5. cleartriggerAfterCompletion();cleanUp();}
}
通过 DefaultGlobalTransaction 的 begin 方法开启全局事务。
public void begin(int timeout, String name) throws TransactionException {if (role ! GlobalTransactionRole.Launcher) {check();if (LOGGER.isDebugEnabled()) {LOGGER.debug(Ignore Begin(): just involved in global transaction [ xid ]);}return;}if (xid ! null) {throw new IllegalStateException();}if (RootContext.getXID() ! null) {throw new IllegalStateException();}//具体开启事务的方法获取TC返回的XIDxid transactionManager.begin(null, null, name, timeout);status GlobalStatus.Begin;RootContext.bind(xid);if (LOGGER.isDebugEnabled()) {LOGGER.debug(Begin a NEW global transaction [ xid ]);}
}
方法开头处if (role ! GlobalTransactionRole.Launcher)对 role 的判断有关键的作用表明当前是全局事务的发起者Launcher还是参与者Participant。如果在分布式事务的下游系统方法中也加上GlobalTransactional注解那么它的角色就是 Participant会忽略后面的 begin 直接 return而判断是 Launcher 还是 Participant 是根据当前上下文是否已存在 XID 来判断没有 XID 的就是 Launcher已经存在 XID的就是 Participant。由此可见全局事务的创建只能由 Launcher 执行而一次分布式事务中也只有一个Launcher 存在。
DefaultTransactionManager负责 TM 与 TC 通讯发送 begin、commit、rollback 指令。
Override
public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)throws TransactionException {GlobalBeginRequest request new GlobalBeginRequest();request.setTransactionName(name);request.setTimeout(timeout);GlobalBeginResponse response (GlobalBeginResponse)syncCall(request);return response.getXid();
}
至此拿到 fescar-server 返回的 XID 表示一个全局事务创建成功日志中也反应了上述流程。
2019-04-09 13:46:57.417 DEBUG 31326 --- [nio-8084-exec-1] c.a.f.c.rpc.netty.AbstractRpcRemoting : offer message: timeout60000,transactionNamepurchase(java.lang.String,java.lang.String,int)
2019-04-09 13:46:57.417 DEBUG 31326 --- [geSend_TMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : write message:FescarMergeMessage timeout60000,transactionNamepurchase(java.lang.String,java.lang.String,int), channel:[id: 0xa148545e, L:/127.0.0.1:56120 - R:/127.0.0.1:8091],active?true,writable?true,isopen?true
2019-04-09 13:46:57.418 DEBUG 31326 --- [lector_TMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:FescarMergeMessage timeout60000,transactionNamepurchase(java.lang.String,java.lang.String,int)
2019-04-09 13:46:57.421 DEBUG 31326 --- [lector_TMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:MergeResultMessage com.alibaba.fescar.core.protocol.transaction.GlobalBeginResponse2dc480dc,messageId:1196
2019-04-09 13:46:57.421 DEBUG 31326 --- [nio-8084-exec-1] c.a.fescar.core.context.RootContext : bind 192.168.224.93:8091:2008502699
2019-04-09 13:46:57.421 DEBUG 31326 --- [nio-8084-exec-1] c.a.f.tm.api.DefaultGlobalTransaction : Begin a NEW global transaction [192.168.224.93:8091:2008502699]
全局事务创建后就开始执行 business.execute()即业务代码storageFeignClient.deduct(commodityCode, orderCount)进入 RM 处理流程此处的业务逻辑为调用 storage-service 的扣减库存接口。
RM 处理流程
GetMapping(path /deduct)
public Boolean deduct(String commodityCode, Integer count){storageService.deduct(commodityCode,count);return true;
}Transactional
public void deduct(String commodityCode, int count){Storage storage storageDAO.findByCommodityCode(commodityCode);storage.setCount(storage.getCount()-count);storageDAO.save(storage);
}
storage 的接口和 service 方法并未出现 fescar 相关的代码和注解体现了 fescar 的无侵入。那它是如何加入到这次全局事务中的呢答案在ConnectionProxy中这也是前面说为什么必须要使用DataSourceProxy的原因通过 DataSourceProxy 才能在业务代码的本地事务提交时fescar 通过该切入点向 TC 注册分支事务并发送 RM 的处理结果。
由于业务代码本身的事务提交被ConnectionProxy代理实现所以在提交本地事务时实际执行的是ConnectionProxy 的 commit 方法。
public void commit() throws SQLException {//如果当前是全局事务则执行全局事务的提交//判断是不是全局事务就是看当前上下文是否存在XIDif (context.inGlobalTransaction()) {processGlobalTransactionCommit();} else if (context.isGlobalLockRequire()) {processLocalCommitWithGlobalLocks();} else {targetConnection.commit();}
}private void processGlobalTransactionCommit() throws SQLException {try {//首先是向TC注册RM拿到TC分配的branchIdregister();} catch (TransactionException e) {recognizeLockKeyConflictException(e);}try {if (context.hasUndoLog()) {//写入undologUndoLogManager.flushUndoLogs(this);}//提交本地事务写入undo_log和业务数据在同一个本地事务中targetConnection.commit();} catch (Throwable ex) {//向TC发送RM的事务处理失败的通知report(false);if (ex instanceof SQLException) {throw new SQLException(ex);}}//向TC发送RM的事务处理成功的通知report(true);context.reset();
}private void register() throws TransactionException {//注册RM构建request通过netty向TC发送注册指令Long branchId DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(),null, context.getXid(), null, context.buildLockKeys());//将返回的branchId存在上下文中context.setBranchId(branchId);
}
通过日志印证一下上面的流程。
2019-04-09 21:57:48.341 DEBUG 38933 --- [nio-8081-exec-1] o.s.c.a.f.web.FescarHandlerInterceptor : xid in RootContext null xid in RpcContext 192.168.0.2:8091:2008546211
2019-04-09 21:57:48.341 DEBUG 38933 --- [nio-8081-exec-1] c.a.fescar.core.context.RootContext : bind 192.168.0.2:8091:2008546211
2019-04-09 21:57:48.341 DEBUG 38933 --- [nio-8081-exec-1] o.s.c.a.f.web.FescarHandlerInterceptor : bind 192.168.0.2:8091:2008546211 to RootContext
2019-04-09 21:57:48.386 INFO 38933 --- [nio-8081-exec-1] o.h.h.i.QueryTranslatorFactoryInitiator : HHH000397: Using ASTQueryTranslatorFactory
Hibernate: select storage0_.id as id1_0_, storage0_.commodity_code as commodit2_0_, storage0_.count as count3_0_ from storage_tbl storage0_ where storage0_.commodity_code?
Hibernate: update storage_tbl set count? where id?
2019-04-09 21:57:48.673 INFO 38933 --- [nio-8081-exec-1] c.a.fescar.core.rpc.netty.RmRpcClient : will connect to 192.168.0.2:8091
2019-04-09 21:57:48.673 INFO 38933 --- [nio-8081-exec-1] c.a.fescar.core.rpc.netty.RmRpcClient : RM will register :jdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse
2019-04-09 21:57:48.673 INFO 38933 --- [nio-8081-exec-1] c.a.f.c.rpc.netty.NettyPoolableFactory : NettyPool create channel to {address:192.168.0.2:8091,message:{applicationId:storage-service,byteBuffer:{char:\u0000,direct:false,double:0.0,float:0.0,int:0,long:0,readOnly:false,short:0},resourceIds:jdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse,transactionServiceGroup:hello-service-fescar-service-group,typeCode:103,version:0.4.0},transactionRole:RMROLE}
2019-04-09 21:57:48.677 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:RegisterRMRequest{resourceIdsjdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse, applicationIdstorage-service, transactionServiceGrouphello-service-fescar-service-group}
2019-04-09 21:57:48.680 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:version0.4.1,extraDatanull,identifiedtrue,resultCodenull,msgnull,messageId:9
2019-04-09 21:57:48.680 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : com.alibaba.fescar.core.rpc.netty.RmRpcClient7d61f5d4 msgId:9, future :com.alibaba.fescar.core.protocol.MessageFuture186cd3e0, body:version0.4.1,extraDatanull,identifiedtrue,resultCodenull,msgnull
2019-04-09 21:57:48.680 INFO 38933 --- [nio-8081-exec-1] c.a.fescar.core.rpc.netty.RmRpcClient : register RM success. server version:0.4.1,channel:[id: 0xd40718e3, L:/192.168.0.2:62607 - R:/192.168.0.2:8091]
2019-04-09 21:57:48.680 INFO 38933 --- [nio-8081-exec-1] c.a.f.c.rpc.netty.NettyPoolableFactory : register success, cost 3 ms, version:0.4.1,role:RMROLE,channel:[id: 0xd40718e3, L:/192.168.0.2:62607 - R:/192.168.0.2:8091]
2019-04-09 21:57:48.680 DEBUG 38933 --- [nio-8081-exec-1] c.a.f.c.rpc.netty.AbstractRpcRemoting : offer message: transactionId2008546211,branchTypeAT,resourceIdjdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse,lockKeystorage_tbl:1
2019-04-09 21:57:48.681 DEBUG 38933 --- [geSend_RMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : write message:FescarMergeMessage transactionId2008546211,branchTypeAT,resourceIdjdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse,lockKeystorage_tbl:1, channel:[id: 0xd40718e3, L:/192.168.0.2:62607 - R:/192.168.0.2:8091],active?true,writable?true,isopen?true
2019-04-09 21:57:48.681 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:FescarMergeMessage transactionId2008546211,branchTypeAT,resourceIdjdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse,lockKeystorage_tbl:1
2019-04-09 21:57:48.687 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:MergeResultMessage BranchRegisterResponse: transactionId2008546211,branchId2008546212,result code Success,getMsg null,messageId:11
2019-04-09 21:57:48.702 DEBUG 38933 --- [nio-8081-exec-1] c.a.f.rm.datasource.undo.UndoLogManager : Flushing UNDO LOG: {branchId:2008546212,sqlUndoLogs:[{afterImage:{rows:[{fields:[{keyType:PrimaryKey,name:id,type:4,value:1},{keyType:NULL,name:count,type:4,value:993}]}],tableName:storage_tbl},beforeImage:{rows:[{fields:[{keyType:PrimaryKey,name:id,type:4,value:1},{keyType:NULL,name:count,type:4,value:994}]}],tableName:storage_tbl},sqlType:UPDATE,tableName:storage_tbl}],xid:192.168.0.2:8091:2008546211}
2019-04-09 21:57:48.755 DEBUG 38933 --- [nio-8081-exec-1] c.a.f.c.rpc.netty.AbstractRpcRemoting : offer message: transactionId2008546211,branchId2008546212,resourceIdnull,statusPhaseOne_Done,applicationDatanull
2019-04-09 21:57:48.755 DEBUG 38933 --- [geSend_RMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : write message:FescarMergeMessage transactionId2008546211,branchId2008546212,resourceIdnull,statusPhaseOne_Done,applicationDatanull, channel:[id: 0xd40718e3, L:/192.168.0.2:62607 - R:/192.168.0.2:8091],active?true,writable?true,isopen?true
2019-04-09 21:57:48.756 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:FescarMergeMessage transactionId2008546211,branchId2008546212,resourceIdnull,statusPhaseOne_Done,applicationDatanull
2019-04-09 21:57:48.758 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:MergeResultMessage com.alibaba.fescar.core.protocol.transaction.BranchReportResponse582a08cf,messageId:13
2019-04-09 21:57:48.799 DEBUG 38933 --- [nio-8081-exec-1] c.a.fescar.core.context.RootContext : unbind 192.168.0.2:8091:2008546211
2019-04-09 21:57:48.799 DEBUG 38933 --- [nio-8081-exec-1] o.s.c.a.f.web.FescarHandlerInterceptor : unbind 192.168.0.2:8091:2008546211 from RootContext
获取business-service传来的XID绑定XID到当前上下文中执行业务逻辑sql向TC创建本次RM的Netty连接向TC发送分支事务的相关信息获得TC返回的branchId记录Undo Log数据向TC发送本次事务PhaseOne阶段的处理结果从当前上下文中解绑XID
其中第 1 步和第 9 步是在FescarHandlerInterceptor中完成的该类并不属于 fescar是前面提到的 spring-cloud-alibaba-fescar,它实现了基于 feign、rest 通信时将 xid bind 和 unbind 到当前请求上下文中。到这里 RM 完成了 PhaseOne 阶段的工作接着看 PhaseTwo 阶段的处理逻辑。
事务提交
各分支事务执行完成后TC 对各 RM 的汇报结果进行汇总给各 RM 发送 commit 或 rollback 的指令。
2019-04-09 21:57:49.813 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:xid192.168.0.2:8091:2008546211,branchId2008546212,branchTypeAT,resourceIdjdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse,applicationDatanull,messageId:1
2019-04-09 21:57:49.813 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : com.alibaba.fescar.core.rpc.netty.RmRpcClient7d61f5d4 msgId:1, body:xid192.168.0.2:8091:2008546211,branchId2008546212,branchTypeAT,resourceIdjdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse,applicationDatanull
2019-04-09 21:57:49.814 INFO 38933 --- [atch_RMROLE_1_8] c.a.f.core.rpc.netty.RmMessageListener : onMessage:xid192.168.0.2:8091:2008546211,branchId2008546212,branchTypeAT,resourceIdjdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse,applicationDatanull
2019-04-09 21:57:49.816 INFO 38933 --- [atch_RMROLE_1_8] com.alibaba.fescar.rm.AbstractRMHandler : Branch committing: 192.168.0.2:8091:2008546211 2008546212 jdbc:mysql://127.0.0.1:3306/db_storage?useSSLfalse null
2019-04-09 21:57:49.816 INFO 38933 --- [atch_RMROLE_1_8] com.alibaba.fescar.rm.AbstractRMHandler : Branch commit result: PhaseTwo_Committed
2019-04-09 21:57:49.817 INFO 38933 --- [atch_RMROLE_1_8] c.a.fescar.core.rpc.netty.RmRpcClient : RmRpcClient sendResponse branchStatusPhaseTwo_Committed,result code Success,getMsg null
2019-04-09 21:57:49.817 DEBUG 38933 --- [atch_RMROLE_1_8] c.a.f.c.rpc.netty.AbstractRpcRemoting : send response:branchStatusPhaseTwo_Committed,result code Success,getMsg null,channel:[id: 0xd40718e3, L:/192.168.0.2:62607 - R:/192.168.0.2:8091]
2019-04-09 21:57:49.817 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:branchStatusPhaseTwo_Committed,result code Success,getMsg null
从日志中可以看到
RM 收到 XID192.168.0.2:8091:2008546211branchId2008546212 的 commit 通知执行 commit 动作将 commit 结果发送给 TCbranchStatus 为 PhaseTwo_Committed
具体看下二阶段 commit 的执行过程在AbstractRMHandler类的 doBranchCommit 方法
/*** 拿到通知的xid、branchId等关键参数* 然后调用RM的branchCommit*/
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();LOGGER.info(Branch committing: xid branchId resourceId applicationData);BranchStatus status getResourceManager().branchCommit(request.getBranchType(), xid, branchId, resourceId, applicationData);response.setBranchStatus(status);LOGGER.info(Branch commit result: status);
}
最终会将 branchCommit 的请求调用到AsyncWorker的 branchCommit 方法。AsyncWorker 的处理方式是fescar 架构的一个关键部分因为大部分事务都是会正常提交的所以在 PhaseOne 阶段就已经结束了这样就可以将锁最快的释放。PhaseTwo 阶段接收 commit 的指令后异步处理即可。将 PhaseTwo 的时间消耗排除在一次分布式事务之外。
private static final ListPhase2Context ASYNC_COMMIT_BUFFER Collections.synchronizedList( new ArrayListPhase2Context());/*** 将需要提交的XID加入list*/
Override
public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId, String applicationData) throws TransactionException {if (ASYNC_COMMIT_BUFFER.size() ASYNC_COMMIT_BUFFER_LIMIT) {ASYNC_COMMIT_BUFFER.add(new Phase2Context(branchType, xid, branchId, resourceId, applicationData));} else {LOGGER.warn(Async commit buffer is FULL. Rejected branch [ branchId / xid ] will be handled by housekeeping later.);}return BranchStatus.PhaseTwo_Committed;
}/*** 通过定时任务消费list中的XID*/
public synchronized void init() {LOGGER.info(Async Commit Buffer Limit: ASYNC_COMMIT_BUFFER_LIMIT);timerExecutor new ScheduledThreadPoolExecutor(1,new NamedThreadFactory(AsyncWorker, 1, true));timerExecutor.scheduleAtFixedRate(new Runnable() {Overridepublic void run() {try {doBranchCommits();} catch (Throwable e) {LOGGER.info(Failed at async committing ... e.getMessage());}}}, 10, 1000 * 1, TimeUnit.MILLISECONDS);
}private void doBranchCommits() {if (ASYNC_COMMIT_BUFFER.size() 0) {return;}MapString, ListPhase2Context mappedContexts new HashMap();IteratorPhase2Context iterator ASYNC_COMMIT_BUFFER.iterator();//一次定时循环取出ASYNC_COMMIT_BUFFER中的所有待办数据//以resourceId作为key分组待commit数据resourceId是一个数据库的连接url//在前面的日志中可以看到目的是为了覆盖应用的多数据源创建while (iterator.hasNext()) {Phase2Context commitContext iterator.next();ListPhase2Context contextsGroupedByResourceId mappedContexts.get(commitContext.resourceId);if (contextsGroupedByResourceId null) {contextsGroupedByResourceId new ArrayList();mappedContexts.put(commitContext.resourceId, contextsGroupedByResourceId);}contextsGroupedByResourceId.add(commitContext);iterator.remove();}for (Map.EntryString, ListPhase2Context entry : mappedContexts.entrySet()) {Connection conn null;try {try {//根据resourceId获取数据源以及连接DataSourceProxy dataSourceProxy DataSourceManager.get().get(entry.getKey());conn dataSourceProxy.getPlainConnection();} catch (SQLException sqle) {LOGGER.warn(Failed to get connection for async committing on entry.getKey(), sqle);continue;}ListPhase2Context contextsGroupedByResourceId entry.getValue();for (Phase2Context commitContext : contextsGroupedByResourceId) {try {//执行undolog的处理即删除xid、branchId对应的记录UndoLogManager.deleteUndoLog(commitContext.xid, commitContext.branchId, conn);} catch (Exception ex) {LOGGER.warn(Failed to delete undo log [ commitContext.branchId / commitContext.xid ], ex);}}} finally {if (conn ! null) {try {conn.close();} catch (SQLException closeEx) {LOGGER.warn(Failed to close JDBC resource while deleting undo_log , closeEx);}}}}
}
所以对于commit动作的处理RM只需删除xid、branchId对应的undo_log即可。
事务回滚
对于rollback场景的触发有两种情况
分支事务处理异常即ConnectionProxy中report(false)的情况TM捕获到下游系统上抛的异常即发起全局事务标有GlobalTransactional注解的方法捕获到的异常。在前面TransactionalTemplate类的execute模版方法中对business.execute()的调用进行了catchcatch后会调用rollback由TM通知TC对应XID需要回滚事务
public void rollback() throws TransactionException {//只有Launcher能发起这个rollbackif (role GlobalTransactionRole.Participant) {// Participant has no responsibility of committingif (LOGGER.isDebugEnabled()) {LOGGER.debug(Ignore Rollback(): just involved in global transaction [ xid ]);}return;}if (xid null) {throw new IllegalStateException();}status transactionManager.rollback(xid);if (RootContext.getXID() ! null) {if (xid.equals(RootContext.getXID())) {RootContext.unbind();}}
}
TC 汇总后向参与者发送 rollback 指令RM 在AbstractRMHandler类的 doBranchRollback 方法中接收这个rollback 的通知。
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();LOGGER.info(Branch rolling back: xid branchId resourceId);BranchStatus status getResourceManager().branchRollback(request.getBranchType(), xid, branchId, resourceId, applicationData);response.setBranchStatus(status);LOGGER.info(Branch rollback result: status);
}
然后将 rollback 请求传递到DataSourceManager类的 branchRollback 方法。
public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId, String applicationData) throws TransactionException {//根据resourceId获取对应的数据源DataSourceProxy dataSourceProxy get(resourceId);if (dataSourceProxy null) {throw new ShouldNeverHappenException();}try {UndoLogManager.undo(dataSourceProxy, xid, branchId);} catch (TransactionException te) {if (te.getCode() TransactionExceptionCode.BranchRollbackFailed_Unretriable) {return BranchStatus.PhaseTwo_RollbackFailed_Unretryable;} else {return BranchStatus.PhaseTwo_RollbackFailed_Retryable;}}return BranchStatus.PhaseTwo_Rollbacked;
}
最终会执行UndoLogManager类的 undo 方法因为是纯 jdbc 操作代码比较长就不贴出来了可以通过连接到github 查看源码说一下 undo 的具体流程
根据 xid 和 branchId 查找 PhaseOne 阶段提交的 undo_log如果找到了就根据 undo_log 中记录的数据生成回放 sql 并执行即还原 PhaseOne 阶段修改的数据第 2 步处理完后删除该条 undo_log 数据如果第 1 步没有找到对应的 undo_log就插入一条状态为GlobalFinished的 undo_log。出现没找到的原因可能是 PhaseOne 阶段的本地事务异常了导致没有正常写入。 因为 xid 和 branchId 是唯一索引所以第 4步的插入可以防止 PhaseOne 阶段恢复后的成功写入那么 PhaseOne 阶段就会异常这样一来业务数据也就不会提交成功数据达到了最终回滚了的效果。
总结
本地结合分布式业务场景分析了 fescar client 侧的主要处理流程对 TM 和 RM 角色的主要源码进行了解析希望能对大家理解 fescar 的工作原理有所帮助。
随着 fescar 的快速迭代以及后期 Roadmap 规划的不断完善假以时日相信 fescar 能够成为开源分布式事务的标杆解决方案。
原文链接 本文为云栖社区原创内容未经允许不得转载。
