分布式事务之——tcc-transaction分布式TCC型事务框架搭建与实战案例(基于Dubbo/Dubbox)

时间：2023-11-16 15:36:38

转载请注明出处：http://blog.****.net/l1028386804/article/details/73731363

一、背景

有一定分布式开发经验的朋友都知道，产品/项目/系统最初为了能够快速迭代上线，往往不太注重产品/项目/系统的高可靠性、高性能与高扩展性，采用单体应用和单实例数据库的架构方式快速迭代开发；当产品/项目/系统做到一定规模的时候，原有的系统架构则不足以支撑义务发展需要，往往相同的业务则需要重复写很多次，导致代码大量冗余，难以维护和扩展，这时不得不对原有产品/项目/系统进行拆分，引入分布式的系统架构；而对原有产品/项目/系统进行拆分的过程中，对于业务和数据的拆分和迁移则成为了最为棘手的问题，尤其是在原有业务不能下线，拆分后的业务同时上线的场景下这种问题更加突出；项目拆分后，业务被拆分为多个独立的子业务分散到多个子系统中，而原有的单一数据库则被拆分到多个数据库中，拆分后的数据库则同样又面临着让人头疼的分布式事务的问题。

本文就针对项目拆分后数据库的分布式事务问题，基于tcc-transaction分布式TCC型事务进行框架的搭建，同时引入相关的实战案例，来解决让人头疼的分布式事务问题。

二、tcc-transaction框架介绍

介绍：tcc-transaction是开源的TCC补偿性分布式事务框架，Git地址：https://github.com/changmingxie/tcc-transaction
TCC为Try、Confirm、Cancel的缩写：try阶段预留资源尝试提交，confirm阶段确定提交，cancel取消提交释放资源。
1.2.x项目指南地址：https://github.com/changmingxie/tcc-transaction/wiki/%E4%BD%BF%E7%94%A8%E6%8C%87%E5%8D%971.2.x
本文的例子为引入一个本人实际工作中的一个开发场景：创建资产，将资产信息同时同步到Mongo与ES的流程(ES代码不列出了，与mongo类似)，整个流程保证数据一致

三、项目流程

1.下载1.2.x版本源码，并可能需要修改部分代码

因为是第三方包，所以需要自己打包到本地仓库。但包中spring版本为3.2.12.RELEASE，如果本地项目为4.x，比如本人的项目spring版本为4.3.4.RELEASE，如果不修改tcc中的spring版本，将报错无法启动，所以需要对原有框架源码进行相应的修改。
源码修改比较简单，如下

1.1 修改tcc-transaction总pom.xml文件
  1. <!-- 第一处:修改版本为4.3.4 -->
  2. <springframework.version>4.3.4.RELEASE</springframework.version>
  4. <!-- 第二处:修改版本为2.2.1 -->
  5. <dependency>
  6. <groupId>org.quartz-scheduler</groupId>
  7. <artifactId>quartz</artifactId>
  8. <version>2.2.1</version>
  9. <exclusions>
  10. <exclusion>
  11. <groupId>c3p0</groupId>
  12. <artifactId>c3p0</artifactId>
  13. </exclusion>
  14. </exclusions>
  15. </dependency>
  17. <!-- 第三处:修改版本为2.5.3 -->
  18. <dependency>
  19. <groupId>com.alibaba</groupId>
  20. <artifactId>dubbo</artifactId>
  21. <version>2.5.3</version>
  22. </dependency>
1.2 修改 tcc-transaction-spring/src/main/java/org/mengyun/tcctransaction/spring/recover/RecoverScheduledJob.java

