07分布式事务Seata使用（2）

Seata 是一款开源的分布式事务解决方案&＃xff0c;致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式&＃xff0c;为用户打造一站式的分布式解决方案。AT模式是阿里首推的模式,阿里云上有商用版本的GTS&＃xff08;Global Transaction Service 全局事务服务&＃xff09;

官网&＃xff1a;https://seata.io/zh-cn/index.html
源码: https://github.com/seata/seata
官方Demo: https://github.com/seata/seata-samples
seata版本&＃xff1a;v1.4.0

1.1 Seata的三大角色

在 Seata 的架构中&＃xff0c;一共有三个角色&＃xff1a;

• TC (Transaction Coordinator) - 事务协调者&＃xff1a;维护全局和分支事务的状态&＃xff0c;驱动全局事务提交或回滚。
• TM (Transaction Manager) - 事务管理器&＃xff1a;定义全局事务的范围&＃xff1a;开始全局事务、提交或回滚全局事务。
• RM (Resource Manager) - 资源管理器&＃xff1a;管理分支事务处理的资源&＃xff0c;与TC交谈以注册分支事务和报告分支事务的状态&＃xff0c;并驱动分支事务提交或回滚。

其中&＃xff0c;TC 为单独部署的 Server 服务端&＃xff0c;TM 和 RM 为嵌入到应用中的 Client 客户端。

在 Seata 中&＃xff0c;一个分布式事务的生命周期如下&＃xff1a;

1、TM 请求 TC 开启一个全局事务。TC 会生成一个 XID 作为该全局事务的编号。XID&＃xff0c;会在微服务的调用链路中传播&＃xff0c;保证将多个微服务的子事务关联在一起。当一进入事务方法中就会生成XID &＃xff0c; global_table 就是存储的全局事务信息。
2、RM 请求 TC 将本地事务注册为全局事务的分支事务&＃xff0c;通过全局事务的 XID 进行关联。当运行数据库操作方法&＃xff0c;branch_table 存储事务参与者。
3、TM 请求 TC 告诉 XID 对应的全局事务是进行提交还是回滚。
4、TC 驱动 RM 们将 XID 对应的自己的本地事务进行提交还是回滚。

1.2 设计思路

AT模式的核心是对业务无侵入&＃xff0c;是一种改进后的两阶段提交&＃xff0c;其设计思路如图&＃xff1a;

第一阶段

业务数据和回滚日志记录在同一个本地事务中提交&＃xff0c;释放本地锁和连接资源。核心在于对业务sql进行解析&＃xff0c;转换成undolog&＃xff0c;并同时入库&＃xff0c;这是怎么做的呢&＃xff1f;先抛出一个概念DataSourceProxy代理数据源&＃xff0c;通过名字大家大概也能基本猜到是什么个操作&＃xff0c;后面做具体分析。
参考官方文档&＃xff1a; https://seata.io/zh-cn/docs/dev/mode/at-mode.html

第二阶段

分布式事务操作成功&＃xff0c;则TC通知RM异步删除undolog。

分布式事务操作失败&＃xff0c;TM向TC发送回滚请求&＃xff0c;RM 收到协调器TC发来的回滚请求&＃xff0c;通过 XID 和 Branch ID 找到相应的回滚日志记录&＃xff0c;通过回滚记录生成反向的更新 SQL 并执行&＃xff0c;以完成分支的回滚。

整体执行流程

1.3 设计亮点

相比与其它分布式事务框架&＃xff0c;Seata架构的亮点主要有几个:

1、应用层基于SQL解析实现了自动补偿&＃xff0c;从而最大程度的降低业务侵入性&＃xff1b;
2、将分布式事务中TC&＃xff08;事务协调者&＃xff09;独立部署&＃xff0c;负责事务的注册、回滚&＃xff1b;
3、通过全局锁实现了写隔离与读隔离。

1.4 存在的问题

性能损耗

一条Update的SQL&＃xff0c;则需要全局事务xid获取&＃xff08;与TC通讯&＃xff09;、before image&＃xff08;解析SQL&＃xff0c;查询一次数据库&＃xff09;、after image&＃xff08;查询一次数据库&＃xff09;、insert undo log&＃xff08;写一次数据库&＃xff09;、before commit&＃xff08;与TC通讯&＃xff0c;判断锁冲突&＃xff09;&＃xff0c;这些操作都需要一次远程通讯RPC&＃xff0c;而且是同步的。另外undo log写入时blob字段的插入性能也是不高的。每条写SQL都会增加这么多开销,粗略估计会增加5倍响应时间。

