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

简介: 由于系统演进,大佬觉得 需要做微服务,脑子一拍开始对原来的系统进行微服务改造, 在改造过程中,分布式事务不可避免,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 尝试接入项目
先放一张流程图
image

接入流程

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中的数据

相关文章
|
存储 Java 关系型数据库
在Spring Boot中整合Seata框架实现分布式事务
可以在 Spring Boot 中成功整合 Seata 框架,实现分布式事务的管理和处理。在实际应用中,还需要根据具体的业务需求和技术架构进行进一步的优化和调整。同时,要注意处理各种可能出现的问题,以保障分布式事务的顺利执行。
1537 160
|
消息中间件 监控 数据可视化
Apache Airflow 开源最顶级的分布式工作流平台
Apache Airflow 是一个用于创作、调度和监控工作流的平台,通过将工作流定义为代码,实现更好的可维护性和协作性。Airflow 使用有向无环图(DAG)定义任务,支持动态生成、扩展和优雅的管道设计。其丰富的命令行工具和用户界面使得任务管理和监控更加便捷。适用于静态和缓慢变化的工作流,常用于数据处理。
Apache Airflow 开源最顶级的分布式工作流平台
|
人工智能 监控 开发者
阿里云PAI发布DeepRec Extension,打造稳定高效的分布式训练,并宣布开源!
阿里云PAI发布DeepRec Extension,打造稳定高效的分布式训练,并宣布开源!
357 0
|
自然语言处理 负载均衡 API
gRPC 一种现代、开源、高性能的远程过程调用 (RPC) 可以在任何地方运行的框架
gRPC 是一种现代开源高性能远程过程调用(RPC)框架,支持多种编程语言,可在任何环境中运行。它通过高效的连接方式,支持负载平衡、跟踪、健康检查和身份验证,适用于微服务架构、移动设备和浏览器客户端连接后端服务等场景。gRPC 使用 Protocol Buffers 作为接口定义语言,支持四种服务方法:一元 RPC、服务器流式处理、客户端流式处理和双向流式处理。
|
Dubbo 网络协议 Java
RPC框架:一文带你搞懂RPC
这篇文章全面介绍了RPC(远程过程调用)的概念、原理和应用场景,解释了RPC如何工作以及为什么在分布式系统中广泛使用,并探讨了几种常用的RPC框架如Thrift、gRPC、Dubbo和Spring Cloud,同时详细阐述了RPC调用流程和实现透明化远程服务调用的关键技术,包括动态代理和消息的编码解码过程。
RPC框架:一文带你搞懂RPC
|
XML 负载均衡 监控
分布式-dubbo-简易版的RPC框架
分布式-dubbo-简易版的RPC框架
|
存储 关系型数据库 MySQL
深度评测:PolarDB-X 开源分布式数据库的优势与实践
本文对阿里云开源分布式数据库 PolarDB-X 进行了详细评测。PolarDB-X 以其高性能、强可用性和出色的扩展能力在云原生数据库市场中脱颖而出。文章首先介绍了 PolarDB-X 的核心产品优势,包括金融级高可靠性、海量数据处理能力和高效的混合负载处理能力。随后,分析了其分布式架构设计,包括计算节点、存储节点、元数据服务和日志节点的功能分工。评测还涵盖了在 Windows 平台通过 WSL 环境部署 PolarDB-X 的过程,强调了环境准备和工具安装的关键步骤。使用体验方面,PolarDB-X 在处理分布式事务和实时分析时表现稳定,但在网络问题和性能瓶颈上仍需优化。最后,提出了改进建
|
XML 存储 JSON
(十二)探索高性能通信与RPC框架基石:Json、ProtoBuf、Hessian序列化详解
如今这个分布式风靡的时代,网络通信技术,是每位技术人员必须掌握的技能,因为无论是哪种分布式技术,都离不开心跳、选举、节点感知、数据同步……等机制,而究其根本,这些技术的本质都是网络间的数据交互。正因如此,想要构建一个高性能的分布式组件/系统,不得不思考一个问题:怎么才能让数据传输的速度更快?
1232 1
|
负载均衡 Dubbo Java
Dubbo 3.x:探索阿里巴巴的开源RPC框架新技术
随着微服务架构的兴起,远程过程调用(RPC)框架成为了关键组件。Dubbo,作为阿里巴巴的开源RPC框架,已经演进到了3.x版本,带来了许多新特性和技术改进。本文将探讨Dubbo 3.x中的一些最新技术,包括服务注册与发现、负载均衡、服务治理等,并通过代码示例展示其使用方式。
947 9
|
设计模式 负载均衡 网络协议
【分布式技术专题】「分布式技术架构」实践见真知,手把手教你如何实现一个属于自己的RPC框架(架构技术引导篇)
【分布式技术专题】「分布式技术架构」实践见真知,手把手教你如何实现一个属于自己的RPC框架(架构技术引导篇)
1005 0