阿里开源分布式事务框架seata实践(原fescar) springboot +durid+mybitas+自有rpc框架

本文章仅作为seata接入文档，seata原理和源码请自行转至github https://github.com/seata/seata
官方文档地址https://github.com/seata/seata/wiki/Home_Chinese

1 由于系统演进，大佬觉得 需要做微服务，脑子一拍开始对原来的系统进行微服务改造，
在改造过程中，分布式事务不可避免，tcc mq等等概念研究一遍后,结合时间成本，发现阿里gts 比较适合，无奈需要接入外网遂放弃，后来偶然发现seata 开源版gts 尝试接入项目
先放一张流程图

接入流程

1 首先去官网git 下载一份源码，我下载的是0.5.2版本。
2 在本地解压加载到idea下载jar包后直接启动server项目中的启动类Server.java ，在调试过程中发现netty存在有时内存不够问题，遂增加启动参数-XX:MaxDirectMemorySize=1024m
3 server 基于netty开发目前只支持单节点启动，内存大小没有进行压力测试，seata配置文件为registry.conf 附上关键配置

registry { # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa type = "file" file { name = "file.conf" } } config { # file、nacos 、apollo、zk、consul type = "file" file { name = "file.conf" } } 

type 指定配置方式 目前支持file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
默认配置为file 在接入过程中也使用file 配置类型

file.conf 和registry.conf 目前都在resource目录下
附上file.conf 关键配置

service { #vgroup->rgroup vgroup_mapping.my_test_tx_group = "default" #only support single node default.grouplist = "127.0.0.1:8091" #degrade current not support enableDegrade = false #disable disable = false } 

接入过程 vgroup_mapping.my_test_tx_group = "default" 修改配置为自定义配置vgroup_mapping.my_group= "default"
具体原则不清楚

4 架构图中的tc搭建完成，下一步搭建RM 也就是微服务的原子系统 引入seata 的jar包

<!--框架问题，指定durid版本--> <properties> <druid.version>1.1.10</druid.version> <seata.version>0.5.0</seata.version> </properties> <dependency> <groupId>io.seata</groupId> <artifactId>seata-spring</artifactId> <version>${seata.version}</version> </dependency> 

5 ，按照官方文档，需要将数据源替换为seata数据源，本项目是springboot+durid+mybitas 直接上代码 关键代码 DataSourceProxy proxy = new DataSourceProxy(datasource);

@Configuration public class DruidConfig { @Value("${spring.datasource.url}") private String dbUrl; @Value("${spring.datasource.username}") private String username; @Value("${spring.datasource.password}") private String password; @Value("${spring.datasource.driver-class-name}") private String driverClassName; @Value("${spring.datasource.initialSize}") private int initialSize; @Value("${spring.datasource.minIdle}") private int minIdle; @Value("${spring.datasource.maxActive}") private int maxActive; @Value("${spring.datasource.maxWait}") private int maxWait; @Value("${spring.datasource.timeBetweenEvictionRunsMillis}") private int timeBetweenEvictionRunsMillis; @Value("${spring.datasource.minEvictableIdleTimeMillis}") private int minEvictableIdleTimeMillis; @Value("${spring.datasource.validationQuery}") private String validationQuery; @Value("${spring.datasource.testWhileIdle}") private boolean testWhileIdle; @Value("${spring.datasource.testOnBorrow}") private boolean testOnBorrow; @Value("${spring.datasource.testOnReturn}") private boolean testOnReturn; @Value("${spring.datasource.poolPreparedStatements}") private boolean poolPreparedStatements; @Value("${spring.datasource.filters}") private String filters; @Value("${mybatis.mapper-locations}") private String mapperLocation; @Bean @Primary public DataSource druidDataSource() { DruidDataSource datasource = new DruidDataSource(); datasource.setUrl(this.dbUrl); datasource.setUsername(username); datasource.setPassword(password); datasource.setDriverClassName(driverClassName); datasource.setInitialSize(initialSize); datasource.setMinIdle(minIdle); datasource.setMaxActive(maxActive); datasource.setMaxWait(maxWait); datasource.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis); datasource.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis); datasource.setValidationQuery(validationQuery); datasource.setTestWhileIdle(testWhileIdle); datasource.setTestOnBorrow(testOnBorrow); datasource.setTestOnReturn(testOnReturn); datasource.setPoolPreparedStatements(poolPreparedStatements); DataSourceProxy proxy = new DataSourceProxy(datasource); return proxy; } @Bean(name="sqlSessionFactory") public SqlSessionFactoryBean sqlSessionFactory(DataSource dataSource) throws Exception { SqlSessionFactoryBean sqlSessionFactory = new SqlSessionFactoryBean(); ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver(); Resource[] mapperXmlResource = resolver.getResources(mapperLocation); sqlSessionFactory.setDataSource(dataSource); sqlSessionFactory.setMapperLocations(mapperXmlResource); return sqlSessionFactory; } }

