登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›Seata源码—7.Seata TCC模式的事务处理

Seata源码—7.Seata TCC模式的事务处理

匡菲 2025-6-3 00:02:07
大纲
1.Seata TCC分布式事务案例配置
2.Seata TCC案例服务提供者启动分析
3.@TwoPhaseBusinessAction注解扫描源码
4.Seata TCC案例分布式事务入口分析
5.TCC核心注解扫描与代理创建入口源码
6.TCC动态代理拦截器TccActionInterceptor
7.Action拦截处理器ActionInterceptorHandler
8.Seata TCC分布式事务的注册提交回滚处理源码
 
1.Seata TCC分布式事务案例配置
(1)位于seata-samples的tcc模块下的Demo工程
(2)Demo工程的配置文件
(3)Demo工程运行说明
 
(1)位于seata-samples的tcc模块下的Demo工程
dubbo-tcc-sample模块主要演示了TCC模式下分布式事务的提交和回滚。该Demo中一个分布式事务内会有两个TCC事务参与者，这两个TCC事务参与者分别是TccActionOne和TccActionTwo。分布式事务提交则两者均提交，分布式事务回滚则两者均回滚。
 
这两个TCC事务参与者均是Dubbo远程服务。一个应用作为服务提供方，会实现这两个TCC参与者，并将它们发布成Dubbo服务。另外一个应用作为事务发起方，会订阅Dubbo服务，然后调用编排TCC参与者，执行远程Dubbo服务。
 
TccActionOne接口定义如下：
  1. public interface TccActionOne {
  2.     @TwoPhaseBusinessAction(name = "DubboTccActionOne", commitMethod = "commit", rollbackMethod = "rollback")
  3.     public boolean prepare(BusinessActionContext actionContext,
  4.         @BusinessActionContextParameter(paramName = "a") int a);
  5.     public boolean commit(BusinessActionContext actionContext);
  6.     public boolean rollback(BusinessActionContext actionContext);
  7. }
复制代码
TccActionTwo接口定义如下：
  1. public interface TccActionTwo {
  2.     @TwoPhaseBusinessAction(name = "DubboTccActionTwo", commitMethod = "commit", rollbackMethod = "rollback")
  3.     public boolean prepare(BusinessActionContext actionContext,
  4.         @BusinessActionContextParameter(paramName = "b") String b,
  5.         @BusinessActionContextParameter(paramName = "c", index = 1) List list);
  6.     public boolean commit(BusinessActionContext actionContext);
  7.     public boolean rollback(BusinessActionContext actionContext);
  8. }
复制代码
(2)Demo工程的配置文件
一.seata-tcc.xml
  1. <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  2.        xmlns="http://www.springframework.org/schema/beans"
  3.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"
  4.        default-autowire="byName">
  6.    
  7.     <bean >
  8.         <constructor-arg value="tcc-sample"/>
  9.         <constructor-arg value="my_test_tx_group"/>
  10.     </bean>
  12.     <bean id="tccActionOneImpl" />
  13.     <bean id="tccActionTwoImpl" />
  14.     <bean id="tccTransactionService" />
  15. </beans>
复制代码
二.seata-dubbo-provider.xml
  1. <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  2.        xmlns="http://www.springframework.org/schema/beans"
  3.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"
  4.        default-autowire="byName">
  6.    
  7.     <bean >
  8.         <constructor-arg value="tcc-sample"/>
  9.         <constructor-arg value="my_test_tx_group"/>
  10.     </bean>
  12.     <bean id="tccActionOneImpl" />
  13.     <bean id="tccActionTwoImpl" />
  14.     <bean id="tccTransactionService" />
  15. </beans>        
复制代码
三.seata-dubbo-reference.xml
  1. <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  2.        xmlns="http://www.springframework.org/schema/beans"
  3.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"
  4.        default-autowire="byName">
  6.    
  7.     <bean >
  8.         <constructor-arg value="tcc-sample"/>
  9.         <constructor-arg value="my_test_tx_group"/>
  10.     </bean>
  12.     <bean id="tccActionOneImpl" />
  13.     <bean id="tccActionTwoImpl" />
  14.     <bean id="tccTransactionService" />
  15. </beans>        
复制代码
(3)Demo工程运行指南
一.启动Seata Server
 
二.启动Dubbo服务应用
运行DubboTccProviderStarter。该应用会发布Dubbo服务，并且实现了两个TCC参与者。
  1. public class TccProviderStarter extends AbstractStarter {
  2.     public static void main(String[] args) throws Exception {
  3.         new TccProviderStarter().start0(args);
  4.     }
  5.    
  6.     @Override
  7.     protected void start0(String[] args) throws Exception {
  8.         ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(
  9.             new String[]{"spring/seata-tcc.xml", "spring/seata-dubbo-provider.xml"}
  10.         );
  11.         new ApplicationKeeper().keep();
  12.     }
  13. }
  15. public class TccActionOneImpl implements TccActionOne {
  16.     @Override
  17.     public boolean prepare(BusinessActionContext actionContext, int a) {
  18.         String xid = actionContext.getXid();
  19.         System.out.println("TccActionOne prepare, xid:" + xid + ", a:" + a);
  20.         return true;
  21.     }
  22.    
  23.     @Override
  24.     public boolean commit(BusinessActionContext actionContext) {
  25.         String xid = actionContext.getXid();
  26.         System.out.println("TccActionOne commit, xid:" + xid + ", a:" + actionContext.getActionContext("a"));
  27.         ResultHolder.setActionOneResult(xid, "T");
  28.         return true;
  29.     }
  30.    
  31.     @Override
  32.     public boolean rollback(BusinessActionContext actionContext) {
  33.         String xid = actionContext.getXid();
  34.         System.out.println("TccActionOne rollback, xid:" + xid + ", a:" + actionContext.getActionContext("a"));
  35.         ResultHolder.setActionOneResult(xid, "R");
  36.         return true;
  37.     }
  38. }
  40. public class TccActionTwoImpl implements TccActionTwo {
  41.     @Override
  42.     public boolean prepare(BusinessActionContext actionContext, String b, List list) {
  43.         String xid = actionContext.getXid();
  44.         System.out.println("TccActionTwo prepare, xid:" + xid + ", b:" + b + ", c:" + list.get(1));
  45.         return true;
  46.     }
  47.    
  48.     @Override
  49.     public boolean commit(BusinessActionContext actionContext) {
  50.         String xid = actionContext.getXid();
  51.         System.out.println("TccActionTwo commit, xid:" + xid + ", b:" + actionContext.getActionContext("b") + ", c:" + actionContext.getActionContext("c"));
  52.         ResultHolder.setActionTwoResult(xid, "T");
  53.         return true;
  54.     }
  55.    
  56.     @Override
  57.     public boolean rollback(BusinessActionContext actionContext) {
  58.         String xid = actionContext.getXid();
  59.         System.out.println("TccActionTwo rollback, xid:" + xid + ", b:" + actionContext.getActionContext("b") + ", c:" + actionContext.getActionContext("c"));
  60.         ResultHolder.setActionTwoResult(xid, "R");
  61.         return true;
  62.     }
  63. }
