tcc-transaction 分析

时间：2023-12-13 22:43:38

　　tcc-transaction是TCC型事务java实现，具体项目地址  点我。本文通过tcc-transaction和Springcloud,分析下tcc-transaction的原理。

　　要了解一个东西首先就要先会用它，tcc-transaction本身有多个模块，由于我们是和springcloud结合，所以我们只需要引入以下四个模块。

tcc-transaction 分析

这次demo我们就两个服务，分别是Trade(交易)和account(积分)。交易在完成的同时，给当前用户增加指定的积分。各个项目只需要引入一个tcc配置类

@Configuration

@ImportResource(locations = "classpath:tcc-transaction.xml")

public class TccConfig {

    @Autowired

    private TccDataSourceProperties properties;

    @Bean

    public TransactionRepository transactionRepository(ObjectSerializer<?> serializer) {

        SpringJdbcTransactionRepository repository = new SpringJdbcTransactionRepository();
 repository.setDataSource(druidDataSource());

        repository.setDomain("TRADE");

        repository.setTbSuffix("_TRADE");

        repository.setSerializer(serializer);

        return repository;

    }

    /**

     * 设置恢复策略

     * @return

     */

    @Bean

    public DefaultRecoverConfig defaultRecoverConfig(){

        DefaultRecoverConfig defaultRecoverConfig=new DefaultRecoverConfig();

        defaultRecoverConfig.setCronExpression("0 */1 * * * ?");

        defaultRecoverConfig.setMaxRetryCount(120);

        defaultRecoverConfig.setMaxRetryCount(30);

        return defaultRecoverConfig;

    }

    public DataSource druidDataSource(){

        DruidDataSource datasource = new DruidDataSource();

        datasource.setUrl(properties.getUrl());

        datasource.setDriverClassName(properties.getDriverClassName());

        datasource.setUsername(properties.getUsername());

        datasource.setPassword(properties.getPassword());

        datasource.setInitialSize(10);

        datasource.setMinIdle(1);

        datasource.setMaxWait(6000);

        datasource.setMaxActive(10);

        datasource.setMinEvictableIdleTimeMillis(1800000);

        return datasource;

    }

    @Bean

    public ObjectSerializer<?> objectSerializer() {

        return new KryoTransactionSerializer();

    }

}

TransactionRepository中的subffix要和表名一致，例如，trade表的建表语句如下

CREATE TABLE `TCC_TRANSACTION_TRADE` (

  `TRANSACTION_ID` int(11) NOT NULL AUTO_INCREMENT,

  `DOMAIN` varchar(100) DEFAULT NULL,

  `GLOBAL_TX_ID` varbinary(32) NOT NULL,

  `BRANCH_QUALIFIER` varbinary(32) NOT NULL,

  `CONTENT` varbinary(8000) DEFAULT NULL,

  `STATUS` int(11) DEFAULT NULL,

  `TRANSACTION_TYPE` int(11) DEFAULT NULL,

  `RETRIED_COUNT` int(11) DEFAULT NULL,

  `CREATE_TIME` datetime DEFAULT NULL,

  `LAST_UPDATE_TIME` datetime DEFAULT NULL,

  `VERSION` int(11) DEFAULT NULL,

  PRIMARY KEY (`TRANSACTION_ID`),

  UNIQUE KEY `UX_TX_BQ` (`GLOBAL_TX_ID`,`BRANCH_QUALIFIER`)

) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8;

 

引入之后我们便可以正式的使用了，首先我们先发布account的tcc服务