6 新增seata 扫描器配置 直接上代码

@Configuration public class SeataConfiguration { @Value("${spring.application.name}") private String applicationId; /** * 注册一个StatViewServlet * * @return global transaction scanner */ @Bean public GlobalTransactionScanner globalTransactionScanner() { GlobalTransactionScanner globalTransactionScanner = new GlobalTransactionScanner(applicationId, "my_group"); return globalTransactionScanner; } } 

7
新增拦截器过滤器或者切面等，在业务执行前，拦截每个请求，并获取XID并绑定，本人是在service前增加切面，并处理数据
seata目前支持dubbo和springcloud 默认XID放在headers中，由于我们的项目使用的自有的rpc框架，因此需要自己手动获取XID，为了方便
我将XID写在了body中，自己接入的时候，需要按照需要自行设置
上代码

 String xid = RootContext.getXID(); String restXid = StringUtil.getStringValue(esbInput.getParams().get("Seata-Xid")); boolean bind = false; if (StringUtils.isBlank(xid) && StringUtils.isNotBlank(restXid)) { RootContext.bind(restXid); bind = true; if (LOGGER.isDebugEnabled()) { LOGGER.debug("bind[" + restXid + "] to RootContext"); } } try{ //执行方法体 object = joinPoint.proceed(args); }catch (GeneralException e){//对外接口统一异常捕获解析 }finally { if (bind) { String unbindXid = RootContext.unbind(); if (LOGGER.isDebugEnabled()) { LOGGER.debug("unbind[" + unbindXid + "] from RootContext"); } if (!restXid.equalsIgnoreCase(unbindXid)) { LOGGER.warn("xid in change during http rest from " + restXid + " to " + unbindXid); if (unbindXid != null) { RootContext.bind(unbindXid); LOGGER.warn("bind [" + unbindXid + "] back to RootContext"); } } } }

8 增加配置文件 file.conf 和registry.conf 并按照自己的实际情况进行配置，同时按照官方文档，在原子服务数据库新增日志回退表
undo_log 建表语句为

 DROP TABLE IF EXISTS `undo_log`; CREATE TABLE `undo_log` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `branch_id` bigint(20) NOT NULL, `xid` varchar(100) NOT NULL, `rollback_info` longblob NOT NULL, `log_status` int(11) NOT NULL, `log_created` datetime NOT NULL, `log_modified` datetime NOT NULL, `ext` varchar(100) DEFAULT NULL, PRIMARY KEY (`id`), UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`) ) ENGINE=InnoDB AUTO_INCREMENT=94 DEFAULT CHARSET=utf8;

9 创建TM服务
引入jar包`js

 <seata.version>0.5.0</seata.version>

<groupId>io.seata</groupId> <artifactId>seata-spring</artifactId> <version>${seata.version}</version>

`

10
新增scanner 配置 和rm 一致

@Configuration public class SeataConfiguration { @Value("${spring.application.name}") private String applicationId; /** * 注册一个StatViewServlet * * @return global transaction scanner */ @Bean public GlobalTransactionScanner globalTransactionScanner() { GlobalTransactionScanner globalTransactionScanner = new GlobalTransactionScanner(applicationId, "nguc_tx_group"); return globalTransactionScanner; } } 

11 开启事务和事务的提交回滚

private void transactionalRollBack(GlobalTransaction globalTransaction,String xid){ LOGGER.error("分布式事务中断，事务开始回滚"); try { globalTransaction.rollback(); } catch (TransactionException txe) { LOGGER.error("分布式事务回滚失败，全局事务XID : " + xid); } } public XX doTransaction(){ GlobalTransaction globalTransaction = GlobalTransactionContext.getCurrentOrCreate(); //begin GlobalTransactional try { globalTransaction.begin(20000, "test"); } catch (TransactionException e) { LOGGER.error("全局事务开启失败") return outObject; } String xid = RootContext.getXID(); //组合服务标示 try { call(input) } catch (Exception e) { LOGGER.error(e.getMessage(), e); transactionalRollBack(globalTransaction, xid); } try { globalTransaction.rollback(); LOGGER.error("全局事务提交成功，全局事务XID : " + xid); } catch (TransactionException txe) { LOGGER.error("全局事务提交失败，全局事务XID : " + xid); } return xx; } 

12 通过TM调用rm服务，并测试回滚，可以在commit前添加断点查看undo_log中的数据