该文件中 CronTriggerBean类在4.x中已经不存在，也是修改源码主要修改的地方。
修改其中的init方法，修改后如下：

  1. public void init() {
  2. try {
  3. MethodInvokingJobDetailFactoryBean jobDetail = new MethodInvokingJobDetailFactoryBean();
  4. jobDetail.setTargetObject(transactionRecovery);
  5. jobDetail.setTargetMethod("startRecover");
  6. jobDetail.setName("transactionRecoveryJob");
  7. jobDetail.setConcurrent(false);
  8. jobDetail.afterPropertiesSet();
  10. CronTriggerFactoryBean cronTrigger = new CronTriggerFactoryBean();
  11. cronTrigger.setBeanName("transactionRecoveryCronTrigger");
  12. cronTrigger.setJobDetail(jobDetail.getObject());
  14. cronTrigger.setCronExpression(transactionConfigurator.getRecoverConfig().getCronExpression());
  15. cronTrigger.afterPropertiesSet();
  17. scheduler.scheduleJob(jobDetail.getObject(), cronTrigger.getObject());
  19. scheduler.start();
  21. } catch (Exception e) {
  22. throw new SystemException(e);
  23. }
  24. }

各位也可参考如下的修改：https://github.com/changmingxie/tcc-transaction/pull/84/files

1.3 打包并发布

这里我们通过Maven进行打包发布，命令为：

mvn -Dmaven.test.skip=true install

2.项目依赖

参考1.2.x使用指南，引入两个依赖(本人项目dubbo/dubbox框架，我使用并打包时版本为1.2.3.1)。调用方和提供方都需要引入依赖。

3.加载tcc-transaction.xml配置

原文中是配置在web.xml中，我个人试了一下放在dubbo web项目的web.xml中，但配置并没有被加载。该文件的意义只是希望项目启动时被加载，于是直接在dubbo中的一个spring的配置文件中引入，如下：

  1. <!-- TCC Transaction -->
  2. <import resource="classpath:tcc-transaction.xml" />

该文件里面提供各种aop逻辑，项目启动时扫描指定注解，并做增强。

4.设置TransactionRepository

需要为tcc配置数据源，可以是MySQL或其他nosql，本文使用mysql，其他可参见原指南文档。
mysql配置如下：

  1. <!--tcc-->
  2. <bean id="tccDataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
  3. <property name="driverClassName" value="${jdbc.driverClassName}" />
  4. <property name="url" value="${jdbc.tcc.url}" />
  5. <property name="username" value="${jdbc.username}" />
  6. <property name="password" value="${jdbc.password}" />
  7. <property name="initialSize" value="${dbcp.initialSize}" />
  8. <property name="maxActive" value="${dbcp.maxActive}" />
  9. <property name="maxIdle" value="${dbcp.maxIdle}" />
  10. <property name="maxWait" value="${dbcp.maxWait}" />
  11. <property name="poolPreparedStatements" value="${dbcp.poolPreparedStatements}" />
  12. <property name="defaultAutoCommit" value="${dbcp.defaultAutoCommit}" />
  13. <property name="timeBetweenEvictionRunsMillis" value="${dbcp.timeBetweenEvictionRunsMillis}" />
  14. <property name="minEvictableIdleTimeMillis" value="${dbcp.minEvictableIdleTimeMillis}" />
  15. </bean>
  17. <bean id="transactionRepository"
  18. class="org.mengyun.tcctransaction.spring.repository.SpringJdbcTransactionRepository">
  19. <property name="dataSource" ref="tccDataSource"/>
  20. <property name="domain" value="SAAS"/>
  21. <property name="tbSuffix" value="_ASSET"/>
  22. </bean>
  24. <bean class="org.mengyun.tcctransaction.spring.recover.DefaultRecoverConfig">
  25. <property name="maxRetryCount" value="30"/>
  26. <property name="recoverDuration" value="120"/>
  27. <property name="cronExpression" value="0 */1 * * * ?"/>
  28. <property name="delayCancelExceptions">
  29. <util:set>
  30. <value>com.alibaba.dubbo.remoting.TimeoutException</value>
  31. </util:set>
  32. </property>
  33. </bean>