@Compensable(confirmMethod = "confirmAddPoint",cancelMethod = "cancleAddPoint")
public boolean addPoint(TransactionContext transactionContext, Integer point, String userId) {

        Account account = accountService.

                selectOne(new EntityWrapper<Account>().eq("user_id", userId));

        CheckUtil.notNull(account,"account not null");

        return true;

    }

    @Override

    public void confirmAddPoint(TransactionContext transactionContext, Integer point, String userId) {

        //建议用悲观锁先锁定资源再去更新

        logger.info("确定 新增积分");

        throw new RuntimeException("模拟confirm阶段异常抛出");

//        Account account = accountService.

//                selectOne(new EntityWrapper<Account>().eq("user_id", userId));

//        Long point1 = account.getPoint();

//        long l = point1 + point;

//        account.setPoint(l);

//        boolean update = accountService.updateById(account);

//        CheckUtil.notFalse(update,"account update fail");

    }

    @Override

    public void cancleAddPoint(TransactionContext transactionContext, Integer point, String userId) {

        //trying阶段没做任何操作

    }

发布tcc服务有4个约束

  1. 在服务方法上加上@Compensable注解
  2. 服务方法第一个入参类型为org.mengyun.tcctransaction.api.TransactionContext
  3. 服务方法的入参都须能序列化(实现Serializable接口)
  4. try方法、confirm方法和cancel方法入参类型须一样

我们再去调用account的tcc服务

@Compensable(confirmMethod = "confirmPay",cancelMethod = "canselPay")
public boolean pay(Long tradeId,String userId) {

        Trade trade = tradeService.selectById(tradeId);

        CheckUtil.notNull(trade,"trade not null");

        CheckUtil.notNull(trade.getPrice(),"price not null");

        CheckUtil.notFalse(trade.getStatus().compareTo(TradeStatusEnum.CONFIRMINF.getValue())==0,"trade is finish");
Result result = accountRouteService.addPoint(null,trade.getPrice().intValue(), userId);

        CheckUtil.notFalse(result.isSuccess(),"积分增加失败");

        return true;

    }

    @Override

    public void confirmPay(Long tradeId,String userId) {

        logger.info("开始confirm pay");

        Trade trade = tradeService.selectById(tradeId);

        //订单已完结不作任何处理

        if (trade.getStatus().compareTo(TradeStatusEnum.CONFIRMINF.getValue())!=0){

            return;

        }

        Trade trade1=new Trade();

        trade1.setId(trade.getId());

        trade1.setStatus(TradeStatusEnum.SUCCESS.getValue());

        boolean update = tradeService.updateById(trade1);

        CheckUtil.notFalse(update,"trade fail to success");

    }

    @Override

    public void canselPay(Long tradeId,String userId) {

        Trade trade = tradeService.selectById(tradeId);

        //订单已完结不作任何处理

        if (trade.getStatus().compareTo(TradeStatusEnum.CONFIRMINF.getValue())!=0){

            return;

        }

        Trade trade1=new Trade();

        trade1.setId(trade.getId());

        trade1.setStatus(TradeStatusEnum.FAIL.getValue());

        boolean update = tradeService.updateById(trade1);

        CheckUtil.notFalse(update,"trade fail to failed");

    }

与发布一个Tcc服务不同，本地Tcc服务方法有三个约束：

  1. 在服务方法上加上@Compensable注解
  2. 服务方法的入参都须能序列化(实现Serializable接口)
  3. try方法、confirm方法和cancel方法入参类型须一样

即与发布Tcc服务不同的是本地Tcc服务无需声明服务方法第一个入参类型为org.mengyun.tcctransaction.api.TransactionContext。

更细节的使用我们还是要参看官网，下面我们启动服务，故意在confirm阶段抛异常，就会发现trade和account会一直在confirm阶段不断重试，然后我们将confirm阶段异常去掉，两个服务都会同时达到成功，就像处于同一个数据库事务一样。

