當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

源码｜详解分布式事务之 Seata-Client 原理及流程

發(fā)布時(shí)間：2024/8/23 编程问答 31 豆豆

生活随笔收集整理的這篇文章主要介紹了源码｜详解分布式事务之 Seata-Client 原理及流程小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

前言

在分布式系統(tǒng)中，分布式事務(wù)是一個(gè)必須要解決的問題，目前使用較多的是最終一致性方案。自年初阿里開源了Fescar（四月初更名為Seata）后，該項(xiàng)目受到了極大的關(guān)注，目前已接近 8000 Star。Seata?以高性能和零侵入的特性為目標(biāo)解決微服務(wù)領(lǐng)域的分布式事務(wù)難題，目前正處于快速迭代中，近期小目標(biāo)是生產(chǎn)可用的 Mysql 版本。

本文主要基于 spring cloud + spring jpa + spring cloud alibaba fescar + mysql + seata 的結(jié)構(gòu)，搭建一個(gè)分布式系統(tǒng)的 demo，通過 seata 的 debug 日志和源代碼，從 client 端（RM、TM）的角度分析其工作流程及原理。

為了更好地理解全文，我們來熟悉一下相關(guān)概念：

XID：全局事務(wù)的唯一標(biāo)識，由 ip:port:sequence 組成；
Transaction Coordinator (TC)：事務(wù)協(xié)調(diào)器，維護(hù)全局事務(wù)的運(yùn)行狀態(tài)，負(fù)責(zé)協(xié)調(diào)并驅(qū)動全局事務(wù)的提交或回滾；
Transaction Manager (TM )：控制全局事務(wù)的邊界，負(fù)責(zé)開啟一個(gè)全局事務(wù)，并最終發(fā)起全局提交或全局回滾的決議；
Resource Manager (RM)：控制分支事務(wù)，負(fù)責(zé)分支注冊、狀態(tài)匯報(bào)，并接收事務(wù)協(xié)調(diào)器的指令，驅(qū)動分支（本地）事務(wù)的提交和回滾；

提示：文中代碼是基于 fescar-0.4.1 版本，由于項(xiàng)目剛更名為 seata 不久，其中一些包名、類名、jar包等名稱還沒統(tǒng)一更換過來，故下文中仍使用 fescar 進(jìn)行表述。

分布式框架支持

Fescar 使用 XID 表示一個(gè)分布式事務(wù)，XID 需要在一次分布式事務(wù)請求所涉的系統(tǒng)中進(jìn)行傳遞，從而向 feacar-server 發(fā)送分支事務(wù)的處理情況，以及接收 feacar-server 的 commit、rollback 指令。 Fescar 官方已支持全版本的 dubbo 協(xié)議，而對于 spring cloud（spring-boot）的分布式項(xiàng)目社區(qū)也提供了相應(yīng)的實(shí)現(xiàn)

<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-alibaba-fescar</artifactId><version>2.1.0.BUILD-SNAPSHOT</version> </dependency>

該組件實(shí)現(xiàn)了基于 RestTemplate、Feign 通信時(shí)的 XID 傳遞功能。

業(yè)務(wù)邏輯

業(yè)務(wù)邏輯是經(jīng)典的下訂單、扣余額、減庫存流程。根據(jù)模塊劃分為三個(gè)獨(dú)立的服務(wù)，且分別連接對應(yīng)的數(shù)據(jù)庫：

訂單：order-server
賬戶：account-server
庫存：storage-server

另外還有發(fā)起分布式事務(wù)的業(yè)務(wù)系統(tǒng)：

業(yè)務(wù)：business-server

項(xiàng)目結(jié)構(gòu)如下圖?：

正常業(yè)務(wù):

business發(fā)起購買請求

storage扣減庫存

order創(chuàng)建訂單

account扣減余額

異常業(yè)務(wù)：

business發(fā)起購買請求

storage扣減庫存

order創(chuàng)建訂單

account扣減余額異常

正常流程下 2、3、4 步的數(shù)據(jù)正常更新全局 commit，異常流程下的數(shù)據(jù)則由于第 4 步的異常報(bào)錯(cuò)全局回滾。

配置文件

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?中可以指定具體配置的形式，默認(rèn)使用 file 類型，在 file.conf 中有 3 部分配置內(nèi)容：

