TCC-transaction学习

2018-10-15

1. TCC事务简介

TCC事务 即：Try-Confirm-Cancel 。它是基于业务层面的事务定义，把事务运行过程分成 Try、Confirm / Cancel 两个阶段。在每个阶段的逻辑由业务代码控制。每一个初步操作，最终都会被确认或取消。因此，针对一个具体的业务服务，TCC事务机制需要业务系统提供三段业务逻辑：初步操作Try、确认操作Confirm、取消操作Cancel。
Try 从执行阶段来看，与传统事务机制中业务逻辑相同。但从业务角度来看，却不一样。TCC机制中的Try仅是一个初步操作，它和后续的确认一起才能真正构成一个完整的业务逻辑。TCC机制将传统事务机制中的业务逻辑一分为二，拆分后保留的部分即为初步操作（Try）；而分离出的部yfyf分即为确认操作（Confirm），被延迟到事务提交阶段执行。
- 完成所有业务检查( 一致性 )
- 预留必须业务资源( 准隔离性 )
Confirm 是对 Try 操作的一个补充。当TCC事务管理器决定commit全局事务时，就会逐个执行Try操作指定的Confirm操作，将Try未完成的事项最终完成。
Cancel 是对Try操作的一个回撤。当TCC事务管理器决定rollback全局事务时，就会逐个执行Try操作指定的Cancel操作，将Try操作已完成的事项全部撤回。

整体流程如图

TCC优缺点
- 优点：让应用自己定义数据库操作的粒度，使得降低锁冲突、提高吞吐量成为可能。
- 不足：
  - 对应用的侵入性强。业务逻辑的每个分支都需要实现try、confirm、cancel三个操作，应用侵入性较强，改造成本高。
  - 实现难度较大。需要按照网络状态、系统故障等不同的失败原因实现不同的回滚策略。为了满足一致性的要求，confirm和cancel接口必须实现幂等。

2. TCC-transaction 框架介绍

源码地址：https://github.com/changmingxie/tcc-transaction

TCC-transaction是开源的TCC补偿性分布式事务框架，使用Java实现，不和底层使用的rpc框架耦合，可以使用doubbo,thrift,web service,http等接口。事务管理器日志持久化支持多种方式，如mysql，zookeeper等。

1. 接入准备

引用tcc-transaction的Maven依赖

<dependency>
        <groupId>org.mengyun</groupId>
        <artifactId>tcc-transaction-spring</artifactId>
        <version>${project.version}</version>
</dependency>

加载tcc-transaction.xml配置

启动应用时，需要将tcc-transaction-spring jar中的tcc-transaction.xml加入到classpath中。如在web.xml中配置：

<context-param>
    <param-name>contextConfigLocation</param-name>
    <param-value>classpath:tcc-transaction.xml
    </param-value>
</context-param>

设置TransactionRepository

选择持久化方式。可以选择FileSystemTransactionRepository、SpringJdbcTransactionRepository、RedisTransactionRepository或ZooKeeperTransactionRepository。

如SpringJdbcTransactionRepository

<bean id="transactionRepository"
      class="org.mengyun.tcctransaction.spring.repository.SpringJdbcTransactionRepository">
    <property name="dataSource" ref="dataSource"/>
</bean>

<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource"
      destroy-method="close">
    <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
    <property name="url" value="jdbc:mysql://127.0.0.1:3306/test"/>
    <property name="username" value="root"/>
    <property name="password" value=""/>
</bean>

新建TCC表
- 执行 tcc-transaction-tutorial-sample/src/dbscripts/下，create_db_cap.sql，create_db_ord.sql等四个文件

2. 本地部署调试

在本机中需要有Mysql环境，如果调试dubbo需要按照zookeeper。

可以参照https://github.com/lingo0/tcc-transaction中分支normalDebug分支。

修改所有tccjdbc.properties文件中的jdbc的配置，
新建dubbo-capital，dubbo-redpacket，dubbo-order三个tomcat启动环境。

3. TCC-transaction 事务测试

异常情况1：try失败

可以参照https://github.com/lingo0/tcc-transaction中分支error-test-1分支。

具体异常描述：红包账户冻结成功(try)、资金账户冻结成功(try)，订单操作异常(try)

在makePayment中添加抛出异常，使之try失败。

此时order主业务流程try异常，支付失败，事务需要回滚，TCC事务协调器执行cancel操作，会将红包账户冻结金额、资金账户冻结金额全部回滚。

