前言

在分布式系統(tǒng)中，分布式事務(wù)是需要解決的問題，目前更常用的是最終一致性方案。年初阿里開業(yè)的Fescar (4月初更名為Seata)后，該項目受到極大關(guān)注，目前已接近8000名明星。Seata以高性能和零入侵為目標，正在解決微服務(wù)領(lǐng)域的分布式事務(wù)問題，目前正在快速迭代，最近的小目標是生產(chǎn)可用的MySQL版本。

本文主要基于spring cloud spring JPA spring cloud Alibaba Fe scar MySQL seata的結(jié)構(gòu)構(gòu)建分布式系統(tǒng)的demo，并通過seata的debug日志和源代碼在客戶端(RM，TM)上(示例條目：)

為了更好地理解全文，讓我們熟悉相關(guān)概念。

XID:由ip:port:sequence組成的全局事務(wù)處理的唯一標識符。事務(wù)協(xié)調(diào)人(TC):事務(wù)協(xié)調(diào)人，保持全局事務(wù)執(zhí)行狀態(tài)，提交或回滾全局事務(wù)，調(diào)整和驅(qū)動事務(wù)管理人(TM):控制全局事務(wù)的界限，打開全局事務(wù)，最終啟動有關(guān)全局提交或全局回滾的決議。資源管理器(RM):控制分支事務(wù)、注冊分支、報告狀態(tài)和接收事務(wù)協(xié)調(diào)程序命令、提交分支(本地)事務(wù)和回滾提示：本文檔的代碼基于版本。由于項目重命名為seata后不久，包名稱、類名、jar包等名稱尚未統(tǒng)一替換，因此使用fescar進行如下說明：

分布式框架支持

Fescar使用XID來表示需要傳遞給分布式事務(wù)請求中涉及的系統(tǒng)的分布式事務(wù)。這將向feacar-server發(fā)送分支事務(wù)處理，并接收feacar-server的commit，rollback指令。Fescar支持完整版本的dubbo協(xié)議，并為spring cloud(spring-boot)的分布式項目社區(qū)提供了適當?shù)膶嵤?

Dependency

GroupIdorg。/groupId

artifactidspring-cloud-Alibaba-fescar/artifact id

版本2.1.0.build-快照/版本

/dependency

此組件實現(xiàn)了基于RestTemplate，F(xiàn)eign通信的XID轉(zhuǎn)發(fā)功能。

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

業(yè)務(wù)邏輯是經(jīng)典的訂貨、余額扣除、庫存減少過程。根據(jù)模塊，它分為三個單獨的服務(wù)，每個服務(wù)都連接到相應(yīng)的數(shù)據(jù)庫。

訂單：order-server帳戶：帳戶-服務(wù)器庫存：storage-server還有啟動分布式事務(wù)的業(yè)務(wù)系統(tǒng)。

工作：business-server項目結(jié)構(gòu)如下圖所示

正常工作：

啟動業(yè)務(wù)申請存儲扣減庫存訂單生成帳戶扣減馀額以上業(yè)務(wù)：

Business啟動申請storage可退回庫存訂單生成帳戶可退回馀額例外正常流程下的第2、3和4步數(shù)據(jù)正常更新全局commit，例外流程下的數(shù)據(jù)由于第4步例外而錯誤地報告全局回退。

配置文件

Fescar的配置門戶文件是regi。如果查看代碼ConfigurationFactory，您會發(fā)現(xiàn)當前無法指定配置文件，因此配置文件名只能是regi。

private static final string registry _ conf=' regi '；

public static final configuration file _ instance=new file configuration(注冊表_ conf)；

可以格式化注冊表中的特定配置。默認值為file類型。包含三個配置內(nèi)容。

Transport transport部分的配置對應(yīng)于NettyServerConfig類，定義了在TM、RM和fescar-server之間使用Nett的netty特定參數(shù)

y 進行通信。

service

