Seata框架源码分析——TCC模式

TCC模式使用示例

本文旨在针对Seata框架的TCC模式的源码进行讲解分析，在此不过多介绍Seata框架。
如果想了解更多有关Seata框架的细节，建议可以阅读我的另外一篇博客：Seata框架源码分析——AT模式

为了更新方便之后的源码分析讲解，首先来看下TCC模式的使用示例：

与AT模式的使用非常类似，TCC模式都是使用注解达到分布式事务控制的效果，使用成本非常低。

业务调用方在接口方法上使用@GlobalTransactional注解，开启全局事务

@Servicepublic class OrderServiceImpl implements OrderService { @Autowired OrderDao orderDao; @DubboReference AccountService accountService; @DubboReference StorageService storageService; @Override @GlobalTransactional public boolean dealTCC(Order order) { // 扣减账户 Account account = new Account(); account.setAccountId(order.getAccountId()); account.setUsed(order.getAmount()); accountService.prepare(null, account); // 扣减库存 Storage storage = new Storage(); storage.setStorageId(order.getStorageId()); storage.setUsed(1); storageService.prepare(null, storage); // 保存订单 this.orderDao.insert(order); return true; }}

业务提供方则在接口方法上使用@TwoPhaseBusinessAction注解，同时指定同个接口中分支事务提交和回滚的方法名：

BusinessActionContext是TCC事务中的上下文@BusinessActionContextParameter注解标注的参数会在上下文中传播，能够在指定的提交和回滚方法中使用

public interface AccountService { boolean dealAT(Account account); @TwoPhaseBusinessAction(name = "account-tx", commitMethod = "commit", rollbackMethod = "rollback") boolean prepare(BusinessActionContext actionContext, @BusinessActionContextParameter(paramName = "account") Account account); boolean commit(BusinessActionContext actionContext); boolean rollback(BusinessActionContext actionContext);}

TCC模式源码分析

有了以上使用示例的了解，接下来就是针对其中的源码实现细节进行分析讲解。
Seata模式很好的融入了Spring框架，因此在系统启动时，会扫描相关的Bean进行初始化，并注入，而这个类就是GlobalTransactionScanner：

public class GlobalTransactionScanner extends AbstractAutoProxyCreator implements ConfigurationChangeListener, InitializingBean, ApplicationContextAware, DisposableBean {...}

在GlobalTransactionScanner中最主要的动作有两个：

初始化拦截器（基于注解使用的框架，都是使用拦截的方式对原方法进行增强，Seata亦是如此）初始化TM、RM客户端（在本文中不在赘述，因为此部分已在AT模式源码分析中详细讲解）

初始化拦截器：

@Overrideprotected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) { if (!doCheckers(bean, beanName)) { // bean校验 return bean; } try { synchronized (PROXYED_SET) { // 同步的方式初始化bean if (PROXYED_SET.contains(beanName)) { return bean; } interceptor = null; // 拦截器 // 判断初始化的bean是属于TCC模式，还是AT模式。不同的模式需要初始化的拦截器不同 // TCC模式：TccActionInterceptor // AT模式：GlobalTransactionalInterceptor if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) { // TCC模式 interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName)); ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (ConfigurationChangeListener)interceptor); } else { // AT模式 Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean); Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean); if (!existsAnnotation(new Class[]{serviceInterface}) && !existsAnnotation(interfacesIfJdk)) { return bean; } if (globalTransactionalInterceptor == null) { globalTransactionalInterceptor = new GlobalTransactionalInterceptor(failureHandlerHook); ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (ConfigurationChangeListener)globalTransactionalInterceptor); } interceptor = globalTransactionalInterceptor; } LOGGER.info("Bean[{}] with name [{}] would use interceptor [{}]", bean.getClass().getName(), beanName, interceptor.getClass().getName()); if (!AopUtils.isAopProxy(bean)) { bean = super.wrapIfNecessary(bean, beanName, cacheKey); } else { AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean); Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null)); int pos; for (Advisor avr : advisor) { // Find the position based on the advisor's order, and add to advisors by pos pos = findAddSeataAdvisorPosition(advised, avr); advised.addAdvisor(pos, avr); } } PROXYED_SET.add(beanName); return bean; } } catch (Exception exx) { throw new RuntimeException(exx); }}

由以上wrapIfNecessary方法可以看到，初始化拦截器的关键点是TCCBeanParserUtils.isTccAutoProxy方法，判断该Bean是否属于TCC模式：

public static boolean isTccAutoProxy(Object bean, String beanName, ApplicationContext applicationContext) { boolean isRemotingBean = parserRemotingServiceInfo(bean, beanName); // 解析bean信息 //get RemotingBean description RemotingDesc remotingDesc = DefaultRemotingParser.get().getRemotingBeanDesc(beanName); //is remoting bean if (isRemotingBean) { if (remotingDesc != null && remotingDesc.getProtocol() == Protocols.IN_JVM) { //LocalTCC return isTccProxyTargetBean(remotingDesc); // 判断bean是否属于TCC模式 } else { // sofa:reference / dubbo:reference, factory bean return false; } } else { if (remotingDesc == null) { //check FactoryBean if (isRemotingFactoryBean(bean, beanName, applicationContext)) { remotingDesc = DefaultRemotingParser.get().getRemotingBeanDesc(beanName); return isTccProxyTargetBean(remotingDesc); } else { return false; } } else { return isTccProxyTargetBean(remotingDesc); } }}

解析bean，并注册TCC模式的RM资源管理器：

protected static boolean parserRemotingServiceInfo(Object bean, String beanName) { RemotingParser remotingParser = DefaultRemotingParser.get().isRemoting(bean, beanName); if (remotingParser != null) { return DefaultRemotingParser.get().parserRemotingServiceInfo(bean, beanName, remotingParser) != null; // 解析bean信息 } return false;}

public RemotingDesc parserRemotingServiceInfo(Object bean, String beanName, RemotingParser remotingParser) { RemotingDesc remotingBeanDesc = remotingParser.getServiceDesc(bean, beanName); if (remotingBeanDesc == null) { return null; } remotingServiceMap.put(beanName, remotingBeanDesc); // 将bean缓存起来 Class<?> interfaceClass = remotingBeanDesc.getInterfaceClass(); // 获取bean接口类 Method[] methods = interfaceClass.getMethods(); // 获取bean接口类的所有方法 if (remotingParser.isService(bean, beanName)) { try { Object targetBean = remotingBeanDesc.getTargetBean(); for (Method m : methods) { // 遍历方法 // 判断方法上是否有标注@TwoPhaseBusinessAction注解 TwoPhaseBusinessAction twoPhaseBusinessAction = m.getAnnotation(TwoPhaseBusinessAction.class); if (twoPhaseBusinessAction != null) { // 注解不为空，说明该bean属于TCC模式，以下封装TCC相关资源信息，并进行注册 TCCResource tccResource = new TCCResource(); tccResource.setActionName(twoPhaseBusinessAction.name()); tccResource.setTargetBean(targetBean); tccResource.setPrepareMethod(m); tccResource.setCommitMethodName(twoPhaseBusinessAction.commitMethod()); // 设置指定的提交方法名 tccResource.setCommitMethod(interfaceClass.getMethod(twoPhaseBusinessAction.commitMethod(), twoPhaseBusinessAction.commitArgsClasses())); // 设置指定的提交方法对象 tccResource.setRollbackMethodName(twoPhaseBusinessAction.rollbackMethod()); // 设置指定的回滚方法名 tccResource.setRollbackMethod(interfaceClass.getMethod(twoPhaseBusinessAction.rollbackMethod(), twoPhaseBusinessAction.rollbackArgsClasses())); // 设置指定的回滚方法对象 // 设置提交/回滚方法参数类型 tccResource.setCommitArgsClasses(twoPhaseBusinessAction.commitArgsClasses()); tccResource.setRollbackArgsClasses(twoPhaseBusinessAction.rollbackArgsClasses()); // set phase two method's keys tccResource.setPhaseTwoCommitKeys(this.getTwoPhaseArgs(tccResource.getCommitMethod(), twoPhaseBusinessAction.commitArgsClasses())); tccResource.setPhaseTwoRollbackKeys(this.getTwoPhaseArgs(tccResource.getRollbackMethod(), twoPhaseBusinessAction.rollbackArgsClasses())); // 注册TCC资源，在后期需要获取此资源，调用其中提交/回滚方法 DefaultResourceManager.get().registerResource(tccResource); } } } catch (Throwable t) { throw new frameworkException(t, "parser remoting service error"); } } if (remotingParser.isReference(bean, beanName)) { //reference bean, TCC proxy remotingBeanDesc.setReference(true); } return remotingBeanDesc;}

注册TCC资源：

@Overridepublic void registerResource(Resource resource) { getResourceManager(resource.getBranchType()).registerResource(resource);}

@Overridepublic void registerResource(Resource resource) { TCCResource tccResource = (TCCResource)resource; tccResourceCache.put(tccResource.getResourceId(), tccResource); // 本地缓存TCC资源 super.registerResource(tccResource); // 调用父类注册方法}

@Overridepublic void registerResource(Resource resource) {// TC服务端注册TCC资源 RmNettyRemotingClient.getInstance().registerResource(resource.getResourceGroupId(), resource.getResourceId());}

经过TCC资源的注册，可以知道在分支事务的本地会根据资源id对资源进行缓存，以及TC服务端针对该资源注册（远程注册逻辑不是重点，不分析，有兴趣的朋友可以自行研究DefaultServerMessageListenerImp.onRegRmMessage方法）。

全局事务拦截

对于注解来说，首先拦截的业务调用方的@GlobalTransactional。拦截的类方法则是GlobalTransactionalInterceptor.invoke

@Overridepublic Object invoke(final MethodInvocation methodInvocation) throws Throwable { Class<?> targetClass = methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null; Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass); if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) { final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod); final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, targetClass, GlobalTransactional.class); // 获取拦截方法上的@GlobalTransactional注解 final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class); boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes); if (!localDisable) { // 判断本地是否允许开启分布式事务 if (globalTransactionalAnnotation != null) { // 注解不为空，则说明拦截的方法上有@GlobalTransactional注解，开启分布式事务 return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation); } else if (globalLockAnnotation != null) { return handleGlobalLock(methodInvocation, globalLockAnnotation); } } } return methodInvocation.proceed(); // 本地不允许开启分布式事务/拦截方法上没有@GlobalTransactional注解，则按原方法执行}

全局事务拦截这部分源码与AT模式如出一辙，最后具体是通过调用TransactionalTemplate.execute方法实现本地事务执行：

public Object execute(TransactionalExecutor business) throws Throwable { TransactionInfo txInfo = business.getTransactionInfo(); // 获取事务相关信息 if (txInfo == null) { throw new ShouldNeverHappenException("transactionInfo does not exist"); } GlobalTransaction tx = GlobalTransactionContext.getCurrent(); // 获取当前事务对象（主要是通过全局事务XID进行判断） Propagation propagation = txInfo.getPropagation(); // 获取事务信息中的事务传播级别 SuspendedResourcesHolder suspendedResourcesHolder = null; try { // 针对各个事务级别做特殊处理，在此不多讲，继续往下看 switch (propagation) { case NOT_SUPPORTED: // If transaction is existing, suspend it. if (existingTransaction(tx)) { suspendedResourcesHolder = tx.suspend(); } // Execute without transaction and return. return business.execute(); case REQUIRES_NEW: // If transaction is existing, suspend it, and then begin new transaction. if (existingTransaction(tx)) { suspendedResourcesHolder = tx.suspend(); tx = GlobalTransactionContext.createNew(); } // Continue and execute with new transaction break; case SUPPORTS: // If transaction is not existing, execute without transaction. if (notExistingTransaction(tx)) { return business.execute(); } // Continue and execute with new transaction break; case REQUIRED: // If current transaction is existing, execute with current transaction, // else continue and execute with new transaction. break; case NEVER: // If transaction is existing, throw exception. if (existingTransaction(tx)) { throw new TransactionException( String.format("Existing transaction found for transaction marked with propagation 'never', xid = %s" , tx.getXid())); } else { // Execute without transaction and return. return business.execute(); } case MANDATORY: // If transaction is not existing, throw exception. if (notExistingTransaction(tx)) { throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'"); } // Continue and execute with current transaction. break; default: throw new TransactionException("Not Supported Propagation:" + propagation); } if (tx == null) { // 当前事务对象为空，说明当前未开启事务，新建事务对象 tx = GlobalTransactionContext.createNew(); } GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo); try { beginTransaction(txInfo, tx); // 开启事务（重点） Object rs; try { rs = business.execute(); // 执行原方法 } catch (Throwable ex) { completeTransactionAfterThrowing(txInfo, tx, ex); // 原方法执行异常，回滚（重点） throw ex; } commitTransaction(tx); // 原方法执行无异常，提交（重点） return rs; } finally { // clear resumeGlobalLockConfig(previousConfig); triggerAfterCompletion(); cleanUp(); } } finally { // If the transaction is suspended, resume it. if (suspendedResourcesHolder != null) { tx.resume(suspendedResourcesHolder); } }}

在上面的execute方法中看到，开启事务、执行原方法、成功提交事务、失败回滚事务 这些关键方法。但是对于TCC模式来说，只有当business.execute()执行原方法时，才会调用到其中标注了@TwoPhaseBusinessAction注解方法，该注解才会被拦截增强，TCC模式才会真正生效。
而拦截这个注解的类方法，就是本文开头所讲解的初始化拦截器TccActionInterceptor.invoke：

@Overridepublic Object invoke(final MethodInvocation invocation) throws Throwable { if (!RootContext.inGlobalTransaction() || disable || RootContext.inSagaBranch()) { // 检查是否允许启用全局事务 return invocation.proceed(); // 禁止启用，直接原方法执行 } Method method = getActionInterfaceMethod(invocation); TwoPhaseBusinessAction businessAction = method.getAnnotation(TwoPhaseBusinessAction.class); // 获取方法上的TwoPhaseBusinessAction注解 if (businessAction != null) { // 注解不为空，则说明是TCC模式 String xid = RootContext.getXID(); // 获取全局事务ID BranchType previousBranchType = RootContext.getBranchType(); if (BranchType.TCC != previousBranchType) { RootContext.bindBranchType(BranchType.TCC); } try { // 处理TCC模式方法 return actionInterceptorHandler.proceed(method, invocation.getArguments(), xid, businessAction, invocation::proceed); } finally { //if not TCC, unbind branchType if (BranchType.TCC != previousBranchType) { RootContext.unbindBranchType(); } //MDC remove branchId MDC.remove(RootContext.MDC_KEY_BRANCH_ID); } } return invocation.proceed(); // 注解为空，则说明不是TCC模式，直接原方法执行}

TCC注解拦截处理，注册分支事务，并返回绑定分支事务ID：

public Object proceed(Method method, Object[] arguments, String xid, TwoPhaseBusinessAction businessAction, Callback