类似的：红包账户try异常、资金账户try异常，远程dubbo接口抛出异常，在主服务的TCC事务协调器获取到异常，由事务恢复任务处理回滚。

如，在远程方法中造成异常，抛出npe
异常情况2：confirm异常

可以参照https://github.com/lingo0/tcc-transaction中分支error-test-2分支。

具体异常描述：订单处理成功(confirm)，资金账户扣减成功(confirm)，但红包账户扣减失败(confirm)。

如图：添加如下异常。

这时候三个try操作均成功，对于业务来说成功。

TCC事务协调器会执行confirm操作。当红包的confirm接口异常时，订单confirm操作未执行成功，系统会不断重试调用订单的confirm操作，直到红包的confirm成功。

如果达到最大重试次数，或者时间，则需要人工处理。类似我们造异常的情况。

类似的，如果try异常，cancel也异常，那么事务协调器会不断重试，直达cancel成功。

总结： try 操作成功，进入 confirm 操作，只要 confirm 处理失败（不管是协调者挂了，还是参与者处理失败或超时），系统通过不断重试直到处理成功。进入 cancel 操作也是一样，只要 cancel 处理失败，系统通过不断重试直到处理成功。

4. TCC-transaction 事务实现

主要代码位置 tcc-transaction-core

1. 主要的几个类

实体类

@Compensable TCC事务方法注解
Transaction 事务实体
Participant 事务参与者
TransactionContext 事务上下文
Propagation 事务传播级别

和spring事务传播级别类似

功能相关类

CompensableTransactionAspect和CompensableTransactionInterceptor 事务执行器

处理事务执行
ResourceCoordinatorAspect 和 ResourceCoordinatorInterceptor 事务资源协调器

处理事务过程中，添加事务参与者
TransactionManager 事务管理器。

提供事务的获取、发起、提交、回滚，参与者的新增等等方法。
TransactionRepository 事务持久化

2.事务执行流程图

第一个切面CompensableTransactionAspect拦截后，执行事务操作。实现功能有：
- 在Try阶段，执行事务发起和传播
- 在 Confirm / Cancel 阶段，对事务提交或回滚
第二个切面ResourceCoordinatorAspect拦截后，事务协调器，添加事务上下文和添加参与者到事务中。

3. 走读代码

5. TCC 事务恢复

事务信息被持久化到外部的存储器中。事务存储是事务恢复的基础。通过读取外部存储器中的异常事务，定时任务会按照一定频率对事务进行重试，直到事务完成或超过最大重试次数。

1. 主要的类

RecoverConfig 事务恢复配置接口
TransactionRecovery 事务恢复逻辑
RecoverScheduledJob 事务恢复定时任务

我们主要看TransactionRecovery

当单个事务超过最大重试次数时，不再重试，只打印异常，此时需要人工介入解决。可以接入报警

当分支事务超过最大可重试时间时，不再重试。

public class TransactionRecovery {

    static final Logger logger = Logger.getLogger(TransactionRecovery.class.getSimpleName());

    private TransactionConfigurator transactionConfigurator;

    /**
     * 启动恢复事务逻辑
     */
    public void startRecover() {

        // 加载异常事务集合
        List<Transaction> transactions = loadErrorTransactions();

        // 恢复异常事务集合
        recoverErrorTransactions(transactions);
    }

    // 加载异常事务集合
    private List<Transaction> loadErrorTransactions() {


        long currentTimeInMillis = Calendar.getInstance().getTimeInMillis();

        TransactionRepository transactionRepository = transactionConfigurator.getTransactionRepository();
        RecoverConfig recoverConfig = transactionConfigurator.getRecoverConfig();

        // 当前时间超过 - 事务恢复间隔 RecoverConfig#getRecoverDuration()
        return transactionRepository.findAllUnmodifiedSince(new Date(currentTimeInMillis - recoverConfig.getRecoverDuration() * 1000));
    }

