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，代码量不多，主要分成下面的模块

最重要的是interceptor，是主要业务逻辑所在。

题外话，有人初看了一遍TCC，拿来和数据库事务来比较，会说‘TCC不就是业务补偿么，没什么难点啊’，貌似有道理，其实不然。还记得数据库事务的ACID吗？因为数据库提供了事务特性，ACID由数据库保证，应用只需要一个简单的@Transactional注解就都搞定了。而分布式事务，就把ACID的特性从数据库里面拿了出来，由应用程序来保证。当然ACID也有了和之前不一样的含义。

以项目中的例子来说明整个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 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事务，在事务完成之后，会自动删除。

domain区分了服务
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，最近项目中还涉及了安全消息，等弄清楚了再来一发。