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

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

2019-05-30 11110

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

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

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

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

热门文章

最新文章

相关课程

相关电子书

相关实验场景