如何在springcloud中使用bytetcc實現數據的強一致性

今天就跟大家聊聊有關如何在springcloud中使用bytetcc實現數據的強一致性，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

1 使用背景和約束

公司使用的是springcloud，面臨分布式事務的場景的時候，可以使用對springcloud支持比較好的byte-tcc框架，git目前2600星，使用起來也非常方便，原理也很清晰，非常適合學習。 https://github.com/liuyangmin... ，結合cloud有幾個重點約束如下，

(1)一個業務接口，需要有三種實現類，分別是try，confirm，cancel，符合tcc的思路。實現方法必須加Transactional，且propagation必須是Required, RequiresNew, Mandatory中的一種。

(2)服務提供方Controller必須添加@Compensable注解，不允許對Feign/Ribbon/RestTemplate等HTTP請求自行進行封裝。想想看為什么？

(3)在每個參與tcc事務的數據庫中創建bytejta表。

配置上也非常簡單，@Import(SpringCloudConfiguration.class)，服務提供方，try上面加上

@Service("accountService")
 @Compensable(
 interfaceClass = IAccountService.class 
 , confirmableKey = "accountServiceConfirm"
 , cancellableKey = "accountServiceCancel"
 )

confirm和cancel對應的

@Service("accountServiceConfirm")
@Service("accountServiceCancel")，

try confirm cancel 對應業務可以理解為 凍結庫存/真正扣減庫存/恢復庫存，或者凍結優惠券/核銷優惠券/恢復優惠券這種實際業務場景。

0.5以后datasource自動使用LocalXADataSource，之前需要手動配置

@Bean(name = "dataSource")
 public DataSource getDataSource() {
  LocalXADataSource dataSource = new LocalXADataSource();
  dataSource.setDataSource(this.invokeGetDataSource());
  return dataSource;
 }

所以，從配置上看，bytetcc和springcloud結合，一個應該是通過引入自己的SpringCloudConfiguration封裝了feign/ribbon/hystrix調用，一個是提供了自己的datasource管理事務。有了自己的datasource，定制自己的transactionManager,就可以在事務前后動手腳，2pc/3pc對事務的管理，體現在控制不同數據庫連接上，微服務為主的tcc，對事務的管理體現在控制各個服務的調用上。

2 業務場景思考

bytetcc的tcc，其實try和cancel是配套的，考慮下業務場景：

（1）如果a服務try成功了，b服務try失敗，則a服務需要回滾，調用a的cancel。這是普遍流程。

（2）如果a 和 b都try成功了，然后a confirm成功，b的confirm失敗，是沒有cancel和confirm配對的。b的confirm會不斷調用直到成功為止。

因為bytetcc的設計思路是，通過try做好準備工作（如鎖定資源），try如果能成功，那么邏輯上confirm一定要成功。如果confirm不成功，則可能是外部環境問題，如網絡問題等，那么環境恢復了遲早應該成功confirm。基于這個思想，try和cancel是互逆的，confirm一旦執行就不可逆。

如果要設計confirm也可逆的，那要么cancel里判斷是該回滾try還是回滾try+confirm，不清晰且實現很麻煩，或者做成“tccc”加一個對應confirm的cancel，由事務管理器統一判斷調用幾個cancel，引入太多不確定。所以業務上可以直接這么設計：try成功，那么confirm是一定要成功的。

3 核心組件SpringCloudConfiguration原理分析

第一步就import的SpringCloudConfiguration是重點，通過它的各種自動裝配基本可以實現bytetcc的全邏輯。要想擴展spring，那就得擴展各種BeanFactoryPostProcessor,SpringCloudConfiguration本身就是個BeanFactoryPostProcessor，但是postProcessBeanFactory沒干啥，應該是引入了各種其他processor進行擴展。如何擴展面臨幾個問題

（1） 如何識別核心的@Compensable注解？

在byte-tcc的一堆resource文件里，配置了各種bean。既然都Springcloud了為啥還用這種文件bean，可能是兼容cloud之外的場景，增強通用性。在bytetcc-supports-tcc.xml里，有個bean <bean class="org.bytesoft.bytetcc.supports.spring.CompensableAnnotationConfigValidator" />，通過關鍵代碼

clazz.getAnnotation(Compensable.class);掃描得到所有注解了Compensable的類，同時解析出來cancel，confirm對應的類。同時校驗一下有沒有加transactional注解。后面很多類似的processor都是用這種注解識別需要的bean

（2）如何改造事務管理器，使之適應分布式微服務環境？

在bytetcc-supports-tcc.xml中，定義了改造過的transactionManager,

<bean id="transactionManager" class="org.bytesoft.bytetcc.TransactionManagerImpl" />