service { #vgroup->rgroup vgrou = "default" #配置Client連接TC的地址 de = "127.0.0.1:8091" #degrade current not support enableDegrade = false #disable 是否啟用seata的分布式事務(wù) disableGlobalTransaction = false }

1. client

client { #RM接收TC的commit通知后緩沖上限 a = 10000 lock { re = 10 re = 30 } }

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

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

@Bean @ConfigurationProperties(prefix = ";) 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 的實現(xiàn)上，即分支事務(wù)加入全局事務(wù)的切入點是在本地事務(wù)的 commit 階段，這樣設(shè)計可以保證業(yè)務(wù)數(shù)據(jù)與 undo_log 是在一個本地事務(wù)中。

undo_log 是需要在業(yè)務(wù)庫上創(chuàng)建的一個表，fescar 依賴該表記錄每筆分支事務(wù)的狀態(tài)及二階段 rollback 的回放數(shù)據(jù)。不用擔心該表的數(shù)據(jù)量過大形成單點問題，在全局事務(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

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

. 8091 ../data

啟動成功輸出：

2019-04-09 20:27:24.637 INFO [main]c.a. -Server started ...

啟動 Client

fescar 的加載入口類位于 GlobalTransactionAutoConfiguration，對基于 spring boot 的項目能夠自動加載，當然也可以通過其他方式示例化 GlobalTransactionScanner。

@Configuration @EnableConfigurationProperties({Fe}) public class GlobalTransactionAutoConfiguration { private final ApplicationContext applicationContext; private final FescarProperties fescarProperties; public GlobalTransactionAutoConfiguration(ApplicationContext applicationContext, FescarProperties fescarProperties) { = applicationContext; = fescarProperties; } /** * 示例化GlobalTransactionScanner * scanner為client初始化的發(fā)起類 */ @Bean public GlobalTransactionScanner globalTransactionScanner() { String applicationName = .getEnvironment().getProperty(";); String txServiceGroup = .getTxServiceGroup(); if (txServiceGroup)) { txServiceGroup = applicationName + "-fescar-service-group"; .setTxServiceGroup(txServiceGroup); } return new GlobalTransactionScanner(applicationName, txServiceGroup); } }

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

如果不指定服務(wù)組，則默認使用 -fescar-service-group生成名稱，所以不指定啟動會報錯。

@ConfigurationProperties(";) public class FescarProperties { private String txServiceGroup; public FescarProperties() { } public String getTxServiceGroup() { return ; } public void setTxServiceGroup(String txServiceGroup) { = txServiceGroup; } }

獲取 applicationId 和 txServiceGroup 后，創(chuàng)建 GlobalTransactionScanner 對象，主要看類中 initClient 方法。

private void initClient() { if (applicationId) || S(txServiceGroup)) { throw new IllegalArgumentException( "applicationId: " + applicationId + ", txServiceGroup: " + txServiceGroup); } //init TM TMClient.init(applicationId, txServiceGroup); //init RM RMClient.init(applicationId, txServiceGroup); }

方法中可以看到初始化了 TMClient 和 RMClient，對于一個服務(wù)既可以是TM角色也可以是RM角色，至于什么時候是 TM 或者 RM 則要看在一次全局事務(wù)中 @GlobalTransactional 注解標注在哪。 Client 創(chuàng)建的結(jié)果是與 TC 的一個 Netty 連接，所以在啟動日志中可以看到兩個 Netty Channel，其中標明了 transactionRole 分別為 TMROLE 和 RMROLE。

2019-04-09 13:42:57.417 INFO 93715 --- [imeoutChecker_1] c.a.f.c.r : 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.r : 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.r : 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.r : 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.r : 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.r : 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.r : com.alibaba. 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.r : com.alibaba. 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. : 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.r : 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.r : 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]

日志中可以看到

1. 創(chuàng)建Netty連接
2. 發(fā)送注冊請求
3. 得到響應(yīng)結(jié)果
4. RmRpcClient、TmRpcClient 成功實例化

TM 處理流程

在本例中，TM 的角色是 business-service, BusinessService 的 purchase 方法標注了 @GlobalTransactional 注解：

@Service public class BusinessService { @Autowired private StorageFeignClient storageFeignClient; @Autowired private OrderFeignClient orderFeignClient; @GlobalTransactional public void purchase(String userId, String commodityCode, int orderCount){ (commodityCode, orderCount); orderFeignClient.create(userId, commodityCode, orderCount); } }

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

/** * AOP攔截方法調(diào)用 */ @Override public Object invoke(final MethodInvocation methodInvocation) throws Throwable { Class<?> targetClass = () != null ? Ao()) : null); Method specificMethod = Cla(), targetClass); final Method method = BridgeMe(specificMethod); //獲取方法GlobalTransactional注解 final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, GlobalTran); 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 me(); } }

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

public Object execute(TransactionalExecutor business) throws Tran { // 1. get or create a transaction GlobalTransaction tx = GlobalTran(); try { // 2. begin transaction try { triggerBeforeBegin(); (), bu()); triggerAfterBegin(); } catch (TransactionException txe) { throw new Tran(tx, txe, Tran); } Object rs = null; try { // Do Your Business rs = bu(); } catch (Throwable ex) { // 3. any business exception, rollback. try { triggerBeforeRollback(); (); triggerAfterRollback(); // 3.1 Successfully rolled back throw new Tran(tx, Tran, ex); } catch (TransactionException txe) { // 3.2 Failed to rollback throw new Tran(tx, txe, Tran, ex); } } // 4. everything is fine, commit. try { triggerBeforeCommit(); (); triggerAfterCommit(); } catch (TransactionException txe) { // 4.1 Failed to commit throw new Tran(tx, txe, Tran); } return rs; } finally { //5. clear triggerAfterCompletion(); cleanUp(); } }

通過 DefaultGlobalTransaction 的 begin 方法開啟全局事務(wù)。

public void begin(int timeout, String name) throws TransactionException { if (role != GlobalTran) { check(); if ()) { LOGGER.debug("Ignore Begin(): just involved in global transaction [" + xid + "]"); } return; } if (xid != null) { throw new IllegalStateException(); } if () != null) { throw new IllegalStateException(); } //具體開啟事務(wù)的方法，獲取TC返回的XID xid = (null, null, name, timeout); status = GlobalS; Roo(xid); if ()) { LOGGER.debug("Begin a NEW global transaction [" + xid + "]"); } }

方法開頭處if (role != GlobalTran)對 role 的判斷有關(guān)鍵的作用，表明當前是全局事務(wù)的發(fā)起者（Launcher）還是參與者（Participant）。如果在分布式事務(wù)的下游系統(tǒng)方法中也加上@GlobalTransactional注解，那么它的角色就是 Participant，會忽略后面的 begin 直接 return，而判斷是 Launcher 還是 Participant 是根據(jù)當前上下文是否已存在 XID 來判斷，沒有 XID 的就是 Launcher，已經(jīng)存在 XID的就是 Participant。由此可見，全局事務(wù)的創(chuàng)建只能由 Launcher 執(zhí)行，而一次分布式事務(wù)中也只有一個Launcher 存在。

DefaultTransactionManager負責 TM 與 TC 通訊，發(fā)送 begin、commit、rollback 指令。

@Override public String begin(String applicationId, String transactionServiceGroup, String name, int timeout) throws TransactionException { GlobalBeginRequest request = new GlobalBeginRequest(); reque(name); reque(timeout); GlobalBeginResponse response = (GlobalBeginResponse)syncCall(request); return re(); }

至此拿到 fescar-server 返回的 XID 表示一個全局事務(wù)創(chuàng)建成功，日志中也反應(yīng)了上述流程。

2019-04-09 13:46:57.417 DEBUG 31326 --- [nio-8084-exec-1] c.a.f.c.r : offer message: timeout=60000,transactionName=purchase) 2019-04-09 13:46:57.417 DEBUG 31326 --- [geSend_TMROLE_1] c.a.f.c.r : write message:FescarMergeMessage timeout=60000,transactionName=purchase), 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.r : Send:FescarMergeMessage timeout=60000,transactionName=purchase) 2019-04-09 13:46:57.421 DEBUG 31326 --- [lector_TMROLE_1] c.a.f.c.r : 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. : bind 192.168.224.93:8091:2008502699 2019-04-09 13:46:57.421 DEBUG 31326 --- [nio-8084-exec-1] c.a.f. : Begin a NEW global transaction [192.168.224.93:8091:2008502699]

全局事務(wù)創(chuàng)建后，就開始執(zhí)行 bu()，即業(yè)務(wù)代碼(commodityCode, orderCount)進入 RM 處理流程，此處的業(yè)務(wù)邏輯為調(diào)用 storage-service 的扣減庫存接口。

RM 處理流程

@GetMapping(path = "/deduct") public Boolean deduct(String commodityCode, Integer count){ (commodityCode,count); return true; } @Transactional public void deduct(String commodityCode, int count){ Storage storage = (commodityCode); ()-count); (storage); }

storage 的接口和 service 方法并未出現(xiàn) fescar 相關(guān)的代碼和注解，體現(xiàn)了 fescar 的無侵入。那它是如何加入到這次全局事務(wù)中的呢？答案在ConnectionProxy中，這也是前面說為什么必須要使用DataSourceProxy的原因，通過 DataSourceProxy 才能在業(yè)務(wù)代碼的本地事務(wù)提交時，fescar 通過該切入點，向 TC 注冊分支事務(wù)并發(fā)送 RM 的處理結(jié)果。

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

public void commit() throws SQLException { //如果當前是全局事務(wù)，則執(zhí)行全局事務(wù)的提交 //判斷是不是全局事務(wù)，就是看當前上下文是否存在XID if ()) { processGlobalTransactionCommit(); } else if ()) { processLocalCommitWithGlobalLocks(); } else { (); } } private void processGlobalTransactionCommit() throws SQLException { try { //首先是向TC注冊RM，拿到TC分配的branchId register(); } catch (TransactionException e) { recognizeLockKeyConflictException(e); } try { if ()) { //寫入undolog UndoLogManager.flushUndoLogs(this); } //提交本地事務(wù)，寫入undo_log和業(yè)務(wù)數(shù)據(jù)在同一個本地事務(wù)中 (); } catch (Throwable ex) { //向TC發(fā)送RM的事務(wù)處理失敗的通知 report(false); if (ex instanceof SQLException) { throw new SQLException(ex); } } //向TC發(fā)送RM的事務(wù)處理成功的通知 report(true); con(); } private void register() throws TransactionException { //注冊RM，構(gòu)建request通過netty向TC發(fā)送注冊指令 Long branchId = De().branchRegister, getDataSourceProxy().getResourceId(), null, con(), null, con()); //將返回的branchId存在上下文中 con(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. : 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 as id1_0_, as commodit2_0_, as count3_0_ from storage_tbl storage0_ where =? Hibernate: update storage_tbl set count=? where id=? 2019-04-09 21:57:48.673 INFO 38933 --- [nio-8081-exec-1] c.a. : will connect to 192.168.0.2:8091 2019-04-09 21:57:48.673 INFO 38933 --- [nio-8081-exec-1] c.a. : 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.r : 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.r : 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.r : 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.r : com.alibaba.@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. : 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.r : 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.r : 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.r : 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.r : 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.r : 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.da : 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.r : 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.r : 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.r : 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.r : 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. : 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

1. 獲取business-service傳來的XID
2. 綁定XID到當前上下文中
3. 執(zhí)行業(yè)務(wù)邏輯sql
4. 向TC創(chuàng)建本次RM的Netty連接
5. 向TC發(fā)送分支事務(wù)的相關(guān)信息
6. 獲得TC返回的branchId
7. 記錄Undo Log數(shù)據(jù)
8. 向TC發(fā)送本次事務(wù)PhaseOne階段的處理結(jié)果
9. 從當前上下文中解綁XID

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

事務(wù)提交

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

2019-04-09 21:57:49.813 DEBUG 38933 --- [lector_RMROLE_1] c.a.f.c.r : 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.r : com.alibaba.@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.r : 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. : 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. : Branch commit result: PhaseTwo_Committed 2019-04-09 21:57:49.817 INFO 38933 --- [atch_RMROLE_1_8] c.a. : 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.r : 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.r : Send:branchStatus=PhaseTwo_Committed,result code =Success,getMsg =null

從日志中可以看到

1. RM 收到 XID=192.168.0.2:8091:2008546211，branchId=2008546212 的 commit 通知；
2. 執(zhí)行 commit 動作；
3. 將 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 = reque(); long branchId = reque(); String resourceId = reque(); String applicationData = reque(); LOGGER.info("Branch committing: " + xid + " " + branchId + " " + resourceId + " " + applicationData); BranchStatus status = getResourceManager().branchCommi(), xid, branchId, resourceId, applicationData); re(status); LOGGER.info("Branch commit result: " + status); }

