分布式事务seata的AT模式介绍-阿里云开发者社区

开发者社区> 川行> 正文

分布式事务seata的AT模式介绍

简介: seata是阿里开源的一款分布式事务解决方案,致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式,本文主要介绍AT模式的使用。
+关注继续查看

seata是阿里开源的一款分布式事务解决方案,致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式,本文主要介绍AT模式的使用。

seata安装

下载seata服务,官方地址:https://github.com/seata/seata/releases
在Linux下,下载完成后,直接解压,通过命令安装即可:

sh ./bin/seata-server.sh

支持的启动参数

参数 全写 作用 备注
-h --host 指定在注册中心注册的 IP 不指定时获取当前的 IP,外部访问部署在云环境和容器中的 server 建议指定
-p --port 指定 server 启动的端口 默认为 8091
-m --storeMode 事务日志存储方式 支持file和db,默认为 file
-n --serverNode 用于指定seata-server节点ID ,如 1,2,3..., 默认为 1
-e --seataEnv 指定 seata-server 运行环境 如 dev, test 等, 服务启动时会使用 registry-dev.conf 这样的配置

如:

sh ./bin/seata-server.sh -p 8091 -h 127.0.0.1 -m file

seata的AT模式介绍

AT模式实质是两阶段提交协议的演变,具体如下:

  • 一阶段:业务数据和回滚日志记录在同一个本地事务中提交,释放本地锁和连接资源
  • 二阶段:
    提交异步化,非常快速地完成。

回滚通过一阶段的回滚日志进行反向补偿。

业务背景:
用户调用系统A的store服务,store服务调用系统B的company服务,company服务会新增一条数据,然后把companyId返回系统A,然后系统A通过companyId再新增一条store数据。

一般如果store服务执行失败了,直接抛异常了,所以company服务也不会执行,
但如果store服务执行成功了,已经写了一条数据到数据库,执行company服务时失败了,就会产生数据不一致的问题。

使用seata的AT模式,主要分为下面几个步骤:

  • 配置seata服务及创建事务表
  • 调用方配置(对应上面的store服务)
  • 服务提供方配置(对应上面的company服务)

配置seata服务及创建事务表

配置conf/file.conf文件

## transaction log store, only used in server side
store {
  ## store mode: file、db
  mode = "db" //修改为db模式,标识事务信息用db存储
  ## file store property
  file {
    ## store location dir
    dir = "sessionStore"
    # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
    maxBranchSessionSize = 16384
    # globe session size , if exceeded throws exceptions
    maxGlobalSessionSize = 512
    # file buffer size , if exceeded allocate new buffer
    fileWriteBufferCacheSize = 16384
    # when recover batch read size
    sessionReloadReadSize = 100
    # async, sync
    flushDiskMode = async
  }

  ## database store property
  db {
    ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    datasource = "druid"
    ## mysql/oracle/postgresql/h2/oceanbase etc.
    dbType = "mysql"
    driverClassName = "com.mysql.cj.jdbc.Driver"
    url = "jdbc:mysql://192.168.234.1:3306/seata?useUnicode=true&characterEncoding=utf8&useSSL=false&&serverTimezone=UTC" //修改数据库连接
    user = "seata" //修改数据库账号
    password = "123456" //修改数据库密码
    minConn = 5
    maxConn = 30
    globalTable = "global_table"
    branchTable = "branch_table"
    lockTable = "lock_table"
    queryLimit = 100
  }
}
## server configuration, only used in server side
service {
  #vgroup->rgroup
  vgroup_mapping.chuanzh_tx_group = "default" //chuanzh_tx_group为自定义的事务组名称,要和客户端配置保持一致
  #only support single node
  default.grouplist = "192.168.234.128:8091"
  #degrade current not support
  enableDegrade = false
  #disable
  disable = false
  #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent
  max.commit.retry.timeout = "-1"
  max.rollback.retry.timeout = "-1"
}