這里面重寫了begin，commit那一套，結合了tcc專用組件CompensableManager，重新包裝了事務操作。

同時，通過SpringCloudConfiguration配置的CompensableHandlerInterceptor，達到transactionInterceptor的效果。

（3）事務如何恢復？

在bytetcc-supports-logger-primary.xml中，有個ResourceAdapterImpl，這個是啟動bytetcc后臺線程的地方，看bean配置就知道，把兩個bean compensableWork和bytetccCleanupWork注入到ResourceAdapterImpl中統一管理，字面意思上看是補償任務和數據清理任務。這里的compensableWork就是補償和數據恢復專用的job。

<bean id="compensableResourceAdapter" class="org.bytesoft.transaction.adapter.ResourceAdapterImpl">
  <property name="workList">
   <list>
    <ref bean="compensableWork" />
    <ref bean="bytetccCleanupWork" />
   </list>
  </property>
 </bean>

追進去看，通過List<Work> workList收集起來work，然后統一用mananger進行start，看work對象本身就繼承了Runnable，所以這里是開啟了后臺線程，進行事務恢復等操作。

（4）具體的請求接口，怎么擴展？

import的SpringCloudConfiguration的代理組件，通過條件注解和配置，根據是否開啟hystrix和是否引入HystrixFeign的類，注入針對feign或hystrix的CompensableFeignBeanPostProcessor，如下

@org.springframework.context.annotation.Bean
 @ConditionalOnProperty(name = "feign.hystrix.enabled", havingValue = "false", matchIfMissing = true)
 public CompensableFeignBeanPostProcessor feignPostProcessor() {
  return new CompensableFeignBeanPostProcessor();
 }

 @org.springframework.context.annotation.Bean
 @ConditionalOnProperty(name = "feign.hystrix.enabled")
 @ConditionalOnClass(feign.hystrix.HystrixFeign.class)
 public CompensableHystrixBeanPostProcessor hystrixPostProcessor() {
  return new CompensableHystrixBeanPostProcessor();
 }

以CompensableFeignBeanPostProcessor為例，明顯這就是為了對feign接口進行代理的PostProcessor，在postProcessAfterInitialization中，果然通過createProxiedObject（），創建了CompensableFeignHandler的代理類，對springcloud自己的FeignInvocationHandler進行了又一次代理。這樣所有@FeignClient的接口都會經過這個handler

同理如果是hystrix的代理，CompensableHystrixBeanPostProcessor會創建CompensableHystrixFeignHandler代理，代替原來的CompensableHystrixInvocationHandler。

通過這兩種代理，可以對原生的feign/hystrix組件繼續代理，在請求前后做些事情。feign/hystrix其實本身也是結合了ribbon的代理，所以很多spring的擴展就是一層層的代理疊加，為我們擴展組件提供了一種思路。那么bytetcc的核心流程肯定就蘊含在這個請求代理中。

（5）如何控制請求哪一種方法？

bytetcc-supports-springcloud-primary.xml中，有個controller，CompensableCoordinatorController，可以看到里面封裝了幾種方法，prepare，commit，rollback，額外還有recover，forget，名字上可以看出是恢復，刪除事務。結合第四點，原來的feign調用被代理一層，請求的真實url應該被改過，改成了請求這一個controller的方法，通過這個controller再決定后面做什么。

@RequestMapping(value = "/org/bytesoft/bytetcc/prepare/{xid}", method = RequestMethod.POST)
@RequestMapping(value = "/org/bytesoft/bytetcc/commit/{xid}/{opc}", method = RequestMethod.POST)
@RequestMapping(value = "/org/bytesoft/bytetcc/rollback/{xid}", method = RequestMethod.POST)
@RequestMapping(value = "/org/bytesoft/bytetcc/recover/{flag}", method = RequestMethod.GET)
@RequestMapping(value = "/org/bytesoft/bytetcc/forget/{xid}", method = RequestMethod.POST)

綜上所述，通過新的分布式事務管理器的封裝，feign/hystrix請求的代理，controller的控制，后臺補償任務的執行，基本上可以實現強一致性的分布式事務。

4 事務啟動-try過程

（1）產生事務

接到用戶一個請求時， CompensableHandlerInterceptor會先攔截，這是用戶剛發的請求，在這里沒找到事務信息什么都不干就返回true了，如果是被調用者，無論是try/confirm/cancel，都會有個事務上下文信息，解析出事務。

CompensableMethodInterceptor->excute(),獲得了@transactional和@composable注解，包括 confirm/cancel方法信息，封裝到invocation，保存本次調用的一些信息。