需要注意的点：
1.数据源必须配置新的，不能使用之前项目存在的dataSource的bean，也不能在同一库中，不然会导致tcc表数据与本地事务一起回滚，从而无法保存异常事务日志；
2.注意domain、tbSuffix的配置，这两项文档中并没有配置，但源码demo中配置了，用于数据库的表名称等，推荐配置；
3.最后的DefaultRecoverConfig项是可选的，用于恢复与重试，具体作用参考使用指南；
4.defaultAutoCommit必须为true(默认为true)

5.mysql建表脚本

根据以上的tbSufifix配置，脚本如下：

  1. CREATE TABLE `tcc_transaction_asset` (
  2. `TRANSACTION_ID` int(11) NOT NULL AUTO_INCREMENT,
  3. `DOMAIN` varchar(100) DEFAULT NULL,
  4. `GLOBAL_TX_ID` varbinary(32) NOT NULL,
  5. `BRANCH_QUALIFIER` varbinary(32) NOT NULL,
  6. `CONTENT` varbinary(8000) DEFAULT NULL,
  7. `STATUS` int(11) DEFAULT NULL,
  8. `TRANSACTION_TYPE` int(11) DEFAULT NULL,
  9. `RETRIED_COUNT` int(11) DEFAULT NULL,
  10. `CREATE_TIME` datetime DEFAULT NULL,
  11. `LAST_UPDATE_TIME` datetime DEFAULT NULL,
  12. `VERSION` int(11) DEFAULT NULL,
  13. PRIMARY KEY (`TRANSACTION_ID`),
  14. UNIQUE KEY `UX_TX_BQ` (`GLOBAL_TX_ID`,`BRANCH_QUALIFIER`)
  15. ) ENGINE=InnoDB DEFAULT CHARSET=utf8

如果表名称不对，启动过程会报错，这时，对数据表做相应调整即可。

6.发布服务（重点）

6.1 dubbo接口
  1. /**
  2. * @author liuyazhuang
  3. * 资产相关的业务发布Dubbo服务
  4. */
  5. public interface AssetCardService {
  7. /**
  8. * 测试预保存资产(状态为待确认)
  9. */
  10. @Compensable
  11. int testSaveAssetCard(AssetCardModel model);
  13. /**
  14. * 确认保存资产到mysql(状态为正常)
  15. */
  16. int confirmMysqlSaveAssetCard(AssetCardModel model);
  18. /**
  19. * 取消保存资产到msyql(更新状态为删除)
  20. */
  21. int cancelMysqlSaveAssetCard(AssetCardModel model);
  23. /**
  24. * 预保存资产到mongo(状态为待确认)
  25. */
  26. @Compensable
  27. void processMongo(AssetCardModel model);
  29. /**
  30. * 确认保存资产到mongo(状态为正常)
  31. */
  32. void confirmMongoSaveAssetCard(AssetCardModel model);
  34. /**
  35. * 取消保存资产到mongo(更新状态为删除)
  36. */
  37. void cancelMongoSaveAssetCard(AssetCardModel model);
  39. }

需要注意的点：
1.对外提供服务的接口必须有@Compensable注解；
2.对应的confirm与cancel方法必须声明为接口，不能声明为private，即使是public也不行，必须有接口。

