本文章仅作为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中的数据