SpringCloud Alibaba Seata分布式事务
简介
Seata是一款开源的分布式事务解决方案,致力于在微服务架构下提供高性能和简单易用的分布式事务服务
官网:http://seata.io/zh-cn/
一次业务操作需要垮多个数据源或需要垮多个系统进行远程调用,就会产生分布式事务问题
单体应用被拆分微服务应用,原来的三个模块被拆分成三个独立的应用,分别使用三个独立的数据源,业务操作需要调用三个服务来完成。此时每个服务内部的数据一致性由本地事务来保证,但是全局的数据一致性问题没法保证。
比如:
- 仓储服务:对给定的商品扣除仓储数量
- 订单服务:根据采购需求创建订单,该支付状态
- 账户服务:从用户账户中扣除金额
分布式事务处理过程-ID+三组件模型
Transaction ID(XID)
全局唯一的事务id
TC - 事务协调者
维护全局和分支事务的状态,驱动全局事务提交或回滚。
TM - 事务管理器
定义全局事务的范围:开始全局事务、提交或回滚全局事务。
RM - 资源管理器
管理分支事务处理的资源,与TC交谈以注册分支事务和报告分支事务的状态,并驱动分支事务提交或回滚。
TM向TC申请开启一个全局事务,全局事务创建成功并生成一个全局唯一的XID;
XID在微服务调用链路的上下文中传播;
RM向TC注册分支事务,将其纳入XID对应全局事务的管辖;
TM向TC发起针对XID的全局提交或回滚决议;
TC调度XID下管辖的全部分支事务完成提交提交和回滚请求。
下载配置搭建
github下载:https://github.com/seata/seata/releases
linux安装tar包,步骤差不多
0.9.0版本
seata-server-0.9.0.zip解压到指定目录并修改conf目录下的file.conf配置文件,先备份原始file.conf文件,主要修改:自定义事务组名称+事务日志存储模式为db+数据库连接
file.conf中service:
xxx_tx_group只是事务分组名
service{
vgroup_mapping.my_test_tx_group = "fsp_tx_group"
}
store:
数据库连接模式,修改链接
store{
mode="db"
db{
url="jdbc:mysql://127.0.0.1:3306/seata"
user="数据库账号"
password="数据库密码"
}
}
mysql5.7数据库新建库seata
建表db_store.sql在seata-server-0.9.0\seata\conf目录里面,直接在seata库运行
修改seata-server-0.9.0\seata\conf目录下的registry.conf目录下的registry.conf配置文件,注册进nacos,写好地址
registry{
type="nacos"
nacos{
serverAddr="localhost:8848"
}
}
先启动Nacos端口号8848,再启动seata-server,seata-server-0.9.0\seata\bin,seata-server.bat
启动成功
1.1.0版本
大致一样
store:
数据库连接模式,修改链接
store{
mode="db"
db{
url="jdbc:mysql://127.0.0.1:3306/seata"
user="数据库账号"
password="数据库密码"
}
}
mysql5.7以上数据库新建库seata
建表db_store.sql在seata-server-0.9.0(1.0以上没有)\seata\conf目录里面,直接在seata库运行
修改seata-server-1.1.0\seata\conf目录下的registry.conf目录下的registry.conf配置文件,注册进nacos,写好地址
registry{
type="nacos"
nacos{
serverAddr="localhost:8848"
}
}
先启动Nacos端口号8848,再启动seata-server,seata-server-1.1.0\seata\bin,seata-server.bat
启动成功
案例分析
订单微服务,库存微服务,账户微服务
当用户下单时,会在订单服务中创建一个订单,然后通过远程调用库存服务来扣减下的商品的库存,再通过远程调用账户服务来扣减用户账户的余额,最后在订单服务中修改订单状态为已支付。
该操作跨越三个数据库,有两次远程调用,很明显会有分布式事务问题。
创建业务数据库
seata_order:存储订单的数据库
seata_storage:存储库存的数据库
seata_account:存储账户信息的数据库
create database seata_order;
create database seata_storage;
create database seata_account;
CREATE TABLE `t_order` (
`int` bigint(11) NOT NULL AUTO_INCREMENT,
`user_id` bigint(20) DEFAULT NULL COMMENT '用户id',
`product_id` bigint(11) DEFAULT NULL COMMENT '产品id',
`count` int(11) DEFAULT NULL COMMENT '数量',
`money` decimal(11, 0) DEFAULT NULL COMMENT '金额',
`status` int(1) DEFAULT NULL COMMENT '订单状态: 0:创建中 1:已完结',
PRIMARY KEY (`int`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '订单表' ROW_FORMAT = Dynamic;
CREATE TABLE `t_storage` (
`int` bigint(11) NOT NULL AUTO_INCREMENT,
`product_id` bigint(11) DEFAULT NULL COMMENT '产品id',
`total` int(11) DEFAULT NULL COMMENT '总库存',
`used` int(11) DEFAULT NULL COMMENT '已用库存',
`residue` int(11) DEFAULT NULL COMMENT '剩余库存',
PRIMARY KEY (`int`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '库存' ROW_FORMAT = Dynamic;
INSERT INTO `t_storage` VALUES (1, 1, 100, 0, 100);
CREATE TABLE `t_account` (
`id` bigint(11) NOT NULL COMMENT 'id',
`user_id` bigint(11) DEFAULT NULL COMMENT '用户id',
`total` decimal(10, 0) DEFAULT NULL COMMENT '总额度',
`used` decimal(10, 0) DEFAULT NULL COMMENT '已用余额',
`residue` decimal(10, 0) DEFAULT NULL COMMENT '剩余可用额度',
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '账户表' ROW_FORMAT = Dynamic;
INSERT INTO `t_account` VALUES (1, 1, 1000, 0, 1000);
订单-库存-账户3个库下都需要建各自独立的回滚日志表
seata-server-0.9.0\seata\conf\目录下的db_undo_log.sql
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;
业务工程
下订单->减库存->扣余额->改(订单)
依赖
1.1.0/0.9.0根据具体的下载的seata版本对应
<!-- seata -->
<io.seata.version>1.1.0</io.seata.version>
<dependency>
<groupId>io.seata</groupId>
<artifactId>seata-all</artifactId>
<version>${io.seata.version}</version>
</dependency>
<!-- seata-->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-seata</artifactId>
<exclusions>
<exclusion>
<groupId>io.seata</groupId>
<artifactId>seata-all</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.seata</groupId>
<artifactId>seata-all</artifactId>
</dependency>
<!-- seata-->
微服务搭建
启动nacos,再启动seata
以下是0.9.0版本的配置
下订单->减库存->扣余额->改(订单)状态
模拟3个微服务,库存、订单、账户。访问订单微服务接口实现所有操作,订单微服务feign调用库存、账户的接口,这里一个接口就涉及到了3个微服务。
这里只简单介绍订单微服务(其他2个微服务一样的步骤,feign模块):
结合openfeign、nacos-discovery、druid、mybatis依赖
yml配置
server:
port: 2001
spring:
application:
name: seata-order-service
cloud:
alibaba:
seata:
# 自定义事务组名称需要与seata-server中的对应
tx-service-group: fsp_tx_group
nacos:
discovery:
server-addr: 127.0.0.1:8848
datasource:
# 当前数据源操作类型
type: com.alibaba.druid.pool.DruidDataSource
# mysql驱动类
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://192.168.169.130:3306/seata_order?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8
username: root
password: 123456
feign:
hystrix:
enabled: false
logging:
level:
io:
seata: info
mybatis:
mapper-locations: classpath*:mapper/*.xml
type-aliases-package: com.wzq.springcloud.domain
- tx-service-group: fsp_tx_group是自定义事务组名称需要与seata-server中的相对应,fsp可改
- seata日志配置logging
seate配置
拷贝seata-server/conf目录下的file.conf到微服务的resources,注意连接的是对应的seata数据库
拷贝seata-server/conf目录下的registry.conf到微服务的resources,注意对应的注册中心配置
业务代码
feign调用,实现跨微服对不同数据库操作。
service.impl
import com.wzq.springcloud.dao.OrderDao;
import com.wzq.springcloud.domain.Order;
import com.wzq.springcloud.service.AccountService;
import com.wzq.springcloud.service.OrderService;
import com.wzq.springcloud.service.StorageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
* 订单
*
* @author zzyy
* @date 2020/3/8 13:57
**/
@Service
@Slf4j
public class OrderServiceImpl implements OrderService {
@Resource
private OrderDao orderDao;
@Resource
private AccountFeign accountFeign;
@Resource
private StorageFeign storageFeign;
/**
* 创建订单->调用库存服务扣减库存->调用账户服务扣减账户余额->修改订单状态
* 简单说:
* 下订单->减库存->减余额->改状态
* GlobalTransactional seata开启分布式事务,异常时回滚,name保证唯一即可
*
* @param order 订单对象
*/
@Override
//@GlobalTransactional(name = "fsp-create-order", rollbackFor = Exception.class)
public void create(Order order) {
// 1 新建订单
log.info("----->开始新建订单");
orderDao.create(order);
// 2 扣减库存
log.info("----->订单微服务开始调用库存,做扣减Count");
storageFeign.decrease(order.getProductId(), order.getCount());
log.info("----->订单微服务开始调用库存,做扣减End");
// 3 扣减账户
log.info("----->订单微服务开始调用账户,做扣减Money");
accountFeign.decrease(order.getUserId(), order.getMoney());
log.info("----->订单微服务开始调用账户,做扣减End");
// 4 修改订单状态,从0到1,1代表已完成
log.info("----->修改订单状态开始");
orderDao.update(order.getUserId(), 0);
log.info("----->下订单结束了,O(∩_∩)O哈哈~");
}
}
controller:
@RestController
public class OrderController {
@Resource
private OrderService orderService;
/**
* 创建订单
*
* @param order
* @return
*/
@GetMapping("/order/create")
public CommonResult create(Order order) {
orderService.create(order);
return new CommonResult(200, "订单创建成功");
}
}
Config配置
排除springboot自动加载数据源,交给seata管理
package com.wzq.springcloud.config;
import com.alibaba.druid.pool.DruidDataSource;
import io.seata.rm.datasource.DataSourceProxy;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import javax.sql.DataSource;
/**
* seata管理数据源
* 排除springboot加载数据源,交给seata管理
* @author wzq
* @version 1.0
* @create 2020/3/8 15:35
*/
@Configuration
public class DataSourceProxyConfig {
@Value("${mybatis.mapper-locations}")
private String mapperLocations;
/**
* @param sqlSessionFactory SqlSessionFactory
* @return SqlSessionTemplate
*/
@Bean
public SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {
return new SqlSessionTemplate(sqlSessionFactory);
}
/**
* 从配置文件获取属性构造datasource,注意前缀,这里用的是druid,根据自己情况配置,
* 原生datasource前缀取"spring.datasource"
*
* @return
*/
@Bean
@ConfigurationProperties(prefix = "spring.datasource")
public DataSource druidDataSource() {
return new DruidDataSource();
}
/**
* 构造datasource代理对象,替换原来的datasource
*
* @param druidDataSource
* @return
*/
@Primary
@Bean("dataSource")
public DataSourceProxy dataSourceProxy(DataSource druidDataSource) {
return new DataSourceProxy(druidDataSource);
}
@Bean(name = "sqlSessionFactory")
public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception {
SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
bean.setDataSource(dataSourceProxy);
ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
bean.setMapperLocations(resolver.getResources(mapperLocations));
SqlSessionFactory factory;
try {
factory = bean.getObject();
} catch (Exception e) {
throw new RuntimeException(e);
}
return factory;
}
}
主启动类
@EnableDiscoveryClient
@EnableFeignClients
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@MapperScan({"com.wzq.springcloud.dao"})
测试
- 正常下单
http://localhost:2001/order/create?userId=1&productId=1&count=10&money=100
数据库情况,数据修改正常
- 超时异常,没加@GlobalTransactional
停止storage微服务,或者storage微服务的service方法设置超时。即是库存微服务挂了
当库存和账户金额扣减后,订单状态并没有设置为已经完成,没有从零改为1,而且由于feign的重试机制,账户余额还有可能被多次扣减。
数据库数据订单下了,账户也支付了,但是库存没减少,订单状态没变。
- 超时异常,添加@GlobalTransactional
OrderServiceImpl添加@GlobalTransactional,该方法对多数据库进行数据修改
import com.wzq.springcloud.dao.OrderDao;
import com.wzq.springcloud.domain.Order;
import com.wzq.springcloud.service.AccountService;
import com.wzq.springcloud.service.OrderService;
import com.wzq.springcloud.service.StorageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
* 订单
*
* @author zzyy
* @date 2020/3/8 13:57
**/
@Service
@Slf4j
public class OrderServiceImpl implements OrderService {
@Resource
private OrderDao orderDao;
@Resource
private AccountFeign accountFeign;
@Resource
private StorageFeign storageFeign;
/**
* 创建订单->调用库存服务扣减库存->调用账户服务扣减账户余额->修改订单状态
* 简单说:
* 下订单->减库存->减余额->改状态
* GlobalTransactional seata开启分布式事务,异常时回滚,name保证唯一即可
*
* @param order 订单对象
*/
@Override
@GlobalTransactional(name = "fsp-create-order", rollbackFor = Exception.class)
public void create(Order order) {
// 1 新建订单
log.info("----->开始新建订单");
orderDao.create(order);
// 2 扣减库存
log.info("----->订单微服务开始调用库存,做扣减Count");
storageFeign.decrease(order.getProductId(), order.getCount());
log.info("----->订单微服务开始调用库存,做扣减End");
// 3 扣减账户
log.info("----->订单微服务开始调用账户,做扣减Money");
accountFeign.decrease(order.getUserId(), order.getMoney());
log.info("----->订单微服务开始调用账户,做扣减End");
// 4 修改订单状态,从0到1,1代表已完成
log.info("----->修改订单状态开始");
orderDao.update(order.getUserId(), 0);
log.info("----->下订单结束了,O(∩_∩)O哈哈~");
}
}
@GlobalTransactional(name = "fsp-create-order", rollbackFor = Exception.class)
name只要跟其他全局异常名字不重复就行,rollbackFor = Exception.class指发生任何异常就回滚。
结果如下:因为没有弄降级和熔断
但是数据库没有任何数据变动。
回滚后日志表会有数据,undo_log
补充回顾
使用1.0以后的版本
TM管理所有全局事务,通过XID全局id,TC管理某个全局事务,RM是某个数据库
分布式事务的执行流程:
- TM开启分布式事务(TM向TC注册全局事务记录)
- 按业务场景,编排数据库、服务等事务内资源(RM向TC汇报资源准备状态)
- TM结束分布式事务,事务一阶段结束(TM通知TC提交/回滚分布式事务)
- TC汇报事务信息,决定分布式事务是提交还是回滚
- TC通知所有RM提交/回滚资源,事务二阶段结束
提供了 AT、TCC、SAGA 和 XA 事务模式
默认AT模式,其他要收费
AT解释:http://seata.io/zh-cn/docs/overview/what-is-seata.html
提供无侵入自动补偿的事务模式,目前已支持 MySQL、 Oracle 、PostgreSQL和 TiDB的AT模式,H2 开发中
一阶段加载:
Seata 会拦截业务SQL
1 解析SQL 语义,找到“业务SQL”要更新的业务数据,在业务数据被更新前,将其保存成“before image"
2 执行“业务SQL更新业务数据,在业务数据更新之后
3 其保存成“after image",最后生成行锁。
以上操作全部在一个数据库事务内完成,这样保证了一阶段操作的原子性
二阶段提交:
因为“业务SQL”在一阶段已经提交至数据库,所以Seata框架只需要将一阶段保存的快照数据和行锁删掉。
三阶段回滚:
二阶段如果回滚的话,Seata就需要回滚一阶段已经执行的“业务SQL”,还原业务数据。
回滚方式便是用“before image“还原业务数据,但在还原前要首先要校验脏写,对比“数据库当前业务数据”和“after image”
如果两份数据完全一致就说明没有脏写,可以还原业务数据,如果不一致就说明有脏写,出现脏与就需要转人工处理
debug3个测试微服务:
产生3个分支,参与全局事务的微服务数据库,类型AT
8091是TC,seata服务器
各个数据库日志表,rollback_info存放着before image,一旦回滚,根据这个image回滚,反向执行。XID是全局事务id。