性价比

为了进行自动补偿&＃xff0c;需要对所有交易生成前后镜像并持久化&＃xff0c;可是在实际业务场景下&＃xff0c;这个是成功率有多高&＃xff0c;或者说分布式事务失败需要回滚的有多少比率&＃xff1f;按照二八原则预估&＃xff0c;为了20%的交易回滚&＃xff0c;需要将80%的成功交易的响应时间增加5倍&＃xff0c;这样的代价相比于让应用开发一个补偿交易是否是值得&＃xff1f;

全局锁

热点数据

相比XA&＃xff0c;Seata 虽然在一阶段成功后会释放数据库锁&＃xff0c;但一阶段在commit前全局锁的判定也拉长了对数据锁的占有时间&＃xff0c;这个开销比XA的prepare低多少需要根据实际业务场景进行测试。全局锁的引入实现了隔离性&＃xff0c;但带来的问题就是阻塞&＃xff0c;降低并发性&＃xff0c;尤其是热点数据&＃xff0c;这个问题会更加严重。

回滚锁释放时间

Seata在回滚时&＃xff0c;需要先删除各节点的undo log&＃xff0c;然后才能释放TC内存中的锁&＃xff0c;所以如果第二阶段是回滚&＃xff0c;释放锁的时间会更长。

死锁问题

Seata的引入全局锁会额外增加死锁的风险&＃xff0c;但如果出现死锁&＃xff0c;会不断进行重试&＃xff0c;最后靠等待全局锁超时&＃xff0c;这种方式并不优雅&＃xff0c;也延长了对数据库锁的占有时间。

2.1 Seata Server&＃xff08;TC&＃xff09;环境搭建

https://seata.io/zh-cn/docs/ops/deploy-guide-beginner.html

步骤一&＃xff1a;下载安装包

https://github.com/seata/seata/releases

Server端存储模式&＃xff08;store.mode&＃xff09;支持三种&＃xff1a;

• file&＃xff1a;(默认)单机模式&＃xff0c;全局事务会话信息内存中读写并持久化本地文件root.data&＃xff0c;性能较高(默认&＃xff09;
• db&＃xff1a;&＃xff08;5.7&＃43;&＃xff09;高可用模式&＃xff0c;全局事务会话信息通过db共享&＃xff0c;相应性能差些
  
  • 打开config/file.conf
  • 修改mode&＃61;“db”
  • 修改数据库连接信息&＃xff08;URL\USERNAME\PASSWORD)
  • 创建数据库seata_server
  • 新建表&＃xff1a; 可以去seata提供的资源信息中下载&＃xff1a;
    
    • 点击查看
    • \script\server\db\mysql.sql
      
      • branch 表 存储事务参与者的信息
        

store {
mode &＃61; "db"
db {
## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp)/HikariDataSource(hikari) etc.
datasource &＃61; "druid"
## mysql/oracle/postgresql/h2/oceanbase etc.
dbType &＃61; "mysql"
driverClassName &＃61; "com.mysql.jdbc.Driver"
url &＃61; "jdbc:mysql://192.168.65.220:3306/seata_server"
user &＃61; "root"
password &＃61; "123456"
minConn &＃61; 5
maxConn &＃61; 30
globalTable &＃61; "global_table"
branchTable &＃61; "branch_table"
lockTable &＃61; "lock_table"
queryLimit &＃61; 100
maxWait &＃61; 5000
}
}


• redis&＃xff1a;Seata-Server 1.3及以上版本支持,性能较高,存在事务信息丢失风险,请提前配置适合当前场景的redis持久化配置

资源目录&＃xff1a;https://github.com/seata/seata/tree/1.3.0/script

• client : 存放client端sql脚本&＃xff0c;参数配置
• config-center: 各个配置中心参数导入脚本&＃xff0c;config.txt(包含server和client&＃xff0c;原名nacos-config.txt)为通用参数文件
• server: server端数据库脚本及各个容器配置

db存储模式&＃43;Nacos(注册&配置中心)部署

步骤五&＃xff1a;配置Nacos注册中心 负责事务参与者&＃xff08;微服务&＃xff09; 和TC通信

将Seata Server注册到Nacos&＃xff0c;修改conf目录下的registry.conf配置

然后启动注册中心Nacos Server

#进入Nacos安装目录&＃xff0c;linux单机启动
bin/startup.sh -m standalone
# windows单机启动
bin/startup.bat


步骤六&＃xff1a;配置Nacos配置中心