复制代码
三.启动事务应用
运行TccConsumerStarter。该应用会订阅Dubbo服务，发起分布式事务，调用上述两个TCC参与者，内含TCC事务提交场景和TCC事务回滚场景的演示。
  1. public class TccConsumerStarter extends AbstractStarter {
  2.     static TccTransactionService tccTransactionService = null;
  3.    
  4.     public static void main(String[] args) throws Exception {
  5.         new TccConsumerStarter().start0(args);
  6.     }
  7.    
  8.     @Override
  9.     protected void start0(String[] args) throws Exception {
  10.         ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(
  11.             new String[]{"spring/seata-tcc.xml", "spring/seata-dubbo-reference.xml"}
  12.         );
  13.         tccTransactionService = (TccTransactionService) applicationContext.getBean("tccTransactionService");
  14.         //分布式事务提交demo
  15.         transactionCommitDemo();
  16.         //分布式事务回滚demo
  17.         transactionRollbackDemo();
  18.     }
  19.    
  20.     private static void transactionCommitDemo() throws InterruptedException {
  21.         String txId = tccTransactionService.doTransactionCommit();
  22.         System.out.println(txId);
  23.         Assert.isTrue(StringUtils.isNotEmpty(txId), "事务开启失败");
  24.         System.out.println("transaction commit demo finish.");
  25.     }
  26.    
  27.     private static void transactionRollbackDemo() throws InterruptedException {
  28.         Map map = new HashMap(16);
  29.         try {
  30.             tccTransactionService.doTransactionRollback(map);
  31.             Assert.isTrue(false, "分布式事务未回滚");
  32.         } catch (Throwable t) {
  33.             Assert.isTrue(true, "分布式事务异常回滚");
  34.         }
  35.         String txId = (String) map.get("xid");
  36.         System.out.println(txId);
  37.         System.out.println("transaction rollback demo finish.");
  38.     }
  39. }
  41. public class TccTransactionService {
  42.     private TccActionOne tccActionOne;
  43.     private TccActionTwo tccActionTwo;
  45.     //提交分布式事务
  46.     @GlobalTransactional
  47.     public String doTransactionCommit() {
  48.         //第一个TCC事务参与者
  49.         boolean result = tccActionOne.prepare(null, 1);
  50.         if (!result) {
  51.             throw new RuntimeException("TccActionOne failed.");
  52.         }
  53.         List list = new ArrayList();
  54.         list.add("c1");
  55.         list.add("c2");
  56.         //第二个TCC事务参与者
  57.         result = tccActionTwo.prepare(null, "two", list);
  58.         if (!result) {
  59.             throw new RuntimeException("TccActionTwo failed.");
  60.         }
  61.         return RootContext.getXID();
  62.     }
  63.    
  64.     //回滚分布式事务
  65.     @GlobalTransactional
  66.     public String doTransactionRollback(Map map) {
  67.         //第一个TCC事务参与者
  68.         boolean result = tccActionOne.prepare(null, 1);
  69.         if (!result) {
  70.             throw new RuntimeException("TccActionOne failed.");
  71.         }
  72.         List list = new ArrayList();
  73.         list.add("c1");
  74.         list.add("c2");
  75.         //第二个TCC事务参与者
  76.         result = tccActionTwo.prepare(null, "two", list);
  77.         if (!result) {
  78.             throw new RuntimeException("TccActionTwo failed.");
  79.         }
  80.         map.put("xid", RootContext.getXID());
  81.         throw new RuntimeException("transacton rollback");
  82.     }
  84.     public void setTccActionOne(TccActionOne tccActionOne) {
  85.         this.tccActionOne = tccActionOne;
  86.     }
  87.    
  88.     public void setTccActionTwo(TccActionTwo tccActionTwo) {
  89.         this.tccActionTwo = tccActionTwo;
  90.     }
  91. }
复制代码
 
2.Seata TCC案例服务提供者启动分析
添加了@TwoPhaseBusinessAction注解的接口发布成Dubbo服务：
1.png
 
3.@TwoPhaseBusinessAction注解扫描源码
(1)全局事务注解扫描器的wrapIfNecessary()方法扫描Spring Bean
(2)TCCBeanParserUtils的isTccAutoProxy()方法判断是否要创建TCC动态代理
 