tcc主要依靠aop拦截来实现。我们可以先看下tcc-transaction.xml配置文件

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xmlns:context="http://www.springframework.org/schema/context"

       xmlns:aop="http://www.springframework.org/schema/aop" xmlns:task="http://www.springframework.org/schema/task"

       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd   http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">

    <!-- 开启Spring对@AspectJ风格切面的支持(因为下面用到自定义的TCC补偿切面类) -->

    <!-- @Aspect注解不能被Spring自动识别并注册为Bean,因此要通过xml的bean配置,或通过@Compenent注解标识其为Spring管理Bean -->

    <aop:aspectj-autoproxy/>

    <context:component-scan base-package="org.mengyun.tcctransaction.spring"/>

    <bean id="springBeanFactory" class="org.mengyun.tcctransaction.spring.support.SpringBeanFactory"/>

    <!-- TCC事务配置器 -->

    <bean id="tccTransactionConfigurator" class="org.mengyun.tcctransaction.spring.support.TccTransactionConfigurator">

    </bean>

    <!-- 事务恢复 -->

    <bean id="transactionRecovery" class="org.mengyun.tcctransaction.recover.TransactionRecovery">

        <property name="transactionConfigurator" ref="tccTransactionConfigurator"/>

    </bean>

    <!-- 可补偿事务拦截器 -->

    <bean id="compensableTransactionInterceptor"

          class="org.mengyun.tcctransaction.interceptor.CompensableTransactionInterceptor">

        <property name="transactionConfigurator" ref="tccTransactionConfigurator"/>

    </bean>

    <!-- 资源协调拦截器 -->

    <bean id="resourceCoordinatorInterceptor"

          class="org.mengyun.tcctransaction.interceptor.ResourceCoordinatorInterceptor">

        <property name="transactionConfigurator" ref="tccTransactionConfigurator"/>

    </bean>

    <!-- TCC补偿切面 -->

    <bean id="tccCompensableAspect" class="org.mengyun.tcctransaction.spring.TccCompensableAspect">

        <property name="compensableTransactionInterceptor" ref="compensableTransactionInterceptor"/>

    </bean>

    <!-- TCC事务上下文切面 -->

    <bean id="transactionContextAspect" class="org.mengyun.tcctransaction.spring.TccTransactionContextAspect">

        <property name="resourceCoordinatorInterceptor" ref="resourceCoordinatorInterceptor"/>

    </bean>

    <!-- 启用定时任务注解 -->

    <task:annotation-driven/>

    <!-- 事务恢复任务调度器 -->

    <bean id="recoverScheduler" class="org.springframework.scheduling.quartz.SchedulerFactoryBean"/>

    <!-- 事务恢复调度任务，初始化方法：init -->

    <bean id="recoverScheduledJob" class="org.mengyun.tcctransaction.spring.recover.RecoverScheduledJob" init-method="init">

        <property name="transactionConfigurator" ref="tccTransactionConfigurator"/>

        <property name="transactionRecovery" ref="transactionRecovery"/>

        <property name="scheduler" ref="recoverScheduler"/>

    </bean>

</beans>

TccTransactionConfigurator配置了tcc的数据库连接和事务管理

public class TccTransactionConfigurator implements TransactionConfigurator {

    /**

     * 事务库

     */

    @Autowired

    private TransactionRepository transactionRepository;

    /**

     * 事务恢复配置

     */

    @Autowired(required = false)

    private RecoverConfig recoverConfig = DefaultRecoverConfig.INSTANCE;

    .....

}

TransactionRepository就是我们在TccConfig中配置的

@Bean

    public TransactionRepository transactionRepository(ObjectSerializer<?> serializer) {

        SpringJdbcTransactionRepository repository = new SpringJdbcTransactionRepository();

        repository.setDataSource(druidDataSource());

        repository.setDomain("ACCOUNT");

        repository.setTbSuffix("_ACCOUNT");

        repository.setSerializer(serializer);

        return repository;

    }

接下来就是两个拦截器和切面

两个拦截器的切点如下

TccCompensableAspect

@Pointcut("@annotation(org.mengyun.tcctransaction.Compensable)")
TccTransactionContextAspect

@Pointcut("execution(public * *(org.mengyun.tcctransaction.api.TransactionContext,..))||@annotation(org.mengyun.tcctransaction.Compensable)")

两个拦截器分别拦截  Compensable 注解和方法参数第一位是是TransactionContext 的方法。当我们调用以下方法时，拦截就会开始

@Compensable(confirmMethod = "confirmPay",cancelMethod = "canselPay")
public boolean pay(Long tradeId,String userId) {

        ......

        return true;

    }