注意&＃xff1a;如果配置了seata server使用nacos作为配置中心&＃xff0c;则配置信息会从nacos读取&＃xff0c;file.conf可以不用配置。 客户端配置registry.conf使用nacos时也要注意group要和seata server中的group一致&＃xff0c;默认group是"DEFAULT_GROUP"

获取/seata/script/config-center/config.txt&＃xff0c;修改配置信息

配置事务分组&＃xff0c; 要与客户端配置的事务分组一致

#my_test_tx_group需要与客户端保持一致 default需要跟客户端和registry.conf中registry中的cluster保持一致

&＃xff08;客户端properties配置&＃xff1a;spring.cloud.alibaba.seata.tx‐service‐group&＃61;my_test_tx_group&＃xff09;

事务分组&＃xff1a; 异地机房停电容错机制
my_test_tx_group 可以自定义 比如&＃xff1a;&＃xff08;guangzhou、shanghai…&＃xff09; &＃xff0c; 对应的client也要去设置

seata.service.vgroup-mapping.projectA&＃61;guangzhou

default 必须要等于 registry.confi cluster &＃61; “default”

配置参数同步到Nacos
shell:

sh ${SEATAPATH}/script/config-center/nacos/nacos-config.sh -h localhost -p 8848 -g SEATA_GROUP -t 5a3c7d6c-f497-4d68-a71a-2e5e3340b3ca


参数说明&＃xff1a;

-h: host&＃xff0c;默认值 localhost
-p: port&＃xff0c;默认值 8848
-g: 配置分组&＃xff0c;默认值为 ‘SEATA_GROUP’
-t: 租户信息&＃xff0c;对应 Nacos 的命名空间ID字段, 默认值为空 ‘’

精简配置

service.vgroupMapping.my_test_tx_group&＃61;default
service.default.grouplist&＃61;127.0.0.1:8091
service.enableDegrade&＃61;false
service.disableGlobalTransaction&＃61;false
store.mode&＃61;db
store.db.datasource&＃61;druid
store.db.dbType&＃61;mysql
store.db.driverClassName&＃61;com.mysql.jdbc.Driver
store.db.url&＃61;jdbc:mysql://127.0.0.1:3306/seata?useUnicode&＃61;true
store.db.user&＃61;root
store.db.password&＃61;root
store.db.minConn&＃61;5
store.db.maxConn&＃61;30
store.db.globalTable&＃61;global_table
store.db.branchTable&＃61;branch_table
store.db.queryLimit&＃61;100
store.db.lockTable&＃61;lock_table
store.db.maxWait&＃61;5000


步骤七&＃xff1a;启动Seata Server

• 源码启动: 执行server模块下io.seata.server.Server.java的main方法
• 命令启动: bin/seata-server.sh -h 127.0.0.1 -p 8091 -m db -n 1 -e test
  

启动Seata Server

bin/seata-server.sh -p 8091 -n 1
bin/seata-server.sh -p 8092 -n 2
bin/seata-server.sh -p 8093 -n 3


启动成功&＃xff0c;默认端口8091

在注册中心中可以查看到seata-server注册成功

2.2 Seata Client快速开始

接入微服务应用

业务场景&＃xff1a;

用户下单&＃xff0c;整个业务逻辑由三个微服务构成&＃xff1a;

• 订单服务&＃xff1a;根据采购需求创建订单。
• 库存服务&＃xff1a;对给定的商品扣除库存数量。

1) 启动Seata server端&＃xff0c;Seata server使用nacos作为配置中心和注册中心&＃xff08;上一步已完成&＃xff09;
2&＃xff09;配置微服务整合seata

第一步&＃xff1a;添加pom依赖

<!-- seata-->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>


第二步&＃xff1a; 各微服务对应数据库中添加undo_log表

CREATE TABLE &＃96;undo_log&＃96; (
&＃96;id&＃96; bigint(20) NOT NULL AUTO_INCREMENT,
&＃96;branch_id&＃96; bigint(20) NOT NULL,
&＃96;xid&＃96; varchar(100) NOT NULL,
&＃96;context&＃96; varchar(128) NOT NULL,
&＃96;rollback_info&＃96; longblob NOT NULL,
&＃96;log_status&＃96; int(11) NOT NULL,
&＃96;log_created&＃96; datetime NOT NULL,
&＃96;log_modified&＃96; datetime NOT NULL,
PRIMARY KEY (&＃96;id&＃96;),
UNIQUE KEY &＃96;ux_undo_log&＃96; (&＃96;xid&＃96;,&＃96;branch_id&＃96;)
) ENGINE&＃61;InnoDB AUTO_INCREMENT&＃61;1 DEFAULT CHARSET&＃61;utf8;


