TCC 开源项目源码学习（一）

TCC 开源项目源码学习（一）学习TCC分布式事务的知识是使用了GIT上的一个开源项目，之前有简单的看过一些，有了一个大概的了解，但是随着时间的‘清洗’，又开始变得‘浑浊不清’了，这次索性把这份源码从头看了下，又把流程推演了好几次，觉得已经懂了，想把理解的东西通过博客写出来。在这个过程中，再次梳理一遍，加深理解，同时也可以发现一些理解偏差和错误的地方。GIT: https://github.com/1755728288…概要在分布式系统中，为了保证执行指令的正确性，引入了事务的概念。一般意义上，事务主要突出的是ACID，即原子性，一致性，隔离性和持久性。而在分布式事务中，最主要的原子性的保证，要保证一个业务操作在其分布式系统中的所有指令全部执行或不执行。因为各个指令的操作耗时以及顺序，所以原子性都对应了一定的时间窗口，比如单机系统下，这个时间窗口非常短，而且其原子性也由数据库事务来保证。而在分布式系统中，原子性就依赖于具体的应用设计了，主要的依据是业务上允许的时间窗口的长短。目前一般来说，若允许的时间窗口较短，就用TCC，若允许的时间窗口较长，则使用最终一致性。最终一致性通常使用消息机制来设计，其核心是消息的安全送达与消费。TCC可以说是一种设计模式，将传统单机系统中的大事务进行拆分，通过小事务来拼装，并协调整体的事务提交和回滚。按字面意思理解，TCC主要分Try，Confirm，Cancel三个阶段，Try预留资源，Confirm使用预留资源执行业务操作，Cancel则是释放资源。貌似简单，但是在实际的业务场景中，往往会困惑我们在这三个阶段分别要做什么，这个是业务设计的一部分内容了，在此不展开叙述。不过要注意的一点时，任何一个阶段的结果都是可见的，比如一个库存子服务的入库方法，在try阶段就直接加到库存上去了，回滚的时候发现刚加上的库存已经有买家下单准备出库了，那就GG了。代码结构先放个图，这是项目的组件，上面绿色区域是框架的子模块，下面黄色的是样例模块。不要被黄色区域吓到，TCC的代码不多，主要是tcc-transaction-core.组件名说明tcc-transaction-core【重要】核心代码，主要的切面和jobtcc-transaction-api注解和常量tcc-transaction-springspring使用的扩展tcc-transaction-dubbodubbo使用时的扩展tcc-transaction-servertcc事务管理控制台tcc-transaction-unit-test单元测试tcc-transaction-tutorial-sample订单场景示例工程，分dubbo和spring版本所以重点只有一个模块,就是tcc-transaction-core，代码量不多，主要分成下面的模块模块名说明interceptor【重要】2个切面，一个是根据事务状态进行tcc阶段方法的选择，一个是组织事务的参与者，用XID将各个参与者绑起来，以便事务的提交和回滚recover恢复在同步调用失败事务的定时处理repository事务对象持久DAO对象，提供了多种选择：缓存，文件系统，jdbc，zookeepercontext传递事务上下文的方法类，一般会在远程方法调用的切面中，将上下文加入到参数列表中common方法枚举和事务枚举serializer事务对象的序列化器，使用了kyto序列化工具support实现了工厂类，管理TCC的类的实例化utils工具类最重要的是interceptor，是主要业务逻辑所在。题外话，有人初看了一遍TCC，拿来和数据库事务来比较，会说‘TCC不就是业务补偿么，没什么难点啊’，貌似有道理，其实不然。还记得数据库事务的ACID吗？因为数据库提供了事务特性，ACID由数据库保证，应用只需要一个简单的@Transactional注解就都搞定了。而分布式事务，就把ACID的特性从数据库里面拿了出来，由应用程序来保证。当然ACID也有了和之前不一样的含义。特性数据库分布式事务A[Atomicity]所有数据库更新一起提交和回滚，数据库通过日志来实现通过协调各个事务的参与方，通过框架来处理异常C[Consistncy]通过数据库约束来实现，比如非空，唯一，外键等 I[Isolation]各个数据库实现方式不一样,主要分读已提交，读未提交等TCC通过try阶段锁定资源来实现隔离，即业务资源的隔离D[Durability]数据库实现数据库实现工作流程以项目中的例子来说明整个TCC的处理流程，这个例子是类似电商下单的场景，在支付的时候可以选择红包或本金，比如一个笔记本4000，可以选择使用3000的本金和1000的红包来支付，有三个服务：订单服务（order），本金服务（capital）和红包服务（redPacket）。主调方是Order，被调方Capital和RedPacket.下面是绞尽脑汁画的调用流程图，为了图的简便直观，省去了cancel的处理，只有try和confirm的调用（cancel的处理和confirm基本一致）。红线是try的调用，蓝线是confirm的调用。粗红线是开始，粗蓝线是结束。在try阶段的makePayment,方法调用了两个子事务的api方法，其他的方法的调用均是全局事务切面决定的。业务方法只要按TCC框架的要求实现即可，其他的交给TCC框架。主要代码分析我们从TCC事务注解开始，先介绍几个基本的类。/属性主要是事务传播属性，提交方法，回滚方法，上下文传递类，是否异步提交，是否异步回滚propagation属性比较重要，值有required，support，mandatory和requireNEW，具体的含义和spring的事务传播类似。/public @interface Compensable { public Propagation propagation() default Propagation.REQUIRED; public String confirmMethod() default “”; public String cancelMethod() default “”; public Class<? extends TransactionContextEditor> transactionContextEditor() default DefaultTransactionContextEditor.class; public boolean asyncConfirm() default false; public boolean asyncCancel() default false;/事务状态，在TCC的不通阶段进行事务状态的变更。/public enum TransactionStatus { TRYING(1), CONFIRMING(2), CANCELLING(3);/** ROOT:事务启动方法，比如例子中order的makePayment方法 PROVIDER:事务协同方法，比如例子中redpacket的record方法 Normal:普通方法，比如API里面的record方法**/public enum MethodType { ROOT, PROVIDER, NORMAL;}/** ROOT:主事务，一般MethodType为ROOT的启动 BRANCH:分支事务**/public enum TransactionType { ROOT(1), BRANCH(2);下面我们看一下事务持久的内容，能更好的帮助理解。每个服务都会有一张事务表，会记录所有的ROOT和BRANCH事务，在事务完成之后，会自动删除。TRANSACTION_IDDOMAINGLOBAL_TX_IDBRANCH_QUALIFIERCONTENTSTATUSTRANSACTION_TYPERETRIED_COUNTVERSION2ORDER事务编号001分支标识A事务对象A210112CAPITAL事务编号001分支标识B事务对象B220112REDPACKET事务编号001分支标识C事务对象C22011domain区分了服务global_tx_id 串起了所有的事务transaction_type 1为主事务，2为分支事务content 用于恢复事务，主事务的content中包含了参与者代码部分的最后是两个切面和恢复的job，不进行特别的解释了，将我的理解放在代码的注释里面。 //CompensableTransactionInterceptor.java public Object interceptCompensableMethod(ProceedingJoinPoint pjp) throws Throwable { //获取所执行事务方法的设置属性 Method method = CompensableMethodUtils.getCompensableMethod(pjp); Compensable compensable = method.getAnnotation(Compensable.class); Propagation propagation = compensable.propagation(); TransactionContext transactionContext = FactoryBuilder.factoryOf(compensable.transactionContextEditor()).getInstance().get(pjp.getTarget(), method, pjp.getArgs()); boolean asyncConfirm = compensable.asyncConfirm(); boolean asyncCancel = compensable.asyncCancel(); //当前是否存在事务， boolean isTransactionActive = transactionManager.isTransactionActive(); /** * 判断方法类型 * MethodType一共有三种，Root，Provider,Normal. 主要通过propagation和isTransactionActive来确定 */ MethodType methodType = CompensableMethodUtils.calculateMethodType(propagation, isTransactionActive, transactionContext); switch (methodType) { case ROOT: //ROOT：（1）propagation为require_new (2)propataion为require，且之前没有有事务存在 //主事务处理方法 return rootMethodProceed(pjp, asyncConfirm, asyncCancel); case PROVIDER: //PROVIDER:(1) propation为require或者mandatory,且之前有事务 //从事务处理方法 return providerMethodProceed(pjp, transactionContext, asyncConfirm, asyncCancel); default: //普通方法执行 return pjp.proceed(); } } private Object rootMethodProceed(ProceedingJoinPoint pjp, boolean asyncConfirm, boolean asyncCancel) throws Throwable { Object returnValue = null; Transaction transaction = null; try { //开启新的主事务，使用新生成的xid，状态为trying，事务类型为ROOT transaction = transactionManager.begin(); try { returnValue = pjp.proceed(); } catch (Throwable tryingException) { if (!isDelayCancelException(tryingException)) { logger.warn(String.format(“compensable transaction trying failed. transaction content:%s”, JSON.toJSONString(transaction)), tryingException); //异常回滚，将事务状态更新为CANCELLING transactionManager.rollback(asyncCancel); } throw tryingException; } //执行confirm方法，将事务状态更新为CONFIRMING。如果是异步，则TM会使用线程池异步执行，否则直接调用，会协调所有的参与者进行提交。并将事务记录删除 transactionManager.commit(asyncConfirm); } finally { //清理事务数据 transactionManager.cleanAfterCompletion(transaction); } return returnValue; } private Object providerMethodProceed(ProceedingJoinPoint pjp, TransactionContext transactionContext, boolean asyncConfirm, boolean asyncCancel) throws Throwable { Transaction transaction = null; try { switch (TransactionStatus.valueOf(transactionContext.getStatus())) { case TRYING: //开启一个子事务，并调用TCC的try方法 transaction = transactionManager.propagationNewBegin(transactionContext); return pjp.proceed(); case CONFIRMING: //获取子事务TRY阶段的事务，并调用TCC的commit方法 try { transaction = transactionManager.propagationExistBegin(transactionContext); transactionManager.commit(asyncConfirm); } catch (NoExistedTransactionException excepton) { //the transaction has been commit,ignore it. } break; case CANCELLING: //获取子事务保存的事务数据，执行cancel方法 try { transaction = transactionManager.propagationExistBegin(transactionContext); transactionManager.rollback(asyncCancel); } catch (NoExistedTransactionException exception) { //the transaction has been rollback,ignore it. } break; } } finally { //清理事务数据 transactionManager.cleanAfterCompletion(transaction); } Method method = ((MethodSignature) (pjp.getSignature())).getMethod(); return ReflectionUtils.getNullValue(method.getReturnType()); } private boolean isDelayCancelException(Throwable throwable) { if (delayCancelExceptions != null) { for (Class delayCancelException : delayCancelExceptions) { Throwable rootCause = ExceptionUtils.getRootCause(throwable); if (delayCancelException.isAssignableFrom(throwable.getClass()) || (rootCause != null && delayCancelException.isAssignableFrom(rootCause.getClass()))) { return true; } } } return false; }总结代码分析的比较少，这份代码还是有很多值得称道的地方，工厂类，缓存，模板方法等设计模式的使用，下次可以从设计模式的角度来进行分析。除了TCC，最近项目中还涉及了安全消息，等弄清楚了再来一发。