上面配置共修改了3个地方:

  1. 存储模式改为db模式,需要创建3张事务表,如下:

    -- the table to store GlobalSession data
     CREATE TABLE IF NOT EXISTS `global_table`
     (
         `xid`                       VARCHAR(128) NOT NULL,
         `transaction_id`            BIGINT,
         `status`                    TINYINT      NOT NULL,
         `application_id`            VARCHAR(32),
         `transaction_service_group` VARCHAR(32),
         `transaction_name`          VARCHAR(128),
         `timeout`                   INT,
         `begin_time`                BIGINT,
         `application_data`          VARCHAR(2000),
         `gmt_create`                DATETIME,
         `gmt_modified`              DATETIME,
         PRIMARY KEY (`xid`),
         KEY `idx_gmt_modified_status` (`gmt_modified`, `status`),
         KEY `idx_transaction_id` (`transaction_id`)
     ) ENGINE = InnoDB
       DEFAULT CHARSET = utf8;
    
     -- the table to store BranchSession data
     CREATE TABLE IF NOT EXISTS `branch_table`
     (
         `branch_id`         BIGINT       NOT NULL,
         `xid`               VARCHAR(128) NOT NULL,
         `transaction_id`    BIGINT,
         `resource_group_id` VARCHAR(32),
         `resource_id`       VARCHAR(256),
         `branch_type`       VARCHAR(8),
         `status`            TINYINT,
         `client_id`         VARCHAR(64),
         `application_data`  VARCHAR(2000),
         `gmt_create`        DATETIME(6),
         `gmt_modified`      DATETIME(6),
         PRIMARY KEY (`branch_id`),
         KEY `idx_xid` (`xid`)
     ) ENGINE = InnoDB
       DEFAULT CHARSET = utf8;
    
     -- the table to store lock data
     CREATE TABLE IF NOT EXISTS `lock_table`
     (
         `row_key`        VARCHAR(128) NOT NULL,
         `xid`            VARCHAR(96),
         `transaction_id` BIGINT,
         `branch_id`      BIGINT       NOT NULL,
         `resource_id`    VARCHAR(256),
         `table_name`     VARCHAR(32),
         `pk`             VARCHAR(36),
         `gmt_create`     DATETIME,
         `gmt_modified`   DATETIME,
         PRIMARY KEY (`row_key`),
         KEY `idx_branch_id` (`branch_id`)
     ) ENGINE = InnoDB
       DEFAULT CHARSET = utf8;
    
  2. 修改数据库连接,注意如果你安装的是MySQL8,则需要修改MySQL8的驱动:driverClassName = "com.mysql.cj.jdbc.Driver",不然会出现启动报错的问题,详细请参考:seata启动MySQL报错 #359
  3. 修改事务的组名,你也可以不修改,我这里使用的是:chuanzh_tx_group
  4. 创建业务事务表,记录业务需要回滚的数据,在分布式事务中,每个参与的业务数据库都需要添加对应的表

    CREATE TABLE `undo_log` (
      `id` bigint(20) NOT NULL AUTO_INCREMENT,
      `branch_id` bigint(20) NOT NULL,
      `xid` varchar(100) NOT NULL,
      `context` varchar(128) 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=1 DEFAULT CHARSET=utf8;

配置conf/registry.conf文件

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "eureka"  修改注册方式,微服务调用使用的是Eureka

  nacos {
    serverAddr = "localhost"
    namespace = ""
    cluster = "default"
  }
  eureka {
    serviceUrl = "http://192.168.234.1:8081/eureka"  //修改Eureka地址
    application = "default"  
    weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = "0"
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
  }
  consul {
    cluster = "default"
    serverAddr = "127.0.0.1:8500"
  }
  etcd3 {
    cluster = "default"
    serverAddr = "http://localhost:2379"
  }
  sofa {
    serverAddr = "127.0.0.1:9603"
    application = "default"
    region = "DEFAULT_ZONE"
    datacenter = "DefaultDataCenter"
    cluster = "default"
    group = "SEATA_GROUP"
    addressWaitTime = "3000"
  }
  file {
    name = "file.conf"
  }
}