    // 恢复异常事务集合
    private void recoverErrorTransactions(List<Transaction> transactions) {


        for (Transaction transaction : transactions) {

            // 超过最大重试次数
            if (transaction.getRetriedCount() > transactionConfigurator.getRecoverConfig().getMaxRetryCount()) {

                // 当单个事务超过最大重试次数时，不再重试，只打印异常，此时需要人工介入解决。
                logger.error(String.format("recover failed with max retry count,will not try again. txid:%s, status:%s,retried count:%d,transaction content:%s", transaction.getXid(), transaction.getStatus().getId(), transaction.getRetriedCount(), JSON.toJSONString(transaction)));
                continue;
            }

            // 分支事务超过最大可重试时间
            if (transaction.getTransactionType().equals(TransactionType.BRANCH)
                    && (transaction.getCreateTime().getTime() +
                    transactionConfigurator.getRecoverConfig().getMaxRetryCount() *
                            transactionConfigurator.getRecoverConfig().getRecoverDuration() * 1000
                    > System.currentTimeMillis())) {
                continue;
            }

            // Confirm / Cancel
            try {
                // 增加重试次数
                transaction.addRetriedCount();

                // 如果事务状态是 confirm ,则重试commit
                if (transaction.getStatus().equals(TransactionStatus.CONFIRMING)) {

                    transaction.changeStatus(TransactionStatus.CONFIRMING);
                    transactionConfigurator.getTransactionRepository().update(transaction);
                    transaction.commit();
                    transactionConfigurator.getTransactionRepository().delete(transaction);

                } else if (transaction.getStatus().equals(TransactionStatus.CANCELLING)
                           // 这里加判断的事务类型为根事务，用于处理延迟回滚异常的事务的回滚。
                        || transaction.getTransactionType().equals(TransactionType.ROOT)) {

                    transaction.changeStatus(TransactionStatus.CANCELLING);
                    transactionConfigurator.getTransactionRepository().update(transaction);
                    transaction.rollback();
                    transactionConfigurator.getTransactionRepository().delete(transaction);
                }

            } catch (Throwable throwable) {

                if (throwable instanceof OptimisticLockException
                        || ExceptionUtils.getRootCause(throwable) instanceof OptimisticLockException) {
                    logger.warn(String.format("optimisticLockException happened while recover. txid:%s, status:%s,retried count:%d,transaction content:%s", transaction.getXid(), transaction.getStatus().getId(), transaction.getRetriedCount(), JSON.toJSONString(transaction)), throwable);
                } else {
                    logger.error(String.format("recover failed, txid:%s, status:%s,retried count:%d,transaction content:%s", transaction.getXid(), transaction.getStatus().getId(), transaction.getRetriedCount(), JSON.toJSONString(transaction)), throwable);
                }
            }
        }
    }

    public void setTransactionConfigurator(TransactionConfigurator transactionConfigurator) {
        this.transactionConfigurator = transactionConfigurator;
    }
}

6. dubbo 支持

TCC-Transaction 通过 Dubbo 隐式传参的功能，避免自己对业务代码的入侵。

通过 Dubbo Proxy 的机制，实现 @Compensable 属性自动生成，优点就是增加开发体验，也避免出错。

dubbo接口上只要加@Compensable并不需要其他参数，也不需要显示传递事务上下文。

1. 实现原理：

通过 Dubbo Proxy 的机制，重写 Javassist 代理生成方式。

修改配置： <dubbo:provider proxy="tccJavassist"/> 让dubbo使用org.mengyun.tcctransaction.dubbo.proxy.javassist.TccJavassistProxyFactory 类生成代理类。

在项目启动时，调用 TccJavassistProxyFactory#getProxy(...) 方法，生成 Dubbo Service 调用代理类，最终将接口生成可调用的类。

2. 生成结果

原来的接口类

public interface RedPacketTradeOrderService {

    @Compensable
    String record(RedPacketTradeOrderDto tradeOrderDto);
}

生成的 Dubbo Service 调用类

public class TccProxy3 extends TccProxy implements TccClassGenerator.DC {  
    public Object newInstance(InvocationHandler paramInvocationHandler) {    
        return new proxy3(paramInvocationHandler);  }
}

生成的 Dubbo Service 调用 Proxy 如下：

public class proxy3 implements TccClassGenerator.DC, RedPacketTradeOrderService, EchoService {
    
    public static Method[] methods;
    private InvocationHandler handler;

    public proxy3() {}

    public proxy3(InvocationHandler paramInvocationHandler) {
        this.handler = paramInvocationHandler;
    }

    @Compensable(propagation = Propagation.SUPPORTS, confirmMethod = "record", cancelMethod = "record", transactionContextEditor = DubboTransactionContextEditor.class)
    public String record(RedPacketTradeOrderDto paramRedPacketTradeOrderDto) {
        Object[] arrayOfObject = new Object[1];
        arrayOfObject[0] = paramRedPacketTradeOrderDto;
        Object localObject = this.handler.invoke(this, methods[0], arrayOfObject);
        return (String) localObject;
    }