首先进入的切面是 org.mengyun.tcctransaction.interceptor.CompensableTransactionInterceptor#interceptCompensableMethod

public Object interceptCompensableMethod(ProceedingJoinPoint pjp) throws Throwable {

        // 从拦截方法的参数中获取事务上下文

        TransactionContext transactionContext = CompensableMethodUtils.getTransactionContextFromArgs(pjp.getArgs());

        // 计算可补偿事务方法类型

        MethodType methodType = CompensableMethodUtils.calculateMethodType(transactionContext, true);

        logger.debug("==>interceptCompensableMethod methodType:" + methodType.toString());

        switch (methodType) {

            case ROOT:

            // 主事务方法的处理

                return rootMethodProceed(pjp);

            case PROVIDER:

            // 服务提供者事务方法处理

                return providerMethodProceed(pjp, transactionContext);

            default:

                return pjp.proceed(); // 其他的方法都是直接执行

        }

    }

切面首先会获取到TransactionContext，但是在服务的发起方TransactionContext是null，在下一步的计算方法类型的步骤中，切面通过是否含有注解和TransactionContext来计算类型。

private Object rootMethodProceed(ProceedingJoinPoint pjp) throws Throwable {
transactionConfigurator.getTransactionManager().begin(); // 事务开始（创建事务日志记录，并在当前线程缓存该事务日志记录）

        Object returnValue = null; // 返回值

        try {

            logger.debug("==>rootMethodProceed try begin");
returnValue = pjp.proceed();  // Try (开始执行被拦截的方法)

            logger.debug("==>rootMethodProceed try end");

        } catch (OptimisticLockException e) {

            logger.warn("==>compensable transaction trying exception.", e);

            throw e; //do not rollback, waiting for recovery job

        } catch (Throwable tryingException) {

            logger.warn("compensable transaction trying failed.", tryingException);
 transactionConfigurator.getTransactionManager().rollback();
throw tryingException;

        }

        logger.info("===>rootMethodProceed begin commit()");
 transactionConfigurator.getTransactionManager().commit(); // Try检验正常后提交(事务管理器在控制提交)

        return returnValue;

    }

在rootMethodProceed中，首先新建一个全局事务，保存到数据库

public void begin() {

        LOG.debug("==>begin()");

        Transaction transaction = new Transaction(TransactionType.ROOT); // 事务类型为ROOT:1

        LOG.debug("==>TransactionType:" + transaction.getTransactionType().toString() + ", Transaction Status:" + transaction.getStatus().toString());

        TransactionRepository transactionRepository = transactionConfigurator.getTransactionRepository();

        transactionRepository.create(transaction); // 创建事务记录,写入事务日志库

        threadLocalTransaction.set(transaction); // 将该事务日志记录存入当前线程的事务局部变量中

    }

接着会执行pjp.proceed(); 还记得上面讲的两个切面么，这个时候会被第二个切面了拦截

org.mengyun.tcctransaction.interceptor.ResourceCoordinatorInterceptor#interceptTransactionContextMethod

public Object interceptTransactionContextMethod(ProceedingJoinPoint pjp) throws Throwable {

        LOG.debug("==>interceptTransactionContextMethod(ProceedingJoinPoint pjp)");

        // 获取当前事务

        Transaction transaction = transactionConfigurator.getTransactionManager().getCurrentTransaction();

        // Trying(判断是否Try阶段的事务)

        if (transaction != null && transaction.getStatus().equals(TransactionStatus.TRYING)) {

            LOG.debug("==>TransactionStatus:" + transaction.getStatus().toString());

            // 从参数获取事务上下文

            TransactionContext transactionContext = CompensableMethodUtils.getTransactionContextFromArgs(pjp.getArgs());

            // 获取事务补偿注解

            Compensable compensable = getCompensable(pjp);

            // 计算方法类型

            MethodType methodType = CompensableMethodUtils.calculateMethodType(transactionContext, compensable != null ? true : false);

            LOG.debug("==>methodType:" + methodType.toString());

            switch (methodType) {

                case ROOT:

                    generateAndEnlistRootParticipant(pjp); // 生成和登记根参与者

                    break;

                case CONSUMER:

                    generateAndEnlistConsumerParticipant(pjp); // 生成并登记消费者的参与者

                    break;

                case PROVIDER:

                    generateAndEnlistProviderParticipant(pjp); // 生成并登记服务提供者的参与者

                    break;

            }

        }

        LOG.debug("==>pjp.proceed(pjp.getArgs())");

        return pjp.proceed(pjp.getArgs());

    }