(1)全局事务注解扫描器的wrapIfNecessary()方法扫描Spring Bean
全局事务注解扫描器GlobalTransactionScanner会在调用initClient()方法初始化Seata Client客户端后，通过wrapIfNecessary()方法扫描Spring Bean中含有@TwoPhaseBusinessAction注解的方法。
2.png
  1. //AbstractAutoProxyCreator：Spring的动态代理自动创建者
  2. //ConfigurationChangeListener：关注配置变更事件的监听器
  3. //InitializingBean：Spring Bean初始化回调
  4. //ApplicationContextAware：用来感知Spring容器
  5. //DisposableBean：支持可抛弃Bean
  6. public class GlobalTransactionScanner extends AbstractAutoProxyCreator
  7.         implements ConfigurationChangeListener, InitializingBean, ApplicationContextAware, DisposableBean {
  8.     ...
  9.    
  10.     //InitializingBean接口的回调方法
  11.     //Spring容器启动和初始化完毕后，会调用如下的afterPropertiesSet()方法进行回调
  12.     @Override
  13.     public void afterPropertiesSet() {
  14.         //是否禁用了全局事务，默认是false
  15.         if (disableGlobalTransaction) {
  16.             if (LOGGER.isInfoEnabled()) {
  17.                 LOGGER.info("Global transaction is disabled.");
  18.             }
  19.             ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (ConfigurationChangeListener)this);
  20.             return;
  21.         }
  22.         //通过CAS操作确保initClient()初始化动作仅仅执行一次
  23.         if (initialized.compareAndSet(false, true)) {
  24.             //initClient()方法会对Seata Client进行初始化，比如和Seata Server建立长连接
  25.             //seata-samples的tcc模块的seata-tcc.xml配置文件里都配置了GlobalTransactionScanner这个Bean
  26.             //而GlobalTransactionScanner这个Bean伴随着Spring容器的初始化完毕，都会回调其初始化逻辑initClient()
  27.             initClient();
  28.         }
  29.     }
  31.     //initClient()是核心方法，负责对Seata Client客户端进行初始化
  32.     private void initClient() {
  33.         if (LOGGER.isInfoEnabled()) {
  34.             LOGGER.info("Initializing Global Transaction Clients ... ");
  35.         }
  36.         if (DEFAULT_TX_GROUP_OLD.equals(txServiceGroup)) {
  37.             LOGGER.warn("...", DEFAULT_TX_GROUP_OLD, DEFAULT_TX_GROUP);
  38.         }
  39.         if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
  40.             throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
  41.         }
  43.         //对于Seata Client来说，最重要的组件有两个：
  44.         //一个是TM，即Transaction Manager，用来管理全局事务
  45.         //一个是RM，即Resource Manager，用来管理各分支事务的数据源
  47.         //init TM
  48.         //TMClient.init()会对客户端的TM全局事务管理器进行初始化
  49.         TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
  50.         if (LOGGER.isInfoEnabled()) {
  51.             LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
  52.         }
  54.         //init RM
  55.         //RMClient.init()会对客户端的RM分支事务资源管理器进行初始化
  56.         RMClient.init(applicationId, txServiceGroup);
  57.         if (LOGGER.isInfoEnabled()) {
  58.             LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
  59.         }
  61.         if (LOGGER.isInfoEnabled()) {
  62.             LOGGER.info("Global Transaction Clients are initialized. ");
  63.         }
  65.         //注册Spring容器被销毁时的回调钩子，释放TM和RM两个组件的一些资源
  66.         registerSpringShutdownHook();
  67.     }
  68.    
  69.     //The following will be scanned, and added corresponding interceptor:
  70.     //添加了如下注解的方法会被扫描到，然后方法会添加相应的拦截器进行拦截
  71.    
  72.     //TM:
  73.     //@see io.seata.spring.annotation.GlobalTransactional // TM annotation
  74.     //Corresponding interceptor:
  75.     //@see io.seata.spring.annotation.GlobalTransactionalInterceptor#handleGlobalTransaction(MethodInvocation, AspectTransactional) // TM handler
  76.    
  77.     //GlobalLock:
  78.     //@see io.seata.spring.annotation.GlobalLock // GlobalLock annotation
  79.     //Corresponding interceptor:
  80.     //@see io.seata.spring.annotation.GlobalTransactionalInterceptor# handleGlobalLock(MethodInvocation, GlobalLock)  // GlobalLock handler
  81.    
  82.     //TCC mode:
  83.     //@see io.seata.rm.tcc.api.LocalTCC // TCC annotation on interface
  84.     //@see io.seata.rm.tcc.api.TwoPhaseBusinessAction // TCC annotation on try method
  85.     //@see io.seata.rm.tcc.remoting.RemotingParser // Remote TCC service parser
  86.     //Corresponding interceptor:
  87.     //@see io.seata.spring.tcc.TccActionInterceptor // the interceptor of TCC mode
  88.    
  89.     @Override
  90.     //由于GlobalTransactionScanner继承了Spring提供的AbstractAutoProxyCreator，
  91.     //所以Spring会把Spring Bean传递给GlobalTransactionScanner.wrapIfNecessary()进行判断；
  92.     //让GlobalTransactionScanner来决定是否要根据Bean的Class、或者Bean的方法是否有上述注解，
  93.     //从而决定是否需要针对Bean的Class创建动态代理，来对添加了注解的方法进行拦截；
  94.     protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
  95.         //do checkers
  96.         if (!doCheckers(bean, beanName)) {
  97.             return bean;
  98.         }
  100.         try {
  101.             synchronized (PROXYED_SET) {
  102.                 if (PROXYED_SET.contains(beanName)) {
  103.                     return bean;
  104.                 }
  105.                 interceptor = null;
  107.                 //check TCC proxy
  108.                 //判断传递进来的Bean是否是TCC动态代理
  109.                 //服务启动时会把所有的Bean传递进来这个wrapIfNecessary()，检查这个Bean是否需要创建TCC动态代理
  110.                 if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {
  111.                     //init tcc fence clean task if enable useTccFence
  112.                     TCCBeanParserUtils.initTccFenceCleanTask(TCCBeanParserUtils.getRemotingDesc(beanName), applicationContext);
  113.                     //TCC interceptor, proxy bean of sofa:reference/dubbo:reference, and LocalTCC
  114.                     interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));
  115.                     ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (ConfigurationChangeListener)interceptor);
  116.                 } else {
  117.                     //获取目标class的接口
  118.                     Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
  119.                     Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);
  121.                     //existsAnnotation()方法会判断Bean的Class或Method是否添加了@GlobalTransactional等注解
  122.                     if (!existsAnnotation(new Class[]{serviceInterface}) && !existsAnnotation(interfacesIfJdk)) {
  123.                         return bean;
  124.                     }
  126.                     if (globalTransactionalInterceptor == null) {
  127.                         //构建一个GlobalTransactionalInterceptor，即全局事务注解的拦截器
  128.                         globalTransactionalInterceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
  129.                         ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (ConfigurationChangeListener)globalTransactionalInterceptor);
  130.                     }
  131.                     interceptor = globalTransactionalInterceptor;
  132.                 }
  134.                 LOGGER.info("Bean[{}] with name [{}] would use interceptor [{}]", bean.getClass().getName(), beanName, interceptor.getClass().getName());
  135.                 if (!AopUtils.isAopProxy(bean)) {//如果这个Bean并不是AOP代理
  136.                     //接下来会基于Spring的AbstractAutoProxyCreator创建针对目标Bean接口的动态代理
  137.                     //这样后续调用到目标Bean的方法，就会调用到TccActionInterceptor拦截器
  138.                     bean = super.wrapIfNecessary(bean, beanName, cacheKey);
  139.                 } else {
  140.                     AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
  141.                     Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
  142.                     int pos;
  143.                     for (Advisor avr : advisor) {
  144.                         //Find the position based on the advisor's order, and add to advisors by pos
  145.                         pos = findAddSeataAdvisorPosition(advised, avr);
  146.                         advised.addAdvisor(pos, avr);
  147.                     }
  148.                 }
  149.                 PROXYED_SET.add(beanName);
  150.                 return bean;
  151.             }
  152.         } catch (Exception exx) {
  153.             throw new RuntimeException(exx);
  154.         }
  155.     }
  156.     ...
  157. }