第三步&＃xff1a;修改register.conf,配置nacos作为registry.type&config.type&＃xff0c;对应seata server也使用nacos
注意&＃xff1a;需要指定group &＃61; “SEATA_GROUP”&＃xff0c;因为Seata Server端指定了group &＃61; “SEATA_GROUP” &＃xff0c;必须保证一致

registry {
# file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
type &＃61; "nacos"
nacos {
serverAddr &＃61; "localhost"
namespace &＃61; ""
cluster &＃61; "default"
group &＃61; "SEATA_GROUP"
}
}
config {
# file、nacos 、apollo、zk、consul、etcd3、springCloudConfig
type &＃61; "nacos"
nacos {
serverAddr &＃61; "localhost"
namespace &＃61; ""
group &＃61; "SEATA_GROUP"
}
}


如果出现这种问题&＃xff1a;

一般大多数情况下都是因为配置不匹配导致的&＃xff1a;

1.检查现在使用的seata服务和项目maven中seata的版本是否一致
2.检查tx-service-group&＃xff0c;nacos.cluster&＃xff0c;nacos.group参数是否和Seata Server中的配置一致

跟踪源码&＃xff1a;seata/discover包下实现了RegistryService#lookup&＃xff0c;用来获取服务列表

NacosRegistryServiceImpl#lookup
》String clusterName &＃61; getServiceGroup(key); #获取seata server集群名称
》List<Instance> firstAllInstances &＃61; getNamingInstance().getAllInstances(getServiceName(), getServiceGroup(), clusters)


第四步&＃xff1a;修改application.yml配置
配置seata 服务事务分组&＃xff0c;要与服务端nacos配置中心中service.vgroup_mapping的后缀对应

server:
port: 8020
spring:
application:
name: order-service
cloud:
nacos:
discovery:
server-addr: 127.0.0.1:8848
alibaba:
seata:
tx-service-group:
my_test_tx_group # seata 服务事务分组
datasource:
type: com.alibaba.druid.pool.DruidDataSource
druid:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://localhost:3306/seata_order?useUnicode&＃61;true&characterEncoding&＃61;UTF-8&serverTimezone&＃61;Asia/Shanghai
username: root
password: root
initial-size: 10
max-active: 100
min-idle: 10
max-wait: 60000
pool-prepared-statements: true
max-pool-prepared-statement-per-connection-size: 20
time-between-eviction-runs-millis: 60000
min-evictable-idle-time-millis: 300000
test-while-idle: true
test-on-borrow: false
test-on-return: false
stat-view-servlet:
enabled: true
url-pattern: /druid/*
filter:
stat:
log-slow-sql: true
slow-sql-millis: 1000
merge-sql: false
wall:
config:
multi-statement-allow: true


第五步&＃xff1a;微服务发起者&＃xff08;TM 方&＃xff09;需要添加&＃64;GlobalTransactional注解

&＃64;Override
//&＃64;Transactional
&＃64;GlobalTransactional(name&＃61;"createOrder")
public Order saveOrder(OrderVo orderVo){
log.info("&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;用户下单&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;");
log.info("当前 XID: {}", RootContext.getXID());

// 保存订单
Order order &＃61; new Order();
order.setUserId(orderVo.getUserId());
order.setCommodityCode(orderVo.getCommodityCode());
order.setCount(orderVo.getCount());
order.setMoney(orderVo.getMoney());
order.setStatus(OrderStatus.INIT.getValue());
Integer saveOrderRecord &＃61; orderMapper.insert(order);
log.info("保存订单{}", saveOrderRecord > 0 ? "成功" : "失败");

//扣减库存
storageFeignService.deduct(orderVo.getCommodityCode(),orderVo.getCount());

//扣减余额
accountFeignService.debit(orderVo.getUserId(),orderVo.getMoney());
//更新订单
Integer updateOrderRecord &＃61; orderMapper.updateOrderStatus(order.getId(),OrderStatus.SUCCESS.getValue());
log.info("更新订单id:{} {}", order.getId(), updateOrderRecord > 0 ? "成功" : "失败");

return order;
}


测试&＃xff1a;
分布式事务成功&＃xff0c;模拟正常下单、扣库存&＃xff0c;扣余额
分布式事务失败&＃xff0c;模拟下单扣库存成功、扣余额失败&＃xff0c;事务是否回滚