最終會將 branchCommit 的請求調(diào)用到AsyncWorker的 branchCommit 方法。AsyncWorker 的處理方式是fescar 架構(gòu)的一個關(guān)鍵部分，因為大部分事務(wù)都是會正常提交的，所以在 PhaseOne 階段就已經(jīng)結(jié)束了，這樣就可以將鎖最快的釋放。PhaseTwo 階段接收 commit 的指令后，異步處理即可。將 PhaseTwo 的時間消耗排除在一次分布式事務(wù)之外。

private static final List<Phase2Context> ASYNC_COMMIT_BUFFER = Collec( 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_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 Branc; } /** * 通過定時任務(wù)消費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)); (new Runnable() { @Override public void run() { try { doBranchCommits(); } catch (Throwable e) { LOGGER.info("Failed at async committing ... " + e.getMessage()); } } }, 10, 1000 * 1, TimeUnit.MILLISECONDS); } private void doBranchCommits() { if () == 0) { return; } Map<String, List<Phase2Context>> mappedContexts = new HashMap<>(); Iterator<Phase2Context> iterator = ASYNC_COMMIT_BUFFER.iterator(); //一次定時循環(huán)取出ASYNC_COMMIT_BUFFER中的所有待辦數(shù)據(jù) //以resourceId作為key分組待commit數(shù)據(jù)，resourceId是一個數(shù)據(jù)庫的連接url //在前面的日志中可以看到，目的是為了覆蓋應(yīng)用的多數(shù)據(jù)源創(chuàng)建 while ()) { Phase2Context commitContext = i(); List<Phase2Context> contextsGroupedByResourceId = ma); if (contextsGroupedByResourceId == null) { contextsGroupedByResourceId = new ArrayList<>(); ma, contextsGroupedByResourceId); } con(commitContext); i(); } for ;String, List<Phase2Context>> entry : ma()) { Connection conn = null; try { try { //根據(jù)resourceId獲取數(shù)據(jù)源以及連接 DataSourceProxy dataSourceProxy = Da().ge()); conn = da(); } catch (SQLException sqle) { LOGGER.warn("Failed to get connection for async committing on " + en(), sqle); continue; } List<Phase2Context> contextsGroupedByResourceId = en(); for (Phase2Context commitContext : contextsGroupedByResourceId) { try { //執(zhí)行undolog的處理，即刪除xid、branchId對應(yīng)的記錄 UndoLogManager.deleteUndoLog, commi, conn); } catch (Exception ex) { LOGGER.warn( "Failed to delete undo log [" + commi + "/" + commi + "]", 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ā)有兩種情況

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

public void rollback() throws TransactionException { //只有Launcher能發(fā)起這個rollback if (role == GlobalTran) { // Participant has no responsibility of committing if ()) { LOGGER.debug("Ignore Rollback(): just involved in global transaction [" + xid + "]"); } return; } if (xid == null) { throw new IllegalStateException(); } status = (xid); if () != null) { if ())) { Roo(); } } }

TC 匯總后向參與者發(fā)送 rollback 指令，RM 在AbstractRMHandler類的 doBranchRollback 方法中接收這個rollback 的通知。

protected void doBranchRollback(BranchRollbackRequest request, BranchRollbackResponse response) throws TransactionException { String xid = reque(); long branchId = reque(); String resourceId = reque(); String applicationData = reque(); LOGGER.info("Branch rolling back: " + xid + " " + branchId + " " + resourceId); BranchStatus status = getResourceManager().branchRollback(), xid, branchId, resourceId, applicationData); re(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 () == Tran) { return Branc; } else { return Branc; } } return Branc; }

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

1. 根據(jù) xid 和 branchId 查找 PhaseOne 階段提交的 undo_log；
2. 如果找到了就根據(jù) undo_log 中記錄的數(shù)據(jù)生成回放 sql 并執(zhí)行，即還原 PhaseOne 階段修改的數(shù)據(jù)；
3. 第 2 步處理完后，刪除該條 undo_log 數(shù)據(jù)；
4. 如果第 1 步?jīng)]有找到對應(yīng)的 undo_log，就插入一條狀態(tài)為GlobalFinished的 undo_log。出現(xiàn)沒找到的原因可能是 PhaseOne 階段的本地事務(wù)異常了，導(dǎo)致沒有正常寫入。 因為 xid 和 branchId 是唯一索引，所以第 4步的插入，可以防止 PhaseOne 階段恢復(fù)后的成功寫入，那么 PhaseOne 階段就會異常，這樣一來業(yè)務(wù)數(shù)據(jù)也就不會提交成功，數(shù)據(jù)達到了最終回滾了的效果。

總結(jié)

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

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

作者：中間件小哥