复制代码
(2)TCCBeanParserUtils的isTccAutoProxy()方法判断是否要创建TCC动态代理
  1. public class TCCBeanParserUtils {
  2.     private TCCBeanParserUtils() {
  3.     }
  5.     //is auto proxy TCC bean
  6.     public static boolean isTccAutoProxy(Object bean, String beanName, ApplicationContext applicationContext) {
  7.         boolean isRemotingBean = parserRemotingServiceInfo(bean, beanName);
  8.         //get RemotingBean description
  9.         RemotingDesc remotingDesc = DefaultRemotingParser.get().getRemotingBeanDesc(beanName);
  10.         //is remoting bean
  11.         if (isRemotingBean) {
  12.             if (remotingDesc != null && remotingDesc.getProtocol() == Protocols.IN_JVM) {
  13.                 //LocalTCC
  14.                 //创建一个local tcc代理
  15.                 return isTccProxyTargetBean(remotingDesc);
  16.             } else {
  17.                 //sofa:reference / dubbo:reference, factory bean
  18.                 return false;
  19.             }
  20.         } else {
  21.             if (remotingDesc == null) {
  22.                 //check FactoryBean
  23.                 if (isRemotingFactoryBean(bean, beanName, applicationContext)) {
  24.                     remotingDesc = DefaultRemotingParser.get().getRemotingBeanDesc(beanName);
  25.                     return isTccProxyTargetBean(remotingDesc);
  26.                 } else {
  27.                     return false;
  28.                 }
  29.             } else {
  30.                 return isTccProxyTargetBean(remotingDesc);
  31.             }
  32.         }
  33.     }
  34.     ...
  35.    
  36.     //is TCC proxy-bean/target-bean: LocalTCC , the proxy bean of sofa:reference/dubbo:reference
  37.     public static boolean isTccProxyTargetBean(RemotingDesc remotingDesc) {
  38.         if (remotingDesc == null) {
  39.             return false;
  40.         }
  41.         //check if it is TCC bean
  42.         boolean isTccClazz = false;
  43.         //针对我们的class拿到一个接口class
  44.         Class<?> tccInterfaceClazz = remotingDesc.getInterfaceClass();
  45.         //获取我们的接口里定义的所有的方法
  46.         Method[] methods = tccInterfaceClazz.getMethods();
  48.         TwoPhaseBusinessAction twoPhaseBusinessAction;
  50.         //遍历所有的方法
  51.         for (Method method : methods) {
  52.             //获取的方法是否加了@TwoPhaseBusinessAction注解
  53.             twoPhaseBusinessAction = method.getAnnotation(TwoPhaseBusinessAction.class);
  54.             if (twoPhaseBusinessAction != null) {
  55.                 isTccClazz = true;
  56.                 break;
  57.             }
  58.         }
  59.         if (!isTccClazz) {
  60.             return false;
  61.         }
  62.         short protocols = remotingDesc.getProtocol();
  63.         //LocalTCC
  64.         if (Protocols.IN_JVM == protocols) {
  65.             //in jvm TCC bean , AOP
  66.             return true;
  67.         }
  68.         //sofa:reference /  dubbo:reference, AOP
  69.         return remotingDesc.isReference();
  70.     }
  71.     ...
  72. }
复制代码
 
4.Seata TCC案例分布式事务入口分析
TccTransactionService作为分布式事务的入口，其提交事务和回滚事务的接口都会被添加上@GlobalTransactional注解。
 
所以应用启动时，TccTransactionService的Bean就会被GlobalTransactionScanner扫描，然后其下添加了@GlobalTransactional注解的接口就会被创建动态代理。
 
在TccTransactionService的提交分布式事务的接口中，会先后调用TccActionOne和TccActionTwo两个Dubbo服务。并且在调用两个Dubbo服务时，会通过ApacheDubboTransactionPropagationFilter传递xid。
3.png
  1. public class TccTransactionService {
  2.     private TccActionOne tccActionOne;
  3.     private TccActionTwo tccActionTwo;
  5.     //提交分布式事务
  6.     @GlobalTransactional
  7.     public String doTransactionCommit() {
  8.         //第一个TCC事务参与者
  9.         boolean result = tccActionOne.prepare(null, 1);
  10.         if (!result) {
  11.             throw new RuntimeException("TccActionOne failed.");
  12.         }
  13.         List list = new ArrayList();
  14.         list.add("c1");
  15.         list.add("c2");
  16.         //第二个TCC事务参与者
  17.         result = tccActionTwo.prepare(null, "two", list);
  18.         if (!result) {
  19.             throw new RuntimeException("TccActionTwo failed.");
  20.         }
  21.         return RootContext.getXID();
  22.     }
  24.     //回滚分布式事务
  25.     @GlobalTransactional
  26.     public String doTransactionRollback(Map map) {
  27.         //第一个TCC事务参与者
  28.         boolean result = tccActionOne.prepare(null, 1);
  29.         if (!result) {
  30.             throw new RuntimeException("TccActionOne failed.");
  31.         }
  32.         List list = new ArrayList();
  33.         list.add("c1");
  34.         list.add("c2");
  35.         //第二个TCC事务参与者
  36.         result = tccActionTwo.prepare(null, "two", list);
  37.         if (!result) {
  38.             throw new RuntimeException("TccActionTwo failed.");
  39.         }
  40.         map.put("xid", RootContext.getXID());
  41.         throw new RuntimeException("transacton rollback");
  42.     }
  44.     public void setTccActionOne(TccActionOne tccActionOne) {
  45.         this.tccActionOne = tccActionOne;
  46.     }
  47.    
  48.     public void setTccActionTwo(TccActionTwo tccActionTwo) {
  49.         this.tccActionTwo = tccActionTwo;
  50.     }
  51. }
复制代码
 
5.TCC核心注解扫描与代理创建入口源码
GlobalTransactionScanner的wrapIfNecessary()方法会扫描Spring Bean。TCCBeanParserUtils的isTccAutoProxy()方法会通过判断扫描的Spring Bean中的方法是否添加了TCC的注解，来决定是否要对Bean的方法进行TCC动态代理。
 