以上修改了使用Eureka方式注册,并配置了Eureka地址,启动MySQL、Eureka服务后,就可以启动seata服务了。

调用方配置(store-server)

maven配置,使用seata-spring-boot-starter,自动配置的方式,不需要再添加file.conf和register.conf文件

    <!--druid-->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid-spring-boot-starter</artifactId>
        <version>${druid-spring-boot-starter.version}</version>
    </dependency>

    <!--seata-->
    <dependency>
        <groupId>io.seata</groupId>
        <artifactId>seata-spring-boot-starter</artifactId>
        <version>1.2.0</version>
    </dependency>

application.properties配置:

server.port=9090
spring.application.name=store-server

mybatis.type-aliases-package=com.chuanzh.model
mybatis.mapper-locations=classpath:mapper/*.xml

spring.datasource.url=jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

#注意这里的事务组配置要和服务端一致
seata.tx-service-group=chuanzh_tx_group
seata.service.vgroup-mapping.chuanzh_tx_group=default
seata.service.grouplist.default=192.168.234.128:8091

logging.level.io.seata=DEBUG
## eureka
eureka.client.serviceUrl.defaultZone= http://localhost:8081/eureka/

数据源配置,因为seata是对数据库的datasource进行了接管和代理,所以在每个参与分布式事务的数据源都要进行如下配置:

@Configuration
public class DataSourceConfiguration {

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource druidDataSource(){
        DruidDataSource druidDataSource = new DruidDataSource();
        return druidDataSource;
    }

    @Primary
    @Bean("dataSource")
    public DataSourceProxy dataSource(DataSource druidDataSource){
        return new DataSourceProxy(druidDataSource);
    }

    @Bean
    public SqlSessionFactory sqlSessionFactory(DataSourceProxy dataSourceProxy)throws Exception{
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSourceProxy);
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver()
                .getResources("classpath*:/mapper/*.xml"));
        sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory());
        return sqlSessionFactoryBean.getObject();
    }

}

注意配置了数据源后,还需要在启动类排除dataSource自动配置,不然会出现循环依赖的问题,如下,其它的解决方法,可以参考:集成fescar数据源循环依赖错误解决方案

@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)

配置请求拦截器,生成一个请求事务ID,用于在微服务中传递

@Configuration
public class SeataRequestInterceptor implements RequestInterceptor {
    @Override
    public void apply(RequestTemplate requestTemplate) {
        String xid = RootContext.getXID();
        if (StringUtils.isNotBlank(xid)) {
            //构建请求头
            requestTemplate.header("TX_XID", xid);
        }
    }
}

服务提供方配置(company-server)

maven、application.properties、数据源配置同调用方配置,区别主要是拦截器的配置,如下:

@Slf4j
@Component
public class SeataHandlerInterceptor implements HandlerInterceptor {

    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        String xid = RootContext.getXID();
        String rpcXid = request.getHeader("TX_XID");
        //获取全局事务编号
        if(log.isDebugEnabled()) {
            log.debug("xid in RootContext {} xid in RpcContext {}", xid, rpcXid);
        }
        if(xid == null && rpcXid != null) {
            //设置全局事务编号
            RootContext.bind(rpcXid);
            if(log.isDebugEnabled()) {
                log.debug("bind {} to RootContext", rpcXid);
            }
        }
        return true;
    }
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception e) {
        String rpcXid = request.getHeader("TX_XID");
        if(!StringUtils.isEmpty(rpcXid)) {
            String unbindXid = RootContext.unbind();
            if(log.isDebugEnabled()) {
                log.debug("unbind {} from RootContext", unbindXid);
            }

            if(!rpcXid.equalsIgnoreCase(unbindXid)) {
                log.warn("xid in change during RPC from {} to {}", rpcXid, unbindXid);
                if(unbindXid != null) {
                    RootContext.bind(unbindXid);
                    log.warn("bind {} back to RootContext", unbindXid);
                }
            }

        }
    }

}
@Configuration
public class WebMvcConfiguration implements WebMvcConfigurer {

    @Autowired
    private SeataHandlerInterceptor seataHandlerInterceptor;

    public void addInterceptors(InterceptorRegistry registry) {
        //注册HandlerInterceptor,拦截所有请求
        registry.addInterceptor(seataHandlerInterceptor).addPathPatterns(new String[]{"/**"});
    }

}

添加全局事务注解

在服务调用方的方法上添加@GlobalTransactional注解,下面模拟了一种场景,如果companyId为偶数,则会抛异常。

    @GlobalTransactional(rollbackFor = Exception.class)
    public void create(StoreEntity storeEntity) throws Exception {
        CompanyEntity companyEntity = new CompanyEntity();
        companyEntity.setName(storeEntity.getName());
        companyEntity = companyFeign.createCompany(companyEntity);

        /**
         * 模拟异常
         */
        if (companyEntity.getId() % 2 == 0) {
            throw new Exception();
        }

        /** 写入store数据 */
        storeEntity.setCompanyId(companyEntity.getId());
        storeMapper.insert(storeEntity);
    }