这里计算出来的类型是root，我们直接看 generateAndEnlistRootParticipant

private Participant generateAndEnlistRootParticipant(ProceedingJoinPoint pjp) {

        LOG.debug("==>generateAndEnlistRootParticipant(ProceedingJoinPoint pjp)");

        MethodSignature signature = (MethodSignature) pjp.getSignature();

        Method method = signature.getMethod();

        Compensable compensable = getCompensable(pjp);

        String confirmMethodName = compensable.confirmMethod(); // 确认方法

        String cancelMethodName = compensable.cancelMethod(); // 取消方法

        // 获取当前事务

        Transaction transaction = transactionConfigurator.getTransactionManager().getCurrentTransaction();

        // 获取事务Xid

        TransactionXid xid = new TransactionXid(transaction.getXid().getGlobalTransactionId());

        LOG.debug("==>TransactionXid：" + TransactionXid.byteArrayToUUID(xid.getGlobalTransactionId()).toString()

                + "|" + TransactionXid.byteArrayToUUID(xid.getBranchQualifier()).toString());

        Class targetClass = ReflectionUtils.getDeclaringType(pjp.getTarget().getClass(), method.getName(), method.getParameterTypes());

        // 构建确认方法的提交上下文

        InvocationContext confirmInvocation = new InvocationContext(targetClass,

                confirmMethodName,

                method.getParameterTypes(), pjp.getArgs());

        // 构建取消方法的提交上下文

        InvocationContext cancelInvocation = new InvocationContext(targetClass,

                cancelMethodName,

                method.getParameterTypes(), pjp.getArgs());

        // 构建参与者对像

        Participant participant =

                new Participant(

                        xid,

                        new Terminator(confirmInvocation, cancelInvocation));

        // 加入参与者

        transaction.enlistParticipant(participant);

        TransactionRepository transactionRepository = transactionConfigurator.getTransactionRepository();

        // 更新事务

        transactionRepository.update(transaction);

        return participant;

    }

generateAndEnlistRootParticipant主要是构建Compensable注解上的两个参数，confirmMethod 和 cancelMethod，并构建两个方法的上下文，序列化保存到数据库中。

现在终于可以执行我们真正的逻辑了。当我们执行到调用远程服务时。因为该方法的第一个参数是TransactionContext。所以又要被第二个切面拦截了。

Result result = accountRouteService.addPoint(null,trade.getPrice().intValue(), userId);
@PostMapping(value = "tccaccount/account/addPoint")
Result addPoint(@RequestBody TransactionContext context, @RequestParam(value = "point") Integer point,
                @RequestParam(value = "userId") String userId);