注意，其中TCC的注解有两个：@LocalTCC、@TwoPhaseBusinessAction
4.png
  1. public class GlobalTransactionScanner extends AbstractAutoProxyCreator        implements ConfigurationChangeListener, InitializingBean, ApplicationContextAware, DisposableBean {    ...    @Override    //由于GlobalTransactionScanner继承了Spring提供的AbstractAutoProxyCreator，    //所以Spring会把Spring Bean传递给GlobalTransactionScanner.wrapIfNecessary()进行判断；    //让GlobalTransactionScanner来决定是否要根据Bean的Class、或者Bean的方法是否有上述注解，    //从而决定是否需要针对Bean的Class创建动态代理，来对添加了注解的方法进行拦截；    protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {        //do checkers        if (!doCheckers(bean, beanName)) {            return bean;        }        try {            synchronized (PROXYED_SET) {                if (PROXYED_SET.contains(beanName)) {                    return bean;                }                interceptor = null;                //check TCC proxy                //判断传递进来的Bean是否是TCC动态代理                //服务启动时会把所有的Bean传递进来这个wrapIfNecessary()，检查这个Bean是否需要创建TCC动态代理                if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {                    //init tcc fence clean task if enable useTccFence                    TCCBeanParserUtils.initTccFenceCleanTask(TCCBeanParserUtils.getRemotingDesc(beanName), applicationContext);                    //TCC interceptor, proxy bean of sofa:reference/dubbo:reference, and LocalTCC                    interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));                    ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (ConfigurationChangeListener)interceptor);                } else {                    //获取目标class的接口                    Class serviceInterface = SpringProxyUtils.findTargetClass(bean);                    Class[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);                    //existsAnnotation()方法会判断Bean的Class或Method是否添加了@GlobalTransactional等注解                    if (!existsAnnotation(new Class[]{serviceInterface}) && !existsAnnotation(interfacesIfJdk)) {                        return bean;                    }                    if (globalTransactionalInterceptor == null) {                        //构建一个GlobalTransactionalInterceptor，即全局事务注解的拦截器                        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并不是AOP代理                    //接下来会基于Spring的AbstractAutoProxyCreator创建针对目标Bean接口的动态代理                    //这样后续调用到目标Bean的方法，就会调用到TccActionInterceptor拦截器                    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);        }    }    ...}public class TCCBeanParserUtils {
  2.     private TCCBeanParserUtils() {
  3.     }
  5.     //is auto proxy TCC bean
  6.     public static boolean isTccAutoProxy(Object bean, String beanName, ApplicationContext applicationContext) {
  7.         boolean isRemotingBean = parserRemotingServiceInfo(bean, beanName);
  8.         //get RemotingBean description
  9.         RemotingDesc remotingDesc = DefaultRemotingParser.get().getRemotingBeanDesc(beanName);
  10.         //is remoting bean
  11.         if (isRemotingBean) {
  12.             if (remotingDesc != null && remotingDesc.getProtocol() == Protocols.IN_JVM) {
  13.                 //LocalTCC
  14.                 //创建一个local tcc代理
  15.                 return isTccProxyTargetBean(remotingDesc);
  16.             } else {
  17.                 //sofa:reference / dubbo:reference, factory bean
  18.                 return false;
  19.             }
  20.         } else {
  21.             if (remotingDesc == null) {
  22.                 //check FactoryBean
  23.                 if (isRemotingFactoryBean(bean, beanName, applicationContext)) {
  24.                     remotingDesc = DefaultRemotingParser.get().getRemotingBeanDesc(beanName);
  25.                     return isTccProxyTargetBean(remotingDesc);
  26.                 } else {
  27.                     return false;
  28.                 }
  29.             } else {
  30.                 return isTccProxyTargetBean(remotingDesc);
  31.             }
  32.         }
  33.     }
  34.     ...
  35.    
  36.     //is TCC proxy-bean/target-bean: LocalTCC , the proxy bean of sofa:reference/dubbo:reference
  37.     public static boolean isTccProxyTargetBean(RemotingDesc remotingDesc) {
  38.         if (remotingDesc == null) {
  39.             return false;
  40.         }
  41.         //check if it is TCC bean
  42.         boolean isTccClazz = false;
  43.         //针对我们的class拿到一个接口class
  44.         Class<?> tccInterfaceClazz = remotingDesc.getInterfaceClass();
  45.         //获取我们的接口里定义的所有的方法
  46.         Method[] methods = tccInterfaceClazz.getMethods();
  48.         TwoPhaseBusinessAction twoPhaseBusinessAction;
  50.         //遍历所有的方法
  51.         for (Method method : methods) {
  52.             //获取的方法是否加了@TwoPhaseBusinessAction注解
  53.             twoPhaseBusinessAction = method.getAnnotation(TwoPhaseBusinessAction.class);
  54.             if (twoPhaseBusinessAction != null) {
  55.                 isTccClazz = true;
  56.                 break;
  57.             }
  58.         }
  59.         if (!isTccClazz) {
  60.             return false;
  61.         }
  62.         short protocols = remotingDesc.getProtocol();
  63.         //LocalTCC
  64.         if (Protocols.IN_JVM == protocols) {
  65.             //in jvm TCC bean , AOP
  66.             return true;
  67.         }
  68.         //sofa:reference /  dubbo:reference, AOP
  69.         return remotingDesc.isReference();
  70.     }
  71.     ...
  72. }
复制代码
 
6.TCC动态代理拦截器TccActionInterceptor
如果调用添加了TCC的注解的方法，就会执行TccActionInterceptor的invoke()方法，此外只有分支事务的方法才会有可能被TCC动态代理。
 