transactionInterceptor，調用bytetcc提供的TransactionManagerImpl,提供了新的begin，啟動tcc事務。注意這里，如果沒有事務上下文，沒有compensable注解，那就走一般的begin，就是一般的本地事務。

有了compensable注解，begin就是上面說到的CompensableManager的compensableBegin方法，初始化了事務上下文環境transanctionContext，還生成了個事務id-xid。

（2）方法執行，CompensableFeignInterceptor,把上面生成的事物上下文環境transactionContext，通過序列化生成字符串，放入request的header中，這樣發起事務無論調用什么服務，事務的上下文信息就保存在request的header里了。

后面根據feign的代理CompensableFeignHandler發出請求，return this.delegate.invoke(proxy, method, args) delegate就是feign，本質上也是使用原生的feign發請求。

既然本質也是feign調用，思考一下為啥還費事代理一次？事務上下文環境在interceptor里面已經設置到request里了，還代理干啥？

往后看，我認為關鍵在這里，

beanRegistry.setLoadBalancerInterceptor(new CompensableLoadBalancerInterceptor(this.statefully)

大家知道，feign通過ribbon組件進行的復雜均衡，即chooseInstance,選擇請求往哪個實例上發，如果還是輪訓或隨機，第一次try請求發到某實例，第二次confirm/cancel發到其他實例，別的實例上沒有try帶來的的事務信息，會非常不方便，也不知道try到底什么情況，

所以這里要多次請求粘滯到一個實例上。所以bytetcc實現了ribbon算法CompensableLoadBalancerRuleImpl，不支持自定義rule

（3）服務接收方接受，首先經過CompensableHandlerInterceptor的preHandle,解析出事務上下文transactionContext,封裝成TransactionRequestImpl，并在response里加上一些header信息。這在發起者那里因為沒有header的上下文，所以在（1）是什么都不做的

再到 CompensableMethodInterceptor, 解析方法。

再到 TransactionManager，即bytetcc的TransactionManagerImpl，到這里的begin，由于剛接到請求的時候，這時候已經有事務了，所以調用的是一般的本地事務compensableManager.begin()，最后開啟一個本地事務，然后執行本地方法，執行commit。

下圖可以簡單介紹這個過程。

5 try調用失敗，cancel過程

（1）如果有任意一個try失敗，那么要把已經成功的try給回滾掉，spring通用的transactionInterceptor的處理過程，invokeWithinTransaction方法，如果有異常，catch住執行

completeTransactionAfterThrowing()，然后到transactionManagerImpl的rollback,繼續到CompensableManager的collback

（2）CompensableTransanctionImpl中，fireRemoteParticipantCancel是真正的rollback，里面維護了一個resourcelist，按順序記錄了其他各個服務在try的時候調用的服務，在這里循環這個list調用SpringCloudCoordinator,拼接cancel地址，帶著事務id發送請求過去。

（3）接收方，CompensableCoordinatorController的rollback，核心是從CompensableTransactionImpl到SpringContainerContextImpl 的 cancel，得到請求的controller的對應的cancel方法，封裝到cancellableKey，然后拿到處理cancel的真實的bean，

Object instance = this.applicationContext.getBean(cancellableKey);
this.cancelComplicated(method, instance, args);

進而執行cancel對應的bean的方法。整個過程可以如下概括

6 try成功，confirm過程

同理，transactionManagerImpl的commit，最終到達CompensableTransactionImp進行fireCommit，先提交本地事務，然后fireRemoteParticipantConfirm，和cancel一模一樣，讀取resourceList,遍歷list發送請求到各個服務端。

各個服務方CompensableCoordinatorController的commit，拿到confirmablekey，找到confirm的bean進行confirm。

7 “compensable”的補償

（1）如果cancel，commit有失敗（失敗包含runtimeexception和自定義的一些異常），那么如何進行補償，上面提到的一開始就啟動的CompensableWork線程的run里面，其實有個while（true），每隔100秒循環一次，調用組件TransactionRecovery（看名字就知道恢復事務用的）的timingRecover，就是定時回復，會調用到CompensableTransactionImpl的recoveryRollback/recoveryCommit，還是SpringCloudCoordinator發送的請求。

（2）如果出現宕機，重啟后也是通過CompensableWork線程的run，第一步是init，嘗試恢復現有的事務。

a 如果try沒有執行完就down機，恢復時把已執行的try給cancel掉。因為事務一般是業務請求觸發的，down機就請求失敗了，沒必要重啟后還恢復剛才的請求。
b 如果是confirm/cancel有沒成功的，會一直定時進行confirm/cancel。

看完上述內容，你們對如何在springcloud中使用bytetcc實現數據的強一致性有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業資訊頻道，感謝大家的支持。