springcloud结合bytetcc实现数据强一致性原理解析

Shun

这篇文章主要介绍了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实现数据强一致性原理剖析的文章就介绍到这了,更多相关springcloud实现数据强一致性内容请搜索CodeAE代码之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持CodeAE代码之家！
原文链接：https://segmentfault.com/a/1190000039311550