在TccActionInterceptor的invoke()方法中，会通过ActionInterceptorHandler的proceed()方法来执行具体拦截逻辑。
  1. public class TccActionInterceptor implements MethodInterceptor, ConfigurationChangeListener, Ordered {
  2.     //Action拦截处理器
  3.     private ActionInterceptorHandler actionInterceptorHandler = new ActionInterceptorHandler();
  4.     private volatile boolean disable = ConfigurationFactory.getInstance().getBoolean(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, DEFAULT_DISABLE_GLOBAL_TRANSACTION);
  5.     ...
  6.    
  7.     //如果调用添加了TCC的注解的方法，就会执行如下invoke()方法
  8.     @Override
  9.     public Object invoke(final MethodInvocation invocation) throws Throwable {
  10.         //当前必须是在全局事务里，也就是说分支事务的方法才会有可能被TCC动态代理
  11.         if (!RootContext.inGlobalTransaction() || disable || RootContext.inSagaBranch()) {
  12.             //not in transaction, or this interceptor is disabled
  13.             return invocation.proceed();
  14.         }
  15.         //本次调用的是哪个方法，在class接口里找到这个方法
  16.         Method method = getActionInterfaceMethod(invocation);
  17.         //然后才能找到接口里定义的那个方法上面加的一个注解
  18.         TwoPhaseBusinessAction businessAction = method.getAnnotation(TwoPhaseBusinessAction.class);
  20.         //try method
  21.         if (businessAction != null) {
  22.             //save the xid
  23.             String xid = RootContext.getXID();
  24.             //save the previous branchType
  25.             BranchType previousBranchType = RootContext.getBranchType();
  26.             //if not TCC, bind TCC branchType
  27.             if (BranchType.TCC != previousBranchType) {
  28.                 RootContext.bindBranchType(BranchType.TCC);
  29.             }
  30.             try {
  31.                 //Handler the TCC Aspect, and return the business result
  32.                 //传入actionInterceptorHandler的参数分别是：方法、调用方法传递进来的参数、全局事务xid、分支事务注解、目标方法执行
  33.                 return actionInterceptorHandler.proceed(method, invocation.getArguments(), xid, businessAction, invocation::proceed);
  34.             } finally {
  35.                 //if not TCC, unbind branchType
  36.                 if (BranchType.TCC != previousBranchType) {
  37.                     RootContext.unbindBranchType();
  38.                 }
  39.                 //MDC remove branchId
  40.                 MDC.remove(RootContext.MDC_KEY_BRANCH_ID);
  41.             }
  42.         }
  44.         //not TCC try method
  45.         return invocation.proceed();
  46.     }
  47.    
  48.     //get the method from interface
  49.     protected Method getActionInterfaceMethod(MethodInvocation invocation) {
  50.         Class<?> interfaceType = null;
  51.         try {
  52.             if (remotingDesc == null) {
  53.                 interfaceType = getProxyInterface(invocation.getThis());
  54.             } else {
  55.                 interfaceType = remotingDesc.getInterfaceClass();
  56.             }
  57.             if (interfaceType == null && remotingDesc != null && remotingDesc.getInterfaceClassName() != null) {
  58.                 interfaceType = Class.forName(remotingDesc.getInterfaceClassName(), true, Thread.currentThread().getContextClassLoader());
  59.             }
  60.             if (interfaceType == null) {
  61.                 return invocation.getMethod();
  62.             }
  63.             return interfaceType.getMethod(invocation.getMethod().getName(), invocation.getMethod().getParameterTypes());
  64.         } catch (NoSuchMethodException e) {
  65.             if (interfaceType != null && !"toString".equals(invocation.getMethod().getName())) {
  66.                 LOGGER.warn("no such method '{}' from interface {}", invocation.getMethod().getName(), interfaceType.getName());
  67.             }
  68.             return invocation.getMethod();
  69.         } catch (Exception e) {
  70.             LOGGER.warn("get Method from interface failed", e);
  71.             return invocation.getMethod();
  72.         }
  73.     }
  75.     //get the interface of proxy
  76.     @Nullable
  77.     protected Class<?> getProxyInterface(Object proxyBean) throws Exception {
  78.         if (DubboUtil.isDubboProxyName(proxyBean.getClass().getName())) {
  79.             //dubbo javaassist proxy
  80.             return DubboUtil.getAssistInterface(proxyBean);
  81.         } else {
  82.             //jdk/cglib proxy
  83.             return SpringProxyUtils.getTargetInterface(proxyBean);
  84.         }
  85.     }
  86.     ...
  87. }
复制代码
 
7.Action拦截处理器ActionInterceptorHandler
Action拦截处理器ActionInterceptorHandler的主要工作是：设置业务动作上下文 + 注册分支事务 + 执行目标方法。
5.png
注意：设置业务动作上下文时，会判断执行方法入参中是否有业务动作上下文。注册分支事务前，会从执行方法入参中提取数据设置到业务动作上下文。
  1. //Handler the TCC Participant Aspect : Setting Context, Creating Branch Record
  2. public class ActionInterceptorHandler {
  3.     ...
  4.     public Object proceed(Method method, Object[] arguments, String xid, TwoPhaseBusinessAction businessAction, Callback<Object> targetCallback) throws Throwable {
  5.         //1.设置业务动作上下文
  6.         //Get action context from arguments, or create a new one and then reset to arguments
  7.         BusinessActionContext actionContext = getOrCreateActionContextAndResetToArguments(method.getParameterTypes(), arguments);
  8.         //Set the xid
  9.         actionContext.setXid(xid);
  10.         //Set the action name，我们自己定义的tcc业务动作名称
  11.         String actionName = businessAction.name();
  12.         actionContext.setActionName(actionName);
  13.         //Set the delay report，延迟report上报
  14.         actionContext.setDelayReport(businessAction.isDelayReport());
  16.         //Creating Branch Record
  17.         //2.发起分支事务的注册，注册成功才会获取到一个branchId
  18.         String branchId = doTccActionLogStore(method, arguments, businessAction, actionContext);
  19.         actionContext.setBranchId(branchId);
  20.         //MDC put branchId
  21.         MDC.put(RootContext.MDC_KEY_BRANCH_ID, branchId);
  23.         //save the previous action context
  24.         BusinessActionContext previousActionContext = BusinessActionContextUtil.getContext();
  25.         try {
  26.             //share actionContext implicitly
  27.             BusinessActionContextUtil.setContext(actionContext);
  29.             if (businessAction.useTCCFence()) {
  30.                 try {
  31.                     //Use TCC Fence, and return the business result
  32.                     return TCCFenceHandler.prepareFence(xid, Long.valueOf(branchId), actionName, targetCallback);
  33.                 } catch (SkipCallbackWrapperException | UndeclaredThrowableException e) {
  34.                     Throwable originException = e.getCause();
  35.                     if (originException instanceof FrameworkException) {
  36.                         LOGGER.error("[{}] prepare TCC fence error: {}", xid, originException.getMessage());
  37.                     }
  38.                     throw originException;
  39.                 }
  40.             } else {
  41.                 //Execute business, and return the business result
  42.                 //3.执行目标方法
  43.                 return targetCallback.execute();
  44.             }
  45.         } finally {
  46.             try {
  47.                 //to report business action context finally if the actionContext.getUpdated() is true
  48.                 BusinessActionContextUtil.reportContext(actionContext);
  49.             } finally {
  50.                 if (previousActionContext != null) {
  51.                     //recovery the previous action context
  52.                     BusinessActionContextUtil.setContext(previousActionContext);
  53.                 } else {
  54.                     //clear the action context
  55.                     BusinessActionContextUtil.clear();
  56.                 }
  57.             }
  58.         }
  59.     }
  60.     ...
  61.    
  62.     //Get or create action context, and reset to arguments
  63.     @Nonnull
  64.     protected BusinessActionContext getOrCreateActionContextAndResetToArguments(Class<?>[] parameterTypes, Object[] arguments) {
  65.         BusinessActionContext actionContext = null;
  66.         //get the action context from arguments
  67.         int argIndex = 0;
  68.         //遍历方法调用时传入的参数类型
  69.         for (Class<?> parameterType : parameterTypes) {
  70.             //如果某个参数类型是BusinessActionContext，因为prepare方法是可以接收一个BusinessActionContext类型的入参的
  71.             if (BusinessActionContext.class.isAssignableFrom(parameterType)) {
  72.                 //尝试获取这个位置的参数对象，但基本是空的
  73.                 actionContext = (BusinessActionContext) arguments[argIndex];
  74.                 if (actionContext == null) {
  75.                     //If the action context exists in arguments but is null, create a new one and reset the action context to the arguments
  76.                     //创建一个BusinessActionContext对象，把一个空的上下文传递到方法入参里
  77.                     actionContext = new BusinessActionContext();
  78.                     arguments[argIndex] = actionContext;
  79.                 } else {
  80.                     //Reset the updated, avoid unnecessary reporting
  81.                     actionContext.setUpdated(null);
  82.                 }
  83.                 break;
  84.             }
  85.             argIndex++;
  86.         }
  87.         //if null, create a new one
  88.         if (actionContext == null) {
  89.             actionContext = new BusinessActionContext();
  90.         }
  91.         return actionContext;
  92.     }
  93.     ...
  94.    
  95.     //Creating Branch Record
  96.     protected String doTccActionLogStore(Method method, Object[] arguments, TwoPhaseBusinessAction businessAction, BusinessActionContext actionContext) {
  97.         String actionName = actionContext.getActionName();
  98.         String xid = actionContext.getXid();
  100.         //region fetch context and init action context
  102.         //从方法入参里提取出来一些数据放入到上下文里去
  103.         Map<String, Object> context = fetchActionRequestContext(method, arguments);
  104.         context.put(Constants.ACTION_START_TIME, System.currentTimeMillis());
  106.         //Init business context
  107.         initBusinessContext(context, method, businessAction);
  108.         //Init running environment context
  109.         initFrameworkContext(context);
  111.         Map<String, Object> originContext = actionContext.getActionContext();
  112.         if (CollectionUtils.isNotEmpty(originContext)) {
  113.             //Merge context and origin context if it exists.
  114.             //@since: above 1.4.2
  115.             originContext.putAll(context);
  116.             context = originContext;
  117.         } else {
  118.             actionContext.setActionContext(context);
  119.         }
  121.         //endregion
  123.         //Init applicationData
  124.         Map<String, Object> applicationContext = Collections.singletonMap(Constants.TCC_ACTION_CONTEXT, context);
  125.         String applicationContextStr = JSON.toJSONString(applicationContext);
  126.         try {
  127.             //registry branch record
  128.             //分支事务注册
  129.             Long branchId = DefaultResourceManager.get().branchRegister(BranchType.TCC, actionName, null, xid, applicationContextStr, null);
  130.             return String.valueOf(branchId);
  131.         } catch (Throwable t) {
  132.             String msg = String.format("TCC branch Register error, xid: %s", xid);
  133.             LOGGER.error(msg, t);
  134.             throw new FrameworkException(t, msg);
  135.         }
  136.     }
  137.     ...
  138.    
  139.     //Extracting context data from parameters, add them to the context
  140.     protected Map<String, Object> fetchActionRequestContext(Method method, Object[] arguments) {
  141.         Map<String, Object> context = new HashMap<>(8);
  143.         Annotation[][] parameterAnnotations = method.getParameterAnnotations();
  144.         //方法的入参也是可以添加注解的
  145.         for (int i = 0; i < parameterAnnotations.length; i++) {
  146.             for (int j = 0; j < parameterAnnotations[i].length; j++) {
  147.                 //如果某个入参添加了一个@BusinessActionContextParameter注解
  148.                 if (parameterAnnotations[i][j] instanceof BusinessActionContextParameter) {
  149.                     //get annotation
  150.                     BusinessActionContextParameter annotation = (BusinessActionContextParameter) parameterAnnotations[i][j];
  151.                     if (arguments[i] == null) {
  152.                         throw new IllegalArgumentException("@BusinessActionContextParameter 's params can not null");
  153.                     }
  155.                     //get param
  156.                     Object paramObject = arguments[i];
  157.                     if (paramObject == null) {
  158.                         continue;
  159.                     }
  161.                     //load param by the config of annotation, and then put into the context
  162.                     //根据注解的配置提取入参的名称和值，把这个名称和值放入到BusinessActionContext里
  163.                     ActionContextUtil.loadParamByAnnotationAndPutToContext(ParamType.PARAM, "", paramObject, annotation, context);
  164.                 }
  165.             }
  166.         }
  167.         return context;
  168.     }
  169. }