经过测试,companyFeign.createCompany服务调用后会先向数据库写一条数据,当create方法执行抛异常,就会事务回滚,删除掉原先的company数据

本文来源自与公众号《百川分享会》:baichuanshare

qrcode_for_gh_5b867016dad2_258.jpg

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

相关文章
Fescar&Seata分布式事务实现原理解析探秘
前言 fescar发布已有时日,分布式事务一直是业界备受关注的领域,fescar发布一个月左右便受到了近5000个star足以说明其热度。当然,在fescar出来之前,已经有比较成熟的分布式事务的解决方案开源了,比较典型的方案如LCN(https://github.com/codingapi/tx-lcn)的2pc型无侵入事务,目前lcn已发展到5.0,已支持和fescar事务模型类似的TCX型事务。
1923 0
(一):C++分布式实时应用框架----整体介绍
C++分布式实时应用框架 (Cpp Distributed Real-time Application Framework) 版权声明:本文版权及所用技术归属smartguys团队所有,对于抄袭,非经同意转载等行为保留法律追究的权利!     在现今软件系统纷纷“云化”的浪潮下,各种支持“云化”的框架、工具层出不穷,但这些现成的工具大多基于JAVA,go等语言,且应用场景均为非实时系统。
1252 0
分布式事务解决方案——柔性事务与服务模式
原创: Hollis 在我的博客中,介绍过很多关于分布式和事务的文章,在阅读本文之前,希望读者可以对这些基础知识有所了解,这里简单把之前的文章列举下,已经按照顺序排好,可按顺序阅读。
1613 0
大白话聊聊分布式事务
# 大白话聊聊分布式事务 ## 什么是分布式事务 简单的来说就是,一个大的操作由两个或者更多的小的操作共同完成。而这些小的操作又分布在不同的网络主机上。这些操作,要么全部成功执行,要么全部不执行。 拿转账的例子来说下什么是分布式事务。张三和李四在不同的城市,存储他们账户信息的服务器也在不同的网络主机上。张三有30元钱,李四有30元钱。张三给李四转账5元就是一个事务。完成这个事务,需要两个
3215 0
产品设计不是命题作文:Design Hackathon 方法介绍
在产品的定义阶段,产品发展形态的可能性是最多的。对于当前国内绝大多数移动互联网创业公司来说,在产品定义初期,往往都是由个别产品负责人或者创始人「决定」产品方向的。这种「命题式」的传统方法,会导致产品的大部分可能性被早早扼杀,很容易让产品设计陷入程式化的思维或是已有的产品模式。在这种方式下,不能说诞生不了好的产品,但突破和创新的难度将会大大提高。传统的「头脑风暴」,在发散思维时往往失于天马行空,忽略了落地的可行性。
78 0
+关注
川行
每天积累一点点,不断超越自己
6
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载