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文件
<code>## 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"
}
</code>
上面配置共修改了3个地方：
- 存储模式改为db模式，需要创建3张事务表，如下：
<code>-- 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;
</code>
- 修改数据库连接，注意如果你安装的是MySQL8，则需要修改MySQL8的驱动：driverClassName = “com.mysql.cj.jdbc.Driver”，不然会出现启动报错的问题，详细请参考：seata启动MySQL报错 #359（https://github.com/seata/seata-samples/issues/359）。
- 修改事务的组名，你也可以不修改，我这里使用的是：chuanzh_tx_group
- 创建业务事务表，记录业务需要回滚的数据，在分布式事务中，每个参与的业务数据库都需要添加对应的表
<code>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;
</code>
配置conf/registry.conf文件
<code>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"
}
}
</code>
以上修改了使用Eureka方式注册，并配置了Eureka地址，启动MySQL、Eureka服务后，就可以启动seata服务了。
调用方配置（store-server）
maven配置，使用seata-spring-boot-starter，自动配置的方式，不需要再添加file.conf和register.conf文件
<code> <!--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>
</code>
application.properties配置：
<code>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/
</code>
数据源配置，因为seata是对数据库的datasource进行了接管和代理，所以在每个参与分布式事务的数据源都要进行如下配置：
<code>@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();
}
}
</code>
注意配置了数据源后，启动会出现循环依赖的问题,如下，
还需要在启动类排除dataSource自动配置，其它的解决方法，可以参考：集成fescar数据源循环依赖错误解决方案（https://blog.csdn.net/kangsa998/article/details/90042406）
<code>@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
</code>
配置请求拦截器，生成一个请求事务ID，用于在微服务中传递
<code>@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);
}
}
}
</code>
服务提供方配置（company-server）
maven、application.properties、数据源配置同调用方配置，区别主要是拦截器的配置，如下：
<code>@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);
}
}
}
}
}
</code>
<code>@Configuration
public class WebMvcConfiguration implements WebMvcConfigurer {
@Autowired
private SeataHandlerInterceptor seataHandlerInterceptor;
public void addInterceptors(InterceptorRegistry registry) {
//注册HandlerInterceptor，拦截所有请求
registry.addInterceptor(seataHandlerInterceptor).addPathPatterns(new String[]{"/**"});
}
}
</code>
添加全局事务注解
在服务调用方的方法上添加@GlobalTransactional注解，下面模拟了一种场景，如果companyId为偶数，则会抛异常。
<code> @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);
}
</code>
经过测试，companyFeign.createCompany服务调用后会先向数据库写一条数据，当create方法执行抛异常，就会事务回滚，删除掉原先的company数据