复制代码
 
8.Seata TCC分布式事务的注册提交回滚处理源码
(1)TCC分支事务的注册
(2)TCC分支事务的提交
(3)TCC分支事务的回滚
6.png
从TCC的分支事务注册提交回滚过程可知：TCC和AT是可以混合使用的。
 
(1)TCC分支事务的注册
Action拦截处理器在注册TCC的分支事务时，会调用DefaultResourceManager的branchRegister()方法。
  1. public class DefaultResourceManager implements ResourceManager {
  2.     //all resource managers
  3.     protected static Map<BranchType, ResourceManager> resourceManagers = new ConcurrentHashMap<>();
  4.    
  5.     private static class SingletonHolder {
  6.         private static DefaultResourceManager INSTANCE = new DefaultResourceManager();
  7.     }
  8.    
  9.     public static DefaultResourceManager get() {
  10.         return SingletonHolder.INSTANCE;
  11.     }
  12.    
  13.     private DefaultResourceManager() {
  14.         initResourceManagers();
  15.     }
  16.    
  17.     protected void initResourceManagers() {
  18.         //init all resource managers
  19.         //通过SPI加载所有的ResourceManager资源管理器
  20.         //比如：DataSourceManager、TCCResourceManager、SagaResourceManager、ResourceManagerXA
  21.         List<ResourceManager> allResourceManagers = EnhancedServiceLoader.loadAll(ResourceManager.class);
  22.         if (CollectionUtils.isNotEmpty(allResourceManagers)) {
  23.             for (ResourceManager rm : allResourceManagers) {
  24.                 resourceManagers.put(rm.getBranchType(), rm);
  25.             }
  26.         }
  27.     }
  28.     ...
  29.    
  30.     //注册分支事务
  31.     @Override
  32.     public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException {
  33.         return getResourceManager(branchType).branchRegister(branchType, resourceId, clientId, xid, applicationData, lockKeys);
  34.     }
  35.    
  36.     public ResourceManager getResourceManager(BranchType branchType) {
  37.         ResourceManager rm = resourceManagers.get(branchType);
  38.         if (rm == null) {
  39.             throw new FrameworkException("No ResourceManager for BranchType:" + branchType.name());
  40.         }
  41.         return rm;
  42.     }
  43.     ...
  44. }
  46. public class TCCResourceManager extends AbstractResourceManager {
  47.     ...
  48.     ...
  49. }
  51. public abstract class AbstractResourceManager implements ResourceManager {
  52.     //registry branch record
  53.     @Override
  54.     public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException {
  55.         try {
  56.             BranchRegisterRequest request = new BranchRegisterRequest();
  57.             request.setXid(xid);//xid是全局事务id
  58.             request.setLockKey(lockKeys);//这次分支事务要更新数据全局锁keys
  59.             request.setResourceId(resourceId);//分支事务对应的资源id
  60.             request.setBranchType(branchType);//分支事务类型
  61.             request.setApplicationData(applicationData);//应用数据
  63.             BranchRegisterResponse response = (BranchRegisterResponse) RmNettyRemotingClient.getInstance().sendSyncRequest(request);
  64.             if (response.getResultCode() == ResultCode.Failed) {
  65.                 throw new RmTransactionException(response.getTransactionExceptionCode(), String.format("Response[ %s ]", response.getMsg()));
  66.             }
  67.             return response.getBranchId();
  68.         } catch (TimeoutException toe) {
  69.             throw new RmTransactionException(TransactionExceptionCode.IO, "RPC Timeout", toe);
  70.         } catch (RuntimeException rex) {
  71.             throw new RmTransactionException(TransactionExceptionCode.BranchRegisterFailed, "Runtime", rex);
  72.         }
  73.     }
  74.     ...
  75. }