transport transport 部分的配置對應(yīng)?NettyServerConfig?類，用于定義 Netty 相關(guān)的參數(shù)，TM、RM 與 fescar-server 之間使用 Netty 進(jìn)行通信。

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的分布式事務(wù)disableGlobalTransaction = false }

client

client {#RM接收TC的commit通知后緩沖上限async.commit.buffer.limit = 10000lock {retry.internal = 10retry.times = 30} }

數(shù)據(jù)源 Proxy

除了前面的配置文件，fescar 在 AT 模式下稍微有點(diǎn)代碼量的地方就是對數(shù)據(jù)源的代理指定，且目前只能基于DruidDataSource的代理。（注：在最新發(fā)布的 0.4.2 版本中已支持任意數(shù)據(jù)源類型）

@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 無侵入的一方面就體現(xiàn)在?ConnectionProxy?的實(shí)現(xiàn)上，即分支事務(wù)加入全局事務(wù)的切入點(diǎn)是在本地事務(wù)的?commit?階段，這樣設(shè)計(jì)可以保證業(yè)務(wù)數(shù)據(jù)與?undo_log?是在一個(gè)本地事務(wù)中。

undo_log?是需要在業(yè)務(wù)庫上創(chuàng)建的一個(gè)表，fescar 依賴該表記錄每筆分支事務(wù)的狀態(tài)及二階段?rollback?的回放數(shù)據(jù)。不用擔(dān)心該表的數(shù)據(jù)量過大形成單點(diǎn)問題，在全局事務(wù)?commit?的場景下事務(wù)對應(yīng)的?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`) ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

啟動 Server

前往https://github.com/seata/seata/releases?下載與 Client 版本對應(yīng)的 fescar-server,避免由于版本的不同導(dǎo)致的協(xié)議不一致問題進(jìn)入解壓之后的 bin 目錄，執(zhí)行：

./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 的項(xiàng)目能夠自動加載，當(dāng)然也可以通過其他方式示例化?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初始化的發(fā)起類*/@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);} }

可以看到支持一個(gè)配置項(xiàng)FescarProperties，用于配置事務(wù)分組名稱：

spring.cloud.alibaba.fescar.tx-service-group=my_test_tx_group

如果不指定服務(wù)組，則默認(rèn)使用spring.application.name+ -fescar-service-group生成名稱，所以不指定spring.application.name啟動會報(bào)錯(cuò)。

@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 后，創(chuàng)建?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，對于一個(gè)服務(wù)既可以是TM角色也可以是RM角色，至于什么時(shí)候是 TM 或者 RM 則要看在一次全局事務(wù)中?@GlobalTransactional?注解標(biāo)注在哪。 Client 創(chuàng)建的結(jié)果是與 TC 的一個(gè) Netty 連接，所以在啟動日志中可以看到兩個(gè) Netty Channel，其中標(biāo)明了 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{applicationId='business-service', transactionServiceGroup='my_test_tx_group'} 2019-04-09 13:42:57.629 DEBUG 93715 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Send:RegisterRMRequest{resourceIds='null', applicationId='business-service', transactionServiceGroup='my_test_tx_group'} 2019-04-09 13:42:57.699 DEBUG 93715 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:version=0.4.1,extraData=null,identified=true,resultCode=null,msg=null,messageId:1 2019-04-09 13:42:57.699 DEBUG 93715 --- [lector_TMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:version=0.4.1,extraData=null,identified=true,resultCode=null,msg=null,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.RmRpcClient@3b06d101 msgId:1, future :com.alibaba.fescar.core.protocol.MessageFuture@28bb1abd, body:version=0.4.1,extraData=null,identified=true,resultCode=null,msg=null 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.TmRpcClient@65fc3fb7 msgId:2, future :com.alibaba.fescar.core.protocol.MessageFuture@9a1e3df, body:version=0.4.1,extraData=null,identified=true,resultCode=null,msg=null 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]

日志中可以看到

創(chuàng)建Netty連接

發(fā)送注冊請求

得到響應(yīng)結(jié)果

RmRpcClient、TmRpcClient?成功實(shí)例化

TM 處理流程

在本例中，TM 的角色是 business-service, BusinessService 的 purchase 方法標(biāo)注了?@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);} }

方法調(diào)用后將會創(chuàng)建一個(gè)全局事務(wù)，首先關(guān)注?@GlobalTransactional?注解的作用，在?GlobalTransactionalInterceptor?中被攔截處理。

/*** AOP攔截方法調(diào)用*/ @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注解，則攔截到相應(yīng)方法處理if (globalTransactionalAnnotation != null) {return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);} else if (globalLockAnnotation != null) {return handleGlobalLock(methodInvocation);} else {return methodInvocation.proceed();} }

handleGlobalTransaction?方法中對?TransactionalTemplate?的 execute 進(jìn)行了調(diào)用，從類名可以看到這是一個(gè)標(biāo)準(zhǔn)的模版方法，它定義了 TM 對全局事務(wù)處理的標(biāo)準(zhǔn)步驟，注釋已經(jīng)比較清楚了。

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 方法開啟全局事務(wù)。

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();}//具體開啟事務(wù)的方法，獲取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 的判斷有關(guān)鍵的作用，表明當(dāng)前是全局事務(wù)的發(fā)起者（Launcher）還是參與者（Participant）。如果在分布式事務(wù)的下游系統(tǒng)方法中也加上@GlobalTransactional注解，那么它的角色就是 Participant，會忽略后面的 begin 直接 return，而判斷是 Launcher 還是 Participant 是根據(jù)當(dāng)前上下文是否已存在 XID 來判斷，沒有 XID 的就是 Launcher，已經(jīng)存在 XID的就是 Participant。由此可見，全局事務(wù)的創(chuàng)建只能由 Launcher 執(zhí)行，而一次分布式事務(wù)中也只有一個(gè)Launcher 存在。

DefaultTransactionManager負(fù)責(zé) TM 與 TC 通訊，發(fā)送 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 表示一個(gè)全局事務(wù)創(chuàng)建成功，日志中也反應(yīng)了上述流程。

2019-04-09 13:46:57.417 DEBUG 31326 --- [nio-8084-exec-1] c.a.f.c.rpc.netty.AbstractRpcRemoting : offer message: timeout=60000,transactionName=purchase(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 timeout=60000,transactionName=purchase(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 timeout=60000,transactionName=purchase(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.GlobalBeginResponse@2dc480dc,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]

全局事務(wù)創(chuàng)建后，就開始執(zhí)行 business.execute()，即業(yè)務(wù)代碼storageFeignClient.deduct(commodityCode, orderCount)進(jìn)入 RM 處理流程，此處的業(yè)務(wù)邏輯為調(diào)用 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 方法并未出現(xiàn) fescar 相關(guān)的代碼和注解，體現(xiàn)了 fescar 的無侵入。那它是如何加入到這次全局事務(wù)中的呢？答案在ConnectionProxy中，這也是前面說為什么必須要使用DataSourceProxy的原因，通過 DataSourceProxy 才能在業(yè)務(wù)代碼的本地事務(wù)提交時(shí)，fescar 通過該切入點(diǎn)，向 TC 注冊分支事務(wù)并發(fā)送 RM 的處理結(jié)果。

由于業(yè)務(wù)代碼本身的事務(wù)提交被ConnectionProxy代理實(shí)現(xiàn)，所以在提交本地事務(wù)時(shí)，實(shí)際執(zhí)行的是ConnectionProxy 的 commit 方法。

public void commit() throws SQLException {//如果當(dāng)前是全局事務(wù)，則執(zhí)行全局事務(wù)的提交//判斷是不是全局事務(wù)，就是看當(dāng)前上下文是否存在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);}//提交本地事務(wù)，寫入undo_log和業(yè)務(wù)數(shù)據(jù)在同一個(gè)本地事務(wù)中targetConnection.commit();} catch (Throwable ex) {//向TC發(fā)送RM的事務(wù)處理失敗的通知report(false);if (ex instanceof SQLException) {throw new SQLException(ex);}}//向TC發(fā)送RM的事務(wù)處理成功的通知report(true);context.reset(); }private void register() throws TransactionException {//注冊RM，構(gòu)建request通過netty向TC發(fā)送注冊指令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?useSSL=false 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?useSSL=false","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{resourceIds='jdbc:mysql://127.0.0.1:3306/db_storage?useSSL=false', applicationId='storage-service', transactionServiceGroup='hello-service-fescar-service-group'} 2019-04-09 21:57:48.680 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:version=0.4.1,extraData=null,identified=true,resultCode=null,msg=null,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.RmRpcClient@7d61f5d4 msgId:9, future :com.alibaba.fescar.core.protocol.MessageFuture@186cd3e0, body:version=0.4.1,extraData=null,identified=true,resultCode=null,msg=null 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: transactionId=2008546211,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/db_storage?useSSL=false,lockKey=storage_tbl:1 2019-04-09 21:57:48.681 DEBUG 38933 --- [geSend_RMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : write message:FescarMergeMessage transactionId=2008546211,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/db_storage?useSSL=false,lockKey=storage_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 transactionId=2008546211,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/db_storage?useSSL=false,lockKey=storage_tbl:1 2019-04-09 21:57:48.687 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:MergeResultMessage BranchRegisterResponse: transactionId=2008546211,branchId=2008546212,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: transactionId=2008546211,branchId=2008546212,resourceId=null,status=PhaseOne_Done,applicationData=null 2019-04-09 21:57:48.755 DEBUG 38933 --- [geSend_RMROLE_1] c.a.f.c.rpc.netty.AbstractRpcRemoting : write message:FescarMergeMessage transactionId=2008546211,branchId=2008546212,resourceId=null,status=PhaseOne_Done,applicationData=null, 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 transactionId=2008546211,branchId=2008546212,resourceId=null,status=PhaseOne_Done,applicationData=null 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.BranchReportResponse@582a08cf,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到當(dāng)前上下文中

執(zhí)行業(yè)務(wù)邏輯sql

向TC創(chuàng)建本次RM的Netty連接

向TC發(fā)送分支事務(wù)的相關(guān)信息

獲得TC返回的branchId

記錄Undo Log數(shù)據(jù)

向TC發(fā)送本次事務(wù)PhaseOne階段的處理結(jié)果

從當(dāng)前上下文中解綁XID

其中第 1 步和第 9 步，是在FescarHandlerInterceptor中完成的，該類并不屬于 fescar，是前面提到的 spring-cloud-alibaba-fescar,它實(shí)現(xiàn)了基于 feign、rest 通信時(shí)將 xid bind 和 unbind 到當(dāng)前請求上下文中。到這里 RM 完成了 PhaseOne 階段的工作，接著看 PhaseTwo 階段的處理邏輯。

事務(wù)提交

各分支事務(wù)執(zhí)行完成后，TC 對各 RM 的匯報(bào)結(jié)果進(jìn)行匯總，給各 RM 發(fā)送 commit 或 rollback 的指令。

2019-04-09 21:57:49.813 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.rpc.netty.MessageCodecHandler : Receive:xid=192.168.0.2:8091:2008546211,branchId=2008546212,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/db_storage?useSSL=false,applicationData=null,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.RmRpcClient@7d61f5d4 msgId:1, body:xid=192.168.0.2:8091:2008546211,branchId=2008546212,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/db_storage?useSSL=false,applicationData=null 2019-04-09 21:57:49.814 INFO 38933 --- [atch_RMROLE_1_8] c.a.f.core.rpc.netty.RmMessageListener : onMessage:xid=192.168.0.2:8091:2008546211,branchId=2008546212,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/db_storage?useSSL=false,applicationData=null 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?useSSL=false 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 branchStatus=PhaseTwo_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:branchStatus=PhaseTwo_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:branchStatus=PhaseTwo_Committed,result code =Success,getMsg =null

從日志中可以看到

RM 收到 XID=192.168.0.2:8091:2008546211，branchId=2008546212 的 commit 通知；

執(zhí)行 commit 動作；

將 commit 結(jié)果發(fā)送給 TC，branchStatus 為 PhaseTwo_Committed；

具體看下二階段 commit 的執(zhí)行過程，在AbstractRMHandler類的 doBranchCommit 方法：

/*** 拿到通知的xid、branchId等關(guān)鍵參數(shù)* 然后調(diào)用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 的請求調(diào)用到AsyncWorker的 branchCommit 方法。AsyncWorker 的處理方式是fescar 架構(gòu)的一個(gè)關(guān)鍵部分，因?yàn)榇蟛糠质聞?wù)都是會正常提交的，所以在 PhaseOne 階段就已經(jīng)結(jié)束了，這樣就可以將鎖最快的釋放。PhaseTwo 階段接收 commit 的指令后，異步處理即可。將 PhaseTwo 的時(shí)間消耗排除在一次分布式事務(wù)之外。

private static final List<Phase2Context> ASYNC_COMMIT_BUFFER = Collections.synchronizedList( new ArrayList<Phase2Context>());/*** 將需要提交的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; }/*** 通過定時(shí)任務(wù)消費(fèi)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;}Map<String, List<Phase2Context>> mappedContexts = new HashMap<>();Iterator<Phase2Context> iterator = ASYNC_COMMIT_BUFFER.iterator();//一次定時(shí)循環(huán)取出ASYNC_COMMIT_BUFFER中的所有待辦數(shù)據(jù)//以resourceId作為key分組待commit數(shù)據(jù)，resourceId是一個(gè)數(shù)據(jù)庫的連接url//在前面的日志中可以看到，目的是為了覆蓋應(yīng)用的多數(shù)據(jù)源創(chuàng)建while (iterator.hasNext()) {Phase2Context commitContext = iterator.next();List<Phase2Context> contextsGroupedByResourceId = mappedContexts.get(commitContext.resourceId);if (contextsGroupedByResourceId == null) {contextsGroupedByResourceId = new ArrayList<>();mappedContexts.put(commitContext.resourceId, contextsGroupedByResourceId);}contextsGroupedByResourceId.add(commitContext);iterator.remove();}for (Map.Entry<String, List<Phase2Context>> entry : mappedContexts.entrySet()) {Connection conn = null;try {try {//根據(jù)resourceId獲取數(shù)據(jù)源以及連接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;}List<Phase2Context> contextsGroupedByResourceId = entry.getValue();for (Phase2Context commitContext : contextsGroupedByResourceId) {try {//執(zhí)行undolog的處理，即刪除xid、branchId對應(yīng)的記錄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對應(yīng)的undo_log即可。

事務(wù)回滾

對于rollback場景的觸發(fā)有兩種情況

分支事務(wù)處理異常，即ConnectionProxy中report(false)的情況

TM捕獲到下游系統(tǒng)上拋的異常，即發(fā)起全局事務(wù)標(biāo)有@GlobalTransactional注解的方法捕獲到的異常。在前面TransactionalTemplate類的execute模版方法中，對business.execute()的調(diào)用進(jìn)行了catch，catch后會調(diào)用rollback，由TM通知TC對應(yīng)XID需要回滾事務(wù)

public void rollback() throws TransactionException {//只有Launcher能發(fā)起這個(gè)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 匯總后向參與者發(fā)送 rollback 指令，RM 在AbstractRMHandler類的 doBranchRollback 方法中接收這個(gè)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 {//根據(jù)resourceId獲取對應(yīng)的數(shù)據(jù)源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; }

最終會執(zhí)行UndoLogManager類的 undo 方法，因?yàn)槭羌?jdbc 操作代碼比較長就不貼出來了，可以通過連接到github 查看源碼，說一下 undo 的具體流程：

根據(jù) xid 和 branchId 查找 PhaseOne 階段提交的 undo_log；

如果找到了就根據(jù) undo_log 中記錄的數(shù)據(jù)生成回放 sql 并執(zhí)行，即還原 PhaseOne 階段修改的數(shù)據(jù)；

第 2 步處理完后，刪除該條 undo_log 數(shù)據(jù)；

如果第 1 步?jīng)]有找到對應(yīng)的 undo_log，就插入一條狀態(tài)為GlobalFinished的 undo_log。出現(xiàn)沒找到的原因可能是 PhaseOne 階段的本地事務(wù)異常了，導(dǎo)致沒有正常寫入。因?yàn)?xid 和 branchId 是唯一索引，所以第 4步的插入，可以防止 PhaseOne 階段恢復(fù)后的成功寫入，那么 PhaseOne 階段就會異常，這樣一來業(yè)務(wù)數(shù)據(jù)也就不會提交成功，數(shù)據(jù)達(dá)到了最終回滾了的效果。

總結(jié)

本地結(jié)合分布式業(yè)務(wù)場景，分析了 fescar client 側(cè)的主要處理流程，對 TM 和 RM 角色的主要源碼進(jìn)行了解析，希望能對大家理解 fescar 的工作原理有所幫助。

隨著 fescar 的快速迭代以及后期 Roadmap 規(guī)劃的不斷完善，假以時(shí)日，相信 fescar 能夠成為開源分布式事務(wù)的標(biāo)桿解決方案。

原文鏈接
本文為云棲社區(qū)原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。

總結(jié)

以上是生活随笔為你收集整理的源码｜详解分布式事务之 Seata-Client 原理及流程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： 90后ACE成长记——从偏居一隅小城里走
下一篇：阿里小二的日常工作要被TA们“接管”了！