public Object interceptTransactionContextMethod(ProceedingJoinPoint pjp) throws Throwable {

        LOG.debug("==>interceptTransactionContextMethod(ProceedingJoinPoint pjp)");

        // 获取当前事务

        Transaction transaction = transactionConfigurator.getTransactionManager().getCurrentTransaction();

        // Trying(判断是否Try阶段的事务)

        if (transaction != null && transaction.getStatus().equals(TransactionStatus.TRYING)) {

            LOG.debug("==>TransactionStatus:" + transaction.getStatus().toString());

            // 从参数获取事务上下文

            TransactionContext transactionContext = CompensableMethodUtils.getTransactionContextFromArgs(pjp.getArgs());

            // 获取事务补偿注解

            Compensable compensable = getCompensable(pjp);

            // 计算方法类型

            MethodType methodType = CompensableMethodUtils.calculateMethodType(transactionContext, compensable != null ? true : false);

            LOG.debug("==>methodType:" + methodType.toString());

            switch (methodType) {

                case ROOT:

                    generateAndEnlistRootParticipant(pjp); // 生成和登记根参与者

                    break;

                case CONSUMER:

                    generateAndEnlistConsumerParticipant(pjp); // 生成并登记消费者的参与者

                    break;

                case PROVIDER:

                    generateAndEnlistProviderParticipant(pjp); // 生成并登记服务提供者的参与者

                    break;

            }

        }

这里methodType计算出来是CONSUMER

private Participant generateAndEnlistConsumerParticipant(ProceedingJoinPoint pjp) {

        LOG.debug("==>generateAndEnlistConsumerParticipant(ProceedingJoinPoint pjp)");

        MethodSignature signature = (MethodSignature) pjp.getSignature();

        Method method = signature.getMethod();

        Transaction transaction = transactionConfigurator.getTransactionManager().getCurrentTransaction(); // 获取当前事务

        TransactionXid xid = new TransactionXid(transaction.getXid().getGlobalTransactionId()); // 获取事务Xid

        LOG.debug("==>TransactionXid：" + TransactionXid.byteArrayToUUID(xid.getGlobalTransactionId()).toString()

                + "|" + TransactionXid.byteArrayToUUID(xid.getBranchQualifier()).toString());

        // 获取事务上下文参数的位置

        int position = CompensableMethodUtils.getTransactionContextParamPosition(((MethodSignature) pjp.getSignature()).getParameterTypes());

        // 给服务接口的TransactionContext参数设值

        pjp.getArgs()[position] = new TransactionContext(xid, transaction.getStatus().getId()); // 构建事务上下文

        Object[] tryArgs = pjp.getArgs(); // 获取服务接口参数

        Object[] confirmArgs = new Object[tryArgs.length]; // 确认提交参数

        Object[] cancelArgs = new Object[tryArgs.length]; // 取消提交参数

        System.arraycopy(tryArgs, 0, confirmArgs, 0, tryArgs.length); // 数组拷贝
 confirmArgs[position] = new TransactionContext(xid, TransactionStatus.CONFIRMING.getId()); // 构建事务确认上下文

        System.arraycopy(tryArgs, 0, cancelArgs, 0, tryArgs.length); // 数组拷贝
  cancelArgs[position] = new TransactionContext(xid, TransactionStatus.CANCELLING.getId()); // 构建事务取消上下文

        Class targetClass = ReflectionUtils.getDeclaringType(pjp.getTarget().getClass(), method.getName(), method.getParameterTypes());

        // 构建确认方法的提交上下文

        InvocationContext confirmInvocation = new InvocationContext(targetClass, method.getName(), method.getParameterTypes(), confirmArgs);

        // 构建取消方法的提交上下文

        InvocationContext cancelInvocation = new InvocationContext(targetClass, method.getName(), method.getParameterTypes(), cancelArgs);

        // 构建参与者对像

        Participant participant =

                new Participant(

                        xid,

                        new Terminator(confirmInvocation, cancelInvocation));

        transaction.enlistParticipant(participant); // 加入到参与者

        TransactionRepository transactionRepository = transactionConfigurator.getTransactionRepository();

        transactionRepository.update(transaction); // 更新事务

        return participant;

    }

这个generateAndEnlistConsumerParticipant 方法干了什么呢。上文我们调用远程服务时，第一个入参的TransactionContext是null,所以这里我们我们构建了一个 TransactionContext 。这里的全局事务id和之前构造的TransactionContext 都是一致的，然后我们再构建 确认与取消方法保存到数据库中。现在我们在trade的切面拦截终于结束了。调用远程方法到account服务时，我们又遇到了新的一次拦截

@Compensable(confirmMethod = "confirmAddPoint",cancelMethod = "cancleAddPoint")

    public boolean addPoint(TransactionContext transactionContext, Integer point, String userId) {

        Account account = accountService.

                selectOne(new EntityWrapper<Account>().eq("user_id", userId));

        CheckUtil.notNull(account,"account not null");

        return true;

    }

这次的拦截和在trade遇到的相似，但是这次算出来的是 PROVIDER 类型

public Object interceptCompensableMethod(ProceedingJoinPoint pjp) throws Throwable {

        // 从拦截方法的参数中获取事务上下文

        TransactionContext transactionContext = CompensableMethodUtils.getTransactionContextFromArgs(pjp.getArgs());

        // 计算可补偿事务方法类型

        MethodType methodType = CompensableMethodUtils.calculateMethodType(transactionContext, true);

        logger.debug("==>interceptCompensableMethod methodType:" + methodType.toString());

        switch (methodType) {

            case ROOT:

                return rootMethodProceed(pjp); // 主事务方法的处理

            case PROVIDER:

                return providerMethodProceed(pjp, transactionContext); // 服务提供者事务方法处理

            default:

                return pjp.proceed(); // 其他的方法都是直接执行

        }

    }

现在我们是trying阶段，所以现在第一个切面和第二个切面只需要在本地数据库保存下事务的二进制流，校验try阶段的逻辑即可。

private Object providerMethodProceed(ProceedingJoinPoint pjp, TransactionContext transactionContext) throws Throwable {

        logger.debug("==>providerMethodProceed transactionStatus:" + TransactionStatus.valueOf(transactionContext.getStatus()).toString());

        switch (TransactionStatus.valueOf(transactionContext.getStatus())) {
case TRYING:

                logger.debug("==>providerMethodProceed try begin");

                // 基于全局事务ID扩展创建新的分支事务，并存于当前线程的事务局部变量中.

                transactionConfigurator.getTransactionManager().propagationNewBegin(transactionContext);

                logger.debug("==>providerMethodProceed try end");

                return pjp.proceed();
case CONFIRMING:
try {

                    logger.debug("==>providerMethodProceed confirm begin");

                    // 找出存在的事务并处理.

                    transactionConfigurator.getTransactionManager().propagationExistBegin(transactionContext);

                    transactionConfigurator.getTransactionManager().commit(); // 提交

                    logger.debug("==>providerMethodProceed confirm end");

                } catch (NoExistedTransactionException excepton) {

                    //the transaction has been commit,ignore it.

                }

                break;
case CANCELLING:
try {

                    logger.debug("==>providerMethodProceed cancel begin");

                    transactionConfigurator.getTransactionManager().propagationExistBegin(transactionContext);

                    transactionConfigurator.getTransactionManager().rollback(); // 回滚

                    logger.debug("==>providerMethodProceed cancel end");

                } catch (NoExistedTransactionException exception) {

                    //the transaction has been rollback,ignore it.

                }

                break;

        }

        Method method = ((MethodSignature) (pjp.getSignature())).getMethod();

        return ReflectionUtils.getNullValue(method.getReturnType());

    }

当account服务执行成功之后，我们回到trade服务  现在returnValue = pjp.proceed(); 这一行终于执行完了。也就是说各个事务的准备阶段都完成了。只有两种情况，成功和非成功(失败和超时)。我们乐观点，看看成功后是怎么提交全部事物的。

private Object rootMethodProceed(ProceedingJoinPoint pjp) throws Throwable {

        logger.debug("==>rootMethodProceed");

        transactionConfigurator.getTransactionManager().begin(); // 事务开始（创建事务日志记录，并在当前线程缓存该事务日志记录）

        Object returnValue = null; // 返回值

        try {

            logger.debug("==>rootMethodProceed try begin");
 returnValue = pjp.proceed();  // Try (开始执行被拦截的方法)

            logger.debug("==>rootMethodProceed try end");

        } catch (OptimisticLockException e) {

            logger.warn("==>compensable transaction trying exception.", e);

            throw e; //do not rollback, waiting for recovery job

        } catch (Throwable tryingException) {

            logger.warn("compensable transaction trying failed.", tryingException);
transactionConfigurator.getTransactionManager().rollback();
throw tryingException;

        }

        logger.info("===>rootMethodProceed begin commit()");
transactionConfigurator.getTransactionManager().commit(); // Try检验正常后提交(事务管理器在控制提交)

        return returnValue;

    }

事务提交总共分为两部分，

1.更改状态保存到数据库。做持久化保存，这样宕机了也能恢复。

2.反射调用切面拦截保存的参与者

tcc-transaction 分析

public void commit() {

        LOG.debug("==>TransactionManager commit()");

        Transaction transaction = getCurrentTransaction();

        transaction.changeStatus(TransactionStatus.CONFIRMING);

        LOG.debug("==>update transaction status to CONFIRMING");

        //更改状态为CONFIRMING,数据库持久化保存

        transactionConfigurator.getTransactionRepository().update(transaction);

        try {

            LOG.info("==>transaction begin commit()");

            //事务提交
            transaction.commit();
transactionConfigurator.getTransactionRepository().delete(transaction);

        } catch (Throwable commitException) {

            LOG.error("compensable transaction confirm failed.", commitException);

            throw new ConfirmingException(commitException);

        }

    }

    public void commit() {

        LOG.debug("==>Transaction.commit()");

        for (Participant participant : participants) {
participant.commit();

        }

    }

    public void commit() {

        LOG.debug("==>Participant.rollback()");
 terminator.commit();

    }

    public void commit() {

        LOG.debug("==>Terminator commit invoke");
invoke(confirmInvocationContext);

    }

    private Object invoke(InvocationContext invocationContext) {

        if (StringUtils.isNotEmpty(invocationContext.getMethodName())) {

            LOG.debug("==>Terminator invoke " + invocationContext.getTargetClass().getName() + "." + invocationContext.getMethodName());

            try {

                Object target = FactoryBuilder.factoryOf(invocationContext.getTargetClass()).getInstance();

                // 找到要调用的目标方法

                Method method = target.getClass().getMethod(invocationContext.getMethodName(), invocationContext.getParameterTypes());

                // 调用服务方法，被再次被TccTransactionContextAspect和ResourceCoordinatorInterceptor拦截，但因为事务状态已经不再是TRYING了，所以直接执行远程服务

                return method.invoke(target, invocationContext.getArgs()); // 调用服务方法

            } catch (Exception e) {

                throw new SystemException(e);

            }

        }

        return null;

    }

本地的反射调用时很好实现的，但是如何远程反射调用其他服务的confirm方法呢。

tcc-transaction根据transactionContext的状态字段，通过切面拦截 当transactionContext变成CONFIRMING时，就会反射调用confirmInvocationContext

case TRYING:

                logger.debug("==>providerMethodProceed try begin");

                // 基于全局事务ID扩展创建新的分支事务，并存于当前线程的事务局部变量中.

                transactionConfigurator.getTransactionManager().propagationNewBegin(transactionContext);

                logger.debug("==>providerMethodProceed try end");

                return pjp.proceed();
case CONFIRMING:
try {

                    logger.debug("==>providerMethodProceed confirm begin");

                    // 找出存在的事务并处理.
                    transactionConfigurator.getTransactionManager().propagationExistBegin(transactionContext);

                    transactionConfigurator.getTransactionManager().commit(); // 提交

                    logger.debug("==>providerMethodProceed confirm end");

                } catch (NoExistedTransactionException excepton) {

                    //the transaction has been commit,ignore it.

                }

                break;

到这里，一个tcc分布式理论上是已经完成了。但是，我们考虑下一种场景，如果在confirm阶段出现异常怎么办呢？

tcc-transaction还提供了恢复功能。他能从数据库找到还未完成的事物，间隔的去执行，我们也可以配置相关的策略

/**

     * 设置恢复策略

     * @return

     */

    @Bean

    public DefaultRecoverConfig defaultRecoverConfig(){

        DefaultRecoverConfig defaultRecoverConfig=new DefaultRecoverConfig();

        defaultRecoverConfig.setCronExpression("0 */1 * * * ?");

        defaultRecoverConfig.setMaxRetryCount(120);

        return defaultRecoverConfig;

    }

以上就是tcc-transaction的分析

相关文章