6.2 dubbo接口实现类
  1. /**
  2. * @author liuyazhuang
  3. * 资产相关的业务发布Dubbo服务的实现
  4. */
  5. @Service
  6. @Component
  7. public class AssetCardServiceImpl implements AssetCardService{
  8. @Override
  9. @Compensable(confirmMethod = "confirmMysqlSaveAssetCard", cancelMethod = "cancelMysqlSaveAssetCard", transactionContextEditor = DubboTransactionContextEditor.class)
  10. @Transactional(propagation = Propagation.REQUIRED, rollbackFor = { Exception.class })
  11. public int testSaveAssetCard(AssetCardModel model){
  13. // 保存mysql，data状态为-1
  14. model.setDataStatus(-1);
  15. assetCardDao.insert(model);
  18. // mongo处理
  19. assetCardService.processMongo(model);
  21. return model.getId();
  22. }
  24. @Override
  25. public int confirmMysqlSaveAssetCard(AssetCardModel model){
  26. System.out.println("============================================================================");
  27. System.out.println("=================mysql:confirm");
  28. System.out.println("============================================================================");
  30. // 更新mysql data_status为0
  31. model.setDataStatus(0);
  32. assetCardDao.updateByPrimaryKey(model);
  34. return model.getId();
  35. }
  37. @Override
  38. public int cancelMysqlSaveAssetCard(AssetCardModel model){
  39. System.out.println("============================================================================");
  40. System.out.println("=================mysql:cancel");
  41. System.out.println("============================================================================");
  43. // 更新mysql data_status为-1
  44. model.setDataStatus(-1);
  45. assetCardDao.updateByPrimaryKey(model);
  47. return model.getId();
  48. }
  50. @Compensable(confirmMethod = "confirmMongoSaveAssetCard", cancelMethod = "cancelMongoSaveAssetCard", transactionContextEditor = DubboTransactionContextEditor.class)
  51. @Override
  52. public void processMongo(AssetCardModel model) {
  54. // 保存mongo，data_statu为-1
  55. model.setDataStatus(-1);
  56. assetCardDaoWrapper.saveMongo(model);
  57. }
  59. @Override
  60. public void confirmMongoSaveAssetCard(AssetCardModel model){
  61. System.out.println("============================================================================");
  62. System.out.println("=================mongo:confirm");
  63. System.out.println("============================================================================");
  65. // 更新mongo data_status为0
  66. model.setDataStatus(0);
  67. assetCardDaoWrapper.updateMongo(model);
  68. }
  70. @Override
  71. public void cancelMongoSaveAssetCard(AssetCardModel model){
  72. System.out.println("============================================================================");
  73. System.out.println("=================mongo:cancel");
  74. System.out.println("============================================================================");
  76. // 更新mongo data_status为-1
  77. model.setDataStatus(-1);
  78. assetCardDao.updateByPrimaryKey(model);
  79. assetCardDaoWrapper.updateMongo(model);
  80. }
  81. }

注意点：
1.对外提供服务的接口必须有@Compensable注解，同时必须有confirmMethod、cancelMethod参数的配置，同时dubbo接口额外增加
"transactionContextEditor = DubboTransactionContextEditor.class"这个配置；
2.提供服务接口与对应另外的两个CC方法参数必须完全一致；
3.该tcc框架可嵌套调用，如上在testSaveAssetCard方法，即try阶段中调用了另一个tcc方法"assetCardService.processMongo()"，理论上嵌套只应该在try阶段进行；
4.confirm、cancel需要实现幂等性，可能会被重试；5.由于网络等因素，可能导致cancel方法先执行，cancel方法一定要做好相应的判断与处理

6.3 调用方
  1. @Override
  2. @Transactional(propagation = Propagation.REQUIRED, rollbackFor = { Exception.class })
  3. public long testSaveAssetCard(AssetCardModel assetCardModel) throws AssetException {
  4. assetCardModel.setId(IdGenerator.getId());
  5. return assetCardService.testSaveAssetCard(assetCardModel);
  6. }

注意点：
1.因为需要回滚更新等操作，所以此业务中id不能用自增，而是需要项目生成；
2.特别注意，调用方必须在事务中，也就是说必须有事务注解，或者能被事务配置切到，没有事务tcc框架调用时会抛异常。
至此，配置已经全部完成。

7.事务查看

源码中提供tcc-transaction-server web项目，该项目提供界面查看事务日志，打包后部署即可，我们这里就不在作详细的描述。

四、TCC执行流程

业务流程使用记录：
前提：用户下单，建立订单，创建支付记录，支付记录状态为待支付
try：
用户金额冻结
调用积分处理TCC：
try：预增加积分
confirm：更新增加积分状态
cancel：取消增加的积分
confirm：
订单支付状态更新为已支付
订单支付记录支付状态更新为已支付
用户金额扣款（以上三个操作在同一本地事务）
cancel：
判断订单支付状态与订单记录支付状态为未支付
用户冻结金额释放

相关文章