    public Object $echo(Object paramObject) {
        Object[] arrayOfObject = new Object[1];
        arrayOfObject[0] = paramObject;
        Object localObject = this.handler.invoke(this, methods[1], arrayOfObject);
        return (Object) localObject;
    }
}

3. 自动生成`@Compensable` 属性关键代码

public Class<?> toClass() {
        // mCtc 非空时，进行释放；下面会进行创建  mCtc --- 动态生成的类
        if (mCtc != null)
            mCtc.detach();
        long id = CLASS_NAME_COUNTER.getAndIncrement();
        try {
            CtClass ctcs = mSuperClass == null ? null : mPool.get(mSuperClass);
            if (mClassName == null)
                mClassName = (mSuperClass == null || javassist.Modifier.isPublic(ctcs.getModifiers())
                        ? TccClassGenerator.class.getName() : mSuperClass + "$sc") + id;

            // 创建mCtc
            mCtc = mPool.makeClass(mClassName);
            if (mSuperClass != null)
                mCtc.setSuperclass(ctcs); // 继承类
            mCtc.addInterface(mPool.get(DC.class.getName())); // add dynamic class tag.
            if (mInterfaces != null) // 实现接口集合
                for (String cl : mInterfaces) mCtc.addInterface(mPool.get(cl));
            if (mFields != null)    // 属性集合
                for (String code : mFields) mCtc.addField(CtField.make(code, mCtc));
            if (mMethods != null) { // 方法集合
                for (String code : mMethods) {
                    if (code.charAt(0) == ':')
                        mCtc.addMethod(CtNewMethod.copy(getCtMethod(mCopyMethods.get(code.substring(1))), code.substring(1, code.indexOf('(')), mCtc, null));
                    else {

                        CtMethod ctMethod = CtNewMethod.make(code, mCtc);

                        if (compensableMethods.contains(code)) {

                            // 设置 @Compensable 属性
                            ConstPool constpool = mCtc.getClassFile().getConstPool();
                            AnnotationsAttribute attr = new AnnotationsAttribute(constpool, AnnotationsAttribute.visibleTag);
                            Annotation annot = new Annotation("org.mengyun.tcctransaction.api.Compensable", constpool);
                            EnumMemberValue enumMemberValue = new EnumMemberValue(constpool);
                            enumMemberValue.setType("org.mengyun.tcctransaction.api.Propagation");
                            enumMemberValue.setValue("SUPPORTS");
                            annot.addMemberValue("propagation", enumMemberValue);
                            annot.addMemberValue("confirmMethod", new StringMemberValue(ctMethod.getName(), constpool));
                            annot.addMemberValue("cancelMethod", new StringMemberValue(ctMethod.getName(), constpool));

                            ClassMemberValue classMemberValue = new ClassMemberValue("org.mengyun.tcctransaction.dubbo.context.DubboTransactionContextEditor", constpool);
                            annot.addMemberValue("transactionContextEditor", classMemberValue);

                            attr.addAnnotation(annot);
                            ctMethod.getMethodInfo().addAttribute(attr);
                        }

                        mCtc.addMethod(ctMethod);
                    }
                }
            }
            if (mDefaultConstructor)      // 空参数构造方法
                mCtc.addConstructor(CtNewConstructor.defaultConstructor(mCtc));

            if (mConstructors != null) {  // 带参数构造方法
                for (String code : mConstructors) {
                    if (code.charAt(0) == ':') {
                        mCtc.addConstructor(CtNewConstructor.copy(getCtConstructor(mCopyConstructors.get(code.substring(1))), mCtc, null));
                    } else {
                        String[] sn = mCtc.getSimpleName().split("\\$+"); // inner class name include $.
                        mCtc.addConstructor(CtNewConstructor.make(code.replaceFirst(SIMPLE_NAME_TAG, sn[sn.length - 1]), mCtc));
                    }
                }
            }
            return mCtc.toClass();
        } catch (RuntimeException e) {
            throw e;
        } catch (NotFoundException e) {
            throw new RuntimeException(e.getMessage(), e);
        } catch (CannotCompileException e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

4. 隐式传递事务上下文给远程dubbo服务

见 DubboTransactionContextEditor中，存放在RpcContext.getContext().setAttachment(TransactionContextConstants.TRANSACTION_CONTEXT, JSON.toJSONString(transactionContext));

远程dubbo服务可以直接获取。