复制代码
(2)TCC分支事务的提交
  1. public class TCCResourceManager extends AbstractResourceManager {
  2.     ...
  3.     //TCC branch commit，分支事务的提交
  4.     @Override
  5.     public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId, String applicationData) throws TransactionException {
  6.         TCCResource tccResource = (TCCResource) tccResourceCache.get(resourceId);
  7.         if (tccResource == null) {
  8.             throw new ShouldNeverHappenException(String.format("TCC resource is not exist, resourceId: %s", resourceId));
  9.         }
  10.         //获取到目标的Bean
  11.         Object targetTCCBean = tccResource.getTargetBean();
  12.         //获取到目标Bean的commit方法
  13.         Method commitMethod = tccResource.getCommitMethod();
  14.         if (targetTCCBean == null || commitMethod == null) {
  15.             throw new ShouldNeverHappenException(String.format("TCC resource is not available, resourceId: %s", resourceId));
  16.         }
  17.         try {
  18.             //BusinessActionContext
  19.             BusinessActionContext businessActionContext = getBusinessActionContext(xid, branchId, resourceId, applicationData);
  20.             Object[] args = this.getTwoPhaseCommitArgs(tccResource, businessActionContext);
  21.             Object ret;
  22.             boolean result;
  23.             //add idempotent and anti hanging
  24.             if (Boolean.TRUE.equals(businessActionContext.getActionContext(Constants.USE_TCC_FENCE))) {
  25.                 try {
  26.                     result = TCCFenceHandler.commitFence(commitMethod, targetTCCBean, xid, branchId, args);
  27.                 } catch (SkipCallbackWrapperException | UndeclaredThrowableException e) {
  28.                     throw e.getCause();
  29.                 }
  30.             } else {
  31.                 ret = commitMethod.invoke(targetTCCBean, args);
  32.                 if (ret != null) {
  33.                     if (ret instanceof TwoPhaseResult) {
  34.                         result = ((TwoPhaseResult) ret).isSuccess();
  35.                     } else {
  36.                         result = (boolean) ret;
  37.                     }
  38.                 } else {
  39.                     result = true;
  40.                 }
  41.             }
  42.             LOGGER.info("TCC resource commit result : {}, xid: {}, branchId: {}, resourceId: {}", result, xid, branchId, resourceId);
  43.             return result ? BranchStatus.PhaseTwo_Committed : BranchStatus.PhaseTwo_CommitFailed_Retryable;
  44.         } catch (Throwable t) {
  45.             String msg = String.format("commit TCC resource error, resourceId: %s, xid: %s.", resourceId, xid);
  46.             LOGGER.error(msg, t);
  47.             return BranchStatus.PhaseTwo_CommitFailed_Retryable;
  48.         }
  49.     }
  50.     ...
  51. }
复制代码
(3)TCC分支事务的回滚
  1. public class TCCResourceManager extends AbstractResourceManager {
  2.     ...
  3.     //TCC branch rollback，分支事务的回滚
  4.     @Override
  5.     public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId, String applicationData) throws TransactionException {
  6.         TCCResource tccResource = (TCCResource) tccResourceCache.get(resourceId);
  7.         if (tccResource == null) {
  8.             throw new ShouldNeverHappenException(String.format("TCC resource is not exist, resourceId: %s", resourceId));
  9.         }
  10.         Object targetTCCBean = tccResource.getTargetBean();
  11.         Method rollbackMethod = tccResource.getRollbackMethod();
  12.         if (targetTCCBean == null || rollbackMethod == null) {
  13.             throw new ShouldNeverHappenException(String.format("TCC resource is not available, resourceId: %s", resourceId));
  14.         }
  15.         try {
  16.             //BusinessActionContext
  17.             BusinessActionContext businessActionContext = getBusinessActionContext(xid, branchId, resourceId, applicationData);
  18.             Object[] args = this.getTwoPhaseRollbackArgs(tccResource, businessActionContext);
  19.             Object ret;
  20.             boolean result;
  21.             //add idempotent and anti hanging
  22.             if (Boolean.TRUE.equals(businessActionContext.getActionContext(Constants.USE_TCC_FENCE))) {
  23.                 try {
  24.                     result = TCCFenceHandler.rollbackFence(rollbackMethod, targetTCCBean, xid, branchId, args, tccResource.getActionName());
  25.                 } catch (SkipCallbackWrapperException | UndeclaredThrowableException e) {
  26.                     throw e.getCause();
  27.                 }
  28.             } else {
  29.                 ret = rollbackMethod.invoke(targetTCCBean, args);
  30.                 if (ret != null) {
  31.                     if (ret instanceof TwoPhaseResult) {
  32.                         result = ((TwoPhaseResult) ret).isSuccess();
  33.                     } else {
  34.                         result = (boolean) ret;
  35.                     }
  36.                 } else {
  37.                     result = true;
  38.                 }
  39.             }
  40.             LOGGER.info("TCC resource rollback result : {}, xid: {}, branchId: {}, resourceId: {}", result, xid, branchId, resourceId);
  41.             return result ? BranchStatus.PhaseTwo_Rollbacked : BranchStatus.PhaseTwo_RollbackFailed_Retryable;
  42.         } catch (Throwable t) {
  43.             String msg = String.format("rollback TCC resource error, resourceId: %s, xid: %s.", resourceId, xid);
  44.             LOGGER.error(msg, t);
  45.             return BranchStatus.PhaseTwo_RollbackFailed_Retryable;
  46.         }
  47.     }
  48.     ...
  49. }
复制代码
 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
匡菲
2025-6-3 00:02:07

0

粉丝关注

12

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