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

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

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

seata安装

下载seata服务，官方地址： github /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文件 store { mode = "db" //修改为db模式，标识事务信息用db存储 file { dir = "sessionStore" maxBranchSessionSize = 16384 maxGlobalSessionSize = 512 fileWriteBufferCacheSize = 16384 sessionReloadReadSize = 100 flushDiskMode = async } db { datasource = "druid" 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 } } service { vgroup_mapping.chuanzh_tx_group = "default" //chuanzh_tx_group为自定义的事务组名称，要和客户端配置保持一致 default.grouplist = "192.168.234.128:8091" enableDegrade = false disable = false max mit.retry.timeout = "-1" max.rollback.retry.timeout = "-1" }

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

存储模式改为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;

修改数据库连接，注意如果你安装的是MySQL8，则需要修改MySQL8的驱动：driverClassName = “com.mysql.cj.jdbc.Driver”，不然会出现启动报错的问题，详细请参考：seata启动MySQL报错 #359。

修改事务的组名，你也可以不修改，我这里使用的是：chuanzh_tx_group

创建业务事务表，记录业务需要回滚的数据，在分布式事务中，每个参与的业务数据库都需要添加对应的表

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 { 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文件

<dependency> <groupId>com.alibaba</groupId> <artifactId>druid-spring-boot-starter</artifactId> <version>${druid-spring-boot-starter.version}</version> </dependency> <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.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) { 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(); } storeEntity.setCompanyId(companyEntity.getId()); storeMapper.insert(storeEntity); }

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