1. 概述
在《Seata 极简入门》文章中,我们对 Seata 进行了简单的了解,并完成了 Seata 的部署。
Seata 是阿里开源的一款开源的分布式事务解决方案,致力于提供高性能和简单易用的分布式事务服务。
目前,在 Spring Cloud 体系中,我们一般采用两种方式来实现服务的调用:
- 方式一,使用 Dubbo 发布服务,并使用 Dubbo 调用服务
- 方式二,使用 Spring MVC 提供 API 接口,并使用 Feign 调用服务
Seata 内置对 Dubbo 和 Feign 这两种方式的集成,提供分布式事务的功能。因此,本文我们来学习如何使用 Seata 来实现它们俩的分布式事务。
友情提示:
- 对 Dubbo 不了解的胖友,可以阅读《芋道 Spring Cloud Alibaba 服务调用 Dubbo 入门》文章。
- 对 Feign 不了解的胖友,可以阅读《芋道 Spring Cloud 声明式调用 Feign 入门》文章。
注意,考虑到 Nacos 作为注册中心在国内越来越流行,本文将采用 Nacos 作为 Spring Cloud 的注册中心。
友情提示:如果对 Nacos 不了解的胖友,可以参考《芋道 Spring Cloud Alibaba 注册中心 Nacos 入门》文章。
2. AT 模式 + Dubbo
在《Dubbo 分布式事务 Seata 入门》的「2. AT 模式」小节,我们学习了使用 Seata 的 AT 模式,解决多个 Dubbo 服务下的分布式事务的问题。
具体的,我们以用户购买商品的业务逻辑,来作为具体示例,一共会有三个 Dubbo 服务,分别对应自己的数据库。整体如下图所示:
考虑到 Dubbo 在 Spring Cloud 项目中的使用,主要是支持使用 Spring Cloud 注册中心,其它使用都是相同的,所以胖友直接参考《Dubbo 分布式事务 Seata 入门》的「2. AT 模式」小节即可。
当然,良心的艿艿还是搭建了示例项目如下:
- 订单服务:
labx-17-sca-seata-at-dubbo-demo-order-service
- 商品服务:
labx-17-sca-seata-at-dubbo-demo-product-service
- 账户服务:
labx-17-sca-seata-at-dubbo-demo-account-service
和《Dubbo 分布式事务 Seata 入门》的「2. AT 模式」小节提供的示例项目的差异点,主要是两个:
- 在
文件中,引入pom.xml
、spring-cloud-starter-dubbo
、spring-cloud-starter-alibaba-nacos-discovery
等等 Spring Cloud 相关的依赖。spring-cloud-alibaba-seata
- 在
配置文件中,设置application.yaml
配置项,使用 Spring Cloud 注册中心。dubbo.registry
3. AT 模式 + Feign
示例代码对应仓库:
- 订单服务:
labx-17-sc-seata-at-feign-demo-order-service
- 商品服务:
labx-17-sc-seata-at-feign-demo-product-service
- 账户服务:
labx-17-sc-seata-at-feign-demo-account-service
本小节,我们将使用 Seata 的 AT 模式,解决多个 Spring Cloud 服务下的分布式事务的问题,即服务提供者通过 SpringMVC 提供 Restful HTTP API 接口,服务消费者通过 Feign 进行 HTTP 调用。
友情提示:对 Seata 的 AT 模式不了解的胖友,可以阅读《Seata 文档 —— AT 模式》文档。
Seata 提供了
spring-cloud-starter-alibaba-seata
项目,对 Spring Cloud 进行集成。实现原理是:
- 服务消费者,使用 Seata 封装的 SeataFeignClient 过滤器,在使用 Feign 发起 HTTP 调用时,将 Seata 全局事务 XID 通过 Header 传递。
- 服务提供者,使用 Seata 提供的 SpringMVC SeataHandlerInterceptor 拦截器,将 Header 中的 Seata 全局事务 XID 解析出来,设置到 Seata 上下文 中。
如此,我们便实现了多个 Spring Cloud 应用的 Seata 全局事务的传播。
我们以用户购买商品的业务逻辑,来作为具体示例,一共会有三个 Spring Cloud 服务,分别对应自己的数据库。整体如下图所示:
下面,我们来新建
labx-17-sc-seata-at-feign-demo
模块,包含三个 Spring Cloud 服务。最终结构如下图:
3.1 初始化数据库
使用
data.sql
脚本,创建
seata_order
、
seata_storage
、
seata_amount
三个库。脚本内容如下:
#Order
DROP DATABASE IF EXISTS seata_order;
CREATE DATABASE seata_order;
CREATE TABLE seata_order.orders
(
id INT(11) NOT NULL AUTO_INCREMENT,
user_id INT(11) DEFAULT NULL,
product_id INT(11) DEFAULT NULL,
pay_amount DECIMAL(10, 0) DEFAULT NULL,
add_time DATETIME DEFAULT CURRENT_TIMESTAMP,
last_update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id)
) ENGINE = InnoDB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8;
CREATE TABLE seata_order.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,
PRIMARY KEY (id),
UNIQUE KEY ux_undo_log (xid, branch_id)
) ENGINE = InnoDB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8;
# Storage
DROP DATABASE IF EXISTS seata_storage;
CREATE DATABASE seata_storage;
CREATE TABLE seata_storage.product
(
id INT(11) NOT NULL AUTO_INCREMENT,
stock INT(11) DEFAULT NULL,
last_update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id)
) ENGINE = InnoDB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8;
INSERT INTO seata_storage.product (id, stock) VALUES (1, 10); # 插入一条产品的库存
CREATE TABLE seata_storage.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,
PRIMARY KEY (id),
UNIQUE KEY ux_undo_log (xid, branch_id)
) ENGINE = InnoDB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8;
# Amount
DROP DATABASE IF EXISTS seata_amount;
CREATE DATABASE seata_amount;
CREATE TABLE seata_amount.account
(
id INT(11) NOT NULL AUTO_INCREMENT,
balance DOUBLE DEFAULT NULL,
last_update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id)
) ENGINE = InnoDB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8;
CREATE TABLE seata_amount.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,
PRIMARY KEY (id),
UNIQUE KEY ux_undo_log (xid, branch_id)
) ENGINE = InnoDB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8;
INSERT INTO seata_amount.account (id, balance) VALUES (1, 1);
其中,每个库中的
undo_log
表,是 Seata AT 模式必须创建的表,主要用于分支事务的回滚。
另外,考虑到测试方便,我们插入了一条
id = 1
的
account
记录,和一条
id = 1
的
product
记录。
3.2 订单服务
新建
labx-17-sc-seata-at-feign-demo-order-service
项目,作为订单服务。它主要提供
/order/create
接口,实现下单逻辑。项目如下图所示:
3.2.1 引入依赖
创建
pom.xml
文件,引入相关的依赖。内容如下:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>labx-17-sc-seata-at-feign-demo</artifactId>
<groupId>cn.iocoder.springboot.labs</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>labx-17-sc-seata-at-feign-demo-order-service</artifactId>
<properties>
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
<spring.boot.version>2.2.4.RELEASE</spring.boot.version>
<spring.cloud.version>Hoxton.SR1</spring.cloud.version>
<spring.cloud.alibaba.version>2.2.0.RELEASE</spring.cloud.alibaba.version>
</properties>
<!--
引入 Spring Boot、Spring Cloud、Spring Cloud Alibaba 三者 BOM 文件,进行依赖版本的管理,防止不兼容。
在 https://dwz.cn/mcLIfNKt 文章中,Spring Cloud Alibaba 开发团队推荐了三者的依赖关系
-->
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>${spring.boot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring.cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-dependencies</artifactId>
<version>${spring.cloud.alibaba.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<!-- 实现对 Spring MVC 的自动化配置 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- 实现对数据库连接池的自动化配置 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency> <!-- 本示例,我们使用 MySQL -->
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.48</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid-spring-boot-starter</artifactId>
<version>1.1.10</version>
</dependency>
<!-- 实现对 MyBatis 的自动化配置 -->
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>2.1.2</version>
</dependency>
<!-- 引入 Spring Cloud Alibaba Seata 相关依赖,使用 Seata 实现分布式事务,并实现对其的自动配置 -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-seata</artifactId>
</dependency>
<dependency> <!-- 主要想使用 seata 1.1.0 版本 -->
<groupId>io.seata</groupId>
<artifactId>seata-spring-boot-starter</artifactId>
<version>1.1.0</version>
</dependency>
<!-- 引入 Spring Cloud Alibaba Nacos Discovery 相关依赖,将 Nacos 作为注册中心,并实现对其的自动配置 -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
<!-- 引入 Spring Cloud OpenFeign 相关依赖,使用 OpenFeign 提供声明式调用,并实现对其的自动配置 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
</dependencies>
</project>
① 引入
spring-cloud-alibaba-seata
依赖,引入 Spring Cloud Alibaba Seata 相关依赖,使用 Seata 实现分布式事务,并实现对其的自动配置。
同时,额外引入
seata-spring-boot-starter
依赖,因为我们想使用 Seata 最新 1.1.0 版本。
② 引入
spring-cloud-starter-alibaba-nacos-discovery
依赖,引入 Spring Cloud Alibaba Nacos Discovery 相关依赖,将 Nacos 作为注册中心,并实现对其的自动配置。
③ 引入
spring-cloud-starter-openfeign
依赖,引入 Spring Cloud OpenFeign 相关依赖,使用 OpenFeign 提供声明式调用,并实现对其的自动配置。
3.2.2 配置文件
创建
application.yaml
配置文件,添加相关的配置项。内容如下:
server:
port: 8081 # 端口
spring:
application:
name: order-service
datasource:
url: jdbc:mysql://127.0.0.1:3306/seata_order?useSSL=false&useUnicode=true&characterEncoding=UTF-8
driver-class-name: com.mysql.jdbc.Driver
username: root
password:
cloud:
# Nacos 作为注册中心的配置项
nacos:
discovery:
server-addr: 127.0.0.1:8848
# Seata 配置项,对应 SeataProperties 类
seata:
application-id: ${spring.application.name} # Seata 应用编号,默认为 ${spring.application.name}
tx-service-group: ${spring.application.name}-group # Seata 事务组编号,用于 TC 集群名
# Seata 服务配置项,对应 ServiceProperties 类
service:
# 虚拟组和分组的映射
vgroup-mapping:
order-service-group: default
# Seata 注册中心配置项,对应 RegistryProperties 类
registry:
type: nacos # 注册中心类型,默认为 file
nacos:
cluster: default # 使用的 Seata 分组
namespace: # Nacos 命名空间
serverAddr: localhost # Nacos 服务地址
①
spring.datasource
配置项,设置连接
seata_order
库。
②
spring.cloud.nacos.discovery
配置项,设置使用 Nacos 作为 Spring Cloud 注册中心的配置项。
③
seata
配置项,设置 Seata 的配置项目,对应 SeataProperties 类。
-
配置项,对应 Seata 应用编号,默认为application-id
。实际上,可以不进行设置。${spring.application.name}
-
配置项,Seata 事务组编号,用于 TC 集群名。tx-service-group
④
seata.service
配置项,设置 Seata 服务配置项,对应 ServiceProperties 类。它主要用于 Seata 在事务分组的特殊设计,可见《Seata 文档 —— 事务分组专题》。如果不能理解的胖友,可以见如下图:
简单来说,就是多了一层虚拟映射。这里,我们不设置
seata.service.grouplist
配置项,因为从注册中心加载 Seata TC Server 的地址。
友情提示:可能胖友对 seata.service.grouplist
配置项有点懵逼,继续往下看就会明白了。
④
seata.registry
配置项,设置 Seata 注册中心配置项,对应 RegistryProperties 类。
-
配置项,设置注册中心的类型,默认为type
。这里,我们设置为false
来使用 Nacos 作为注册中心。nacos
-
配置项,设置 Nacos 注册中心的配置项。nacos
注意!!!我们需要将 Seata TC Server 注册到 Nacos 注册中心中。不会的胖友,《芋道 Seata 极简入门》文章的「3. 部署集群 TC Server」小节。
可能胖友不想从 Nacos 注册中心来读取 Seata TC Server 的地址,可以使用艿艿额外提供的
application-file.yaml
配置文件,覆盖到
application.yaml
配置文件中即可。内容如下:
server:
port: 8081 # 端口
spring:
application:
name: order-service
datasource:
url: jdbc:mysql://127.0.0.1:3306/seata_order?useSSL=false&useUnicode=true&characterEncoding=UTF-8
driver-class-name: com.mysql.jdbc.Driver
username: root
password:
cloud:
# Nacos 作为注册中心的配置项
nacos:
discovery:
server-addr: 127.0.0.1:8848
# Seata 配置项,对应 SeataProperties 类
seata:
application-id: ${spring.application.name} # Seata 应用编号,默认为 ${spring.application.name}
tx-service-group: ${spring.application.name}-group # Seata 事务组编号,用于 TC 集群名
# 服务配置项,对应 ServiceProperties 类
service:
# 虚拟组和分组的映射
vgroup-mapping:
order-service-group: default
# 分组和 Seata 服务的映射
grouplist:
default: 127.0.0.1:8091
差异点在于,删除了
seata.registry
配置项,增加
seata.service.grouplist
配置项来替代。原因如下图:
3.2.3 OrderController
创建 OrderController 类,提供
order/create
下单 HTTP API。代码如下:
@RestController
@RequestMapping("/order")
public class OrderController {
private Logger logger = LoggerFactory.getLogger(OrderController.class);
@Reference
private OrderService orderService;
@PostMapping("/create")
public Integer createOrder(@RequestParam("userId") Long userId,
@RequestParam("productId") Long productId,
@RequestParam("price") Integer price) throws Exception {
logger.info("[createOrder] 收到下单请求,用户:{}, 商品:{}, 价格:{}", userId, productId, price);
return orderService.createOrder(userId, productId, price);
}
}
- 该 API 中,会调用 OrderService 进行下单。
友情提示:因为这个是示例项目,所以直接传入 price
参数,作为订单的金额,实际肯定不是这样的,哈哈哈~
3.2.4 OrderService
创建 OrderService 接口,定义了创建订单的方法。代码如下:
/**
* 订单 Service
*/
public interface OrderService {
/**
* 创建订单
*
* @param userId 用户编号
* @param productId 产品编号
* @param price 价格
* @return 订单编号
* @throws Exception 创建订单失败,抛出异常
*/
Integer createOrder(Long userId, Long productId, Integer price) throws Exception;
}
3.2.5 OrderServiceImpl
创建 OrderServiceImpl 类,实现创建订单的方法。代码如下:
@Service
public class OrderServiceImpl implements OrderService {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private OrderDao orderDao;
@Autowired
private AccountServiceFeignClient accountService;
@Autowired
private ProductServiceFeignClient productService;
@Override
@GlobalTransactional // <1>
public Integer createOrder(Long userId, Long productId, Integer price) {
Integer amount = 1; // 购买数量,暂时设置为 1。
logger.info("[createOrder] 当前 XID: {}", RootContext.getXID());
// <2> 扣减库存
productService.reduceStock(new ProductReduceStockDTO().setProductId(productId).setAmount(amount));
// <3> 扣减余额
accountService.reduceBalance(new AccountReduceBalanceDTO().setUserId(userId).setPrice(price));
// <4> 保存订单
OrderDO order = new OrderDO().setUserId(userId).setProductId(productId).setPayAmount(amount * price);
orderDao.saveOrder(order);
logger.info("[createOrder] 保存订单: {}", order.getId());
// 返回订单编号
return order.getId();
}
}
<1>
处,在类上,添加 Seata
@GlobalTransactional
注解,声明全局事务。
<2>
处,调用 ProductServiceFeignClient 的
#reduceStock(productId, amount)
方法,通过 Feign 远程 HTTP 调用商品服务,进行扣除库存。代码如下:
// ProductServiceFeignClient.java
/**
* `product-service` 服务的 Feign 客户端
*/
@FeignClient(name = "product-service")
public interface ProductServiceFeignClient {
@PostMapping("/product/reduce-stock")
void reduceStock(@RequestBody ProductReduceStockDTO productReduceStockDTO);
}
// ProductReduceStockDTO.java
/**
* 商品减少库存 DTO
*/
public class ProductReduceStockDTO {
/** 商品编号 */
private Long productId;
/** 数量 */
private Integer amount;
// ... 省略 setter/getter 方法
}
- 远程调用失败,又或是扣除库存失败,都会抛出 Exception 异常,从而回滚 Seata 全局事务。
<3>
处,调用 AccountServiceFeignClient 的
#reduceBalance(userId, price)
方法,通过 Feign 远程 HTTP 调用账户服务,进行扣除余额。代码如下:
// AccountServiceFeignClient.java
/**
* `account-service` 服务的 Feign 客户端
*/
@FeignClient(name = "account-service")
public interface AccountServiceFeignClient {
@PostMapping("/account/reduce-balance")
void reduceBalance(@RequestBody AccountReduceBalanceDTO accountReduceBalanceDTO);
}
// AccountReduceBalanceDTO.java
/**
* 账户减少余额 DTO
*/
public class AccountReduceBalanceDTO {
/** 用户编号 */
private Long userId;
/** 扣减金额 */
private Integer price;
// ... 省略 setter/getter 方法
}
- 远程调用失败,又或是扣除余额失败,都会抛出 Exception 异常,从而回滚 Seata 全局事务。
<4>
处,在全部调用成功后,调用 OrderDao 保存订单。
3.2.6 OrderDao
创建 OrderDao 接口,定义保存订单的操作。代码如下:
@Mapper
@Repository
public interface OrderDao {
/**
* 插入订单记录
*
* @param order 订单
* @return 影响记录数量
*/
@Insert("INSERT INTO orders (user_id, product_id, pay_amount) VALUES (#{userId}, #{productId}, #{payAmount})")
@Options(useGeneratedKeys = true, keyColumn = "id", keyProperty = "id")
int saveOrder(OrderDO order);
}
其中,OrderDO 实体类,对应
orders
表。代码如下:
/**
* 订单实体
*/
public class OrderDO {
/** 订单编号 **/
private Integer id;
/** 用户编号 **/
private Long userId;
/** 产品编号 **/
private Long productId;
/** 支付金额 **/
private Integer payAmount;
// ... 省略 setter/getter 方法
}
3.2.7 OrderServiceApplication
创建 OrderServiceApplication 类,用于启动订单服务。代码如下:
@SpringBootApplication
@EnableFeignClients
public class OrderServiceApplication {
public static void main(String[] args) {
SpringApplication.run(OrderServiceApplication.class, args);
}
}
3.3 商品服务
新建
labx-17-sc-seata-at-feign-demo-product-service
项目,作为商品服务。它主要提供
/product/reduce-stock
接口,实现扣除商品的库存逻辑。项目如下图所示:
3.3.1 引入依赖
创建
pom.xml
文件,引入相关的依赖。和「3.2.1 引入依赖」是一致的,就不重复“贴”出来了,胖友点击
pom.xml
文件查看。
3.3.2 配置文件
创建
application.yaml
配置文件,添加相关的配置项。和「3.2.2 配置文件」是一致的,就不重复“贴”出来了,胖友点击
application.yaml
文件查看。
3.3.3 ProductController
创建 ProductController 类,提供
/product/reduce-stock
扣除库存 HTTP API。代码如下:
@RestController
@RequestMapping("/product")
public class ProductController {
private Logger logger = LoggerFactory.getLogger(ProductController.class);
@Autowired
private ProductService productService;
@PostMapping("/reduce-stock")
public void reduceStock(@RequestBody ProductReduceStockDTO productReduceStockDTO)
throws Exception {
logger.info("[reduceStock] 收到减少库存请求, 商品:{}, 价格:{}", productReduceStockDTO.getProductId(),
productReduceStockDTO.getAmount());
productService.reduceStock(productReduceStockDTO.getProductId(), productReduceStockDTO.getAmount());
}
}
- 该 API 中,会调用 ProductService 进行扣除库存。
其中,ProductReduceStockDTO 为商品减少库存 DTO 类,代码如下:
public class ProductReduceStockDTO {
/**
* 商品编号
*/
private Long productId;
/**
* 数量
*/
private Integer amount;
// ... 省略 setter/getter 方法
}
3.3.4 ProductService
创建 ProductService 接口,定义了扣除库存的方法。代码如下:
/**
* 商品 Service
*/
public interface ProductService {
/**
* 扣减库存
*
* @param productId 商品 ID
* @param amount 扣减数量
* @throws Exception 扣减失败时抛出异常
*/
void reduceStock(Long productId, Integer amount) throws Exception;
}
3.3.5 ProductServiceImpl
创建 ProductServiceImpl 类,实现扣减库存的方法。代码如下:
@Service
public class ProductServiceImpl implements ProductService {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private ProductDao productDao;
@Override
@Transactional // <1> 开启新事物
public void reduceStock(Long productId, Integer amount) throws Exception {
logger.info("[reduceStock] 当前 XID: {}", RootContext.getXID());
// <2> 检查库存
checkStock(productId, amount);
logger.info("[reduceStock] 开始扣减 {} 库存", productId);
// <3> 扣减库存
int updateCount = productDao.reduceStock(productId, amount);
// 扣除成功
if (updateCount == 0) {
logger.warn("[reduceStock] 扣除 {} 库存失败", productId);
throw new Exception("库存不足");
}
// 扣除失败
logger.info("[reduceStock] 扣除 {} 库存成功", productId);
}
private void checkStock(Long productId, Integer requiredAmount) throws Exception {
logger.info("[checkStock] 检查 {} 库存", productId);
Integer stock = productDao.getStock(productId);
if (stock < requiredAmount) {
logger.warn("[checkStock] {} 库存不足,当前库存: {}", productId, stock);
throw new Exception("库存不足");
}
}
}
<1>
处,在类上,添加了 Spring
@Transactional
注解,声明本地事务。也就是说,此处会开启一个
seata_product
库的数据库事务。
<2>
处,检查库存是否足够,如果不够则抛出 Exception 异常。因为我们需要通过异常,回滚全局异常。
<3>
处,进行扣除库存,如果扣除失败则抛出 Exception 异常。
3.3.6 ProductDao
创建 ProductDao 接口,定义获取和扣除库存的操作。代码如下:
@Mapper
@Repository
public interface ProductDao {
/**
* 获取库存
*
* @param productId 商品编号
* @return 库存
*/
@Select("SELECT stock FROM product WHERE id = #{productId}")
Integer getStock(@Param("productId") Long productId);
/**
* 扣减库存
*
* @param productId 商品编号
* @param amount 扣减数量
* @return 影响记录行数
*/
@Update("UPDATE product SET stock = stock - #{amount} WHERE id = #{productId} AND stock >= #{amount}")
int reduceStock(@Param("productId") Long productId, @Param("amount") Integer amount);
}
3.3.7 ProductServiceApplication
创建 ProductServiceApplication 类,用于启动商品服务。代码如下:
@SpringBootApplication
public class ProductServiceApplication {
public static void main(String[] args) {
SpringApplication.run(ProductServiceApplication.class, args);
}
}
3.4 账户服务
新建
labx-17-sc-seata-at-feign-demo-account-service
项目,作为账户服务。它主要提供
/account/reduce-balance
接口,实现扣除账户的余额逻辑。项目如下图所示:
3.4.1 引入依赖
创建
pom.xml
文件,引入相关的依赖。和「3.2.1 引入依赖」是一致的,就不重复“贴”出来了,胖友点击
pom.xml
文件查看。
3.4.2 配置文件
创建
application.yaml
配置文件,添加相关的配置项。和「3.2.2 配置文件」是一致的,就不重复“贴”出来了,胖友点击
application.yaml
文件查看。
3.4.3 AccountController
创建 AccountController 类,提供
/account/reduce-balance
扣除余额 HTTP API。代码如下:
@RestController
@RequestMapping("/account")
public class AccountController {
private Logger logger = LoggerFactory.getLogger(AccountController.class);
@Autowired
private AccountService accountService;
@PostMapping("/reduce-balance")
public void reduceBalance(@RequestBody AccountReduceBalanceDTO accountReduceBalanceDTO) throws Exception {
logger.info("[reduceBalance] 收到减少余额请求, 用户:{}, 金额:{}", accountReduceBalanceDTO.getUserId(),
accountReduceBalanceDTO.getPrice());
accountService.reduceBalance(accountReduceBalanceDTO.getUserId(), accountReduceBalanceDTO.getPrice());
}
}
- 该 API 中,会调用 AccountService 进行扣除余额。
其中,AccountReduceBalanceDTO 为账户减少余额 DTO 类,代码如下:
public class AccountReduceBalanceDTO {
/**
* 用户编号
*/
private Long userId;
/**
* 扣减金额
*/
private Integer price;
public Long getUserId() {
return userId;
}
// ... 省略 setter/getter 方法
}
3.4.4 AccountService
创建 AccountService 类,定义扣除余额的方法。代码如下:
/**
* 账户 Service
*/
public interface AccountService {
/**
* 扣除余额
*
* @param userId 用户编号
* @param price 扣减金额
* @throws Exception 失败时抛出异常
*/
void reduceBalance(Long userId, Integer price) throws Exception;
}
3.4.5 AccountServiceImpl
创建 AccountServiceImpl 类,实现扣除余额的方法。代码如下:
@Service
public class AccountServiceImpl implements AccountService {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private AccountDao accountDao;
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW) // <1> 开启新事物
public void reduceBalance(Long userId, Integer price) throws Exception {
logger.info("[reduceBalance] 当前 XID: {}", RootContext.getXID());
// <2> 检查余额
checkBalance(userId, price);
logger.info("[reduceBalance] 开始扣减用户 {} 余额", userId);
// <3> 扣除余额
int updateCount = accountDao.reduceBalance(price);
// 扣除成功
if (updateCount == 0) {
logger.warn("[reduceBalance] 扣除用户 {} 余额失败", userId);
throw new Exception("余额不足");
}
logger.info("[reduceBalance] 扣除用户 {} 余额成功", userId);
}
private void checkBalance(Long userId, Integer price) throws Exception {
logger.info("[checkBalance] 检查用户 {} 余额", userId);
Integer balance = accountDao.getBalance(userId);
if (balance < price) {
logger.warn("[checkBalance] 用户 {} 余额不足,当前余额:{}", userId, balance);
throw new Exception("余额不足");
}
}
}
<1>
处,在类上,添加了 Spring
@Transactional
注解,声明本地事务。也就是说,此处会开启一个
seata_account
库的数据库事务。
<2>
处,检查余额是否足够,如果不够则抛出 Exception 异常。因为我们需要通过异常,回滚全局异常。
<3>
处,进行扣除余额,如果扣除失败则抛出 Exception 异常。
3.4.6 AccountDao
创建 AccountDao 接口,定义查询和扣除余额的操作。代码如下:
@Mapper
@Repository
public interface AccountDao {
/**
* 获取账户余额
*
* @param userId 用户 ID
* @return 账户余额
*/
@Select("SELECT balance FROM account WHERE id = #{userId}")
Integer getBalance(@Param("userId") Long userId);
/**
* 扣减余额
*
* @param price 需要扣减的数目
* @return 影响记录行数
*/
@Update("UPDATE account SET balance = balance - #{price} WHERE id = 1 AND balance >= ${price}")
int reduceBalance(@Param("price") Integer price);
}
3.4.7 AccountServiceApplication
创建 AccountServiceApplication 类,用于启动商品服务。代码如下:
@SpringBootApplication
public class AccountServiceApplication {
public static void main(String[] args) {
SpringApplication.run(AccountServiceApplication.class, args);
}
}
3.5 简单测试
下面,我们将测试两种情况:
- 分布式事务正常提交
- 分布式事务异常回滚
Debug 执行 OrderServiceApplication 启动订单服务。此时,我们可以看到 Seata 相关日志如下:
友情提示:日志的顺序,艿艿做了简单的整理,为了更容易阅读。
# ... 上面还有 Seata 相关 Bean 初始化的日志,忘记加进来了,嘿嘿~
# 连接到 Seata TC Server 服务器
2020-04-05 10:51:00.892 INFO 52124 --- [ main] i.s.c.r.netty.NettyClientChannelManager : will connect to 127.0.0.1:8091
2020-04-05 10:51:00.893 INFO 52124 --- [ main] i.s.core.rpc.netty.NettyPoolableFactory : NettyPool create channel to transactionRole:RMROLE,address:127.0.0.1:8091,msg:< RegisterRMRequest{resourceIds='jdbc:mysql://127.0.0.1:3306/seata_order', applicationId='order-service', transactionServiceGroup='order-service-group'} >
# 加载 Seata 序列化器
2020-04-05 10:51:01.042 INFO 52124 --- [lector_RMROLE_1] i.s.common.loader.EnhancedServiceLoader : load Serializer[SEATA] extension by class[io.seata.serializer.seata.SeataSerializer]
# 注册 Seata Resource Manager 到 Seata TC Server 成功
2020-04-05 10:51:00.892 INFO 52124 --- [ main] io.seata.core.rpc.netty.RmRpcClient : RM will register :jdbc:mysql://127.0.0.1:3306/seata_order
2020-04-05 10:51:01.054 INFO 52124 --- [ main] io.seata.core.rpc.netty.RmRpcClient : register RM success. server version:1.1.0,channel:[id: 0xc7553923, L:/127.0.0.1:64449 - R:/127.0.0.1:8091]
2020-04-05 10:51:01.061 INFO 52124 --- [ main] i.s.core.rpc.netty.NettyPoolableFactory : register success, cost 34 ms, version:1.1.0,role:RMROLE,channel:[id: 0xc7553923, L:/127.0.0.1:64449 - R:/127.0.0.1:8091]
# ... 上面还有 Seata 相关 Bean 初始化的日志,忘记加进来了,嘿嘿~
# 给数据源增加 Seata 的数据源代理
2020-04-06 22:14:03.201 INFO 99838 --- [ main] s.s.a.d.SeataDataSourceBeanPostProcessor : Auto proxy of [dataSource]
# 加载 Druid 提供的 SQL 解析器
2020-04-06 22:14:03.443 INFO 99838 --- [ main] i.s.common.loader.EnhancedServiceLoader : load DbTypeParser[druid] extension by class[io.seata.sqlparser.druid.DruidDelegatingDbTypeParser]
# 加载 Seata 使用 Nacos 注册中心的拓展,从而加载 Seata TC Server 的地址
2020-04-06 22:14:03.462 INFO 99838 --- [ main] i.s.common.loader.EnhancedServiceLoader : load RegistryProvider[Nacos] extension by class[io.seata.discovery.registry.nacos.NacosRegistryProvider]
# 连接到某个 Seata TC Server 服务器
2020-04-06 22:14:03.787 INFO 99838 --- [ main] i.s.c.r.netty.NettyClientChannelManager : will connect to 10.37.129.2:18091
2020-04-06 22:14:03.787 INFO 99838 --- [ main] io.seata.core.rpc.netty.RmRpcClient : RM will register :jdbc:mysql://127.0.0.1:3306/seata_order
2020-04-06 22:14:03.789 INFO 99838 --- [ main] i.s.core.rpc.netty.NettyPoolableFactory : NettyPool create channel to transactionRole:RMROLE,address:10.37.129.2:18091,msg:< RegisterRMRequest{resourceIds='jdbc:mysql://127.0.0.1:3306/seata_order', applicationId='order-service', transactionServiceGroup='order-service-group'} >
# 加载 Seata 序列化器
2020-04-06 22:14:03.911 INFO 99838 --- [lector_RMROLE_1] i.s.common.loader.EnhancedServiceLoader : load Serializer[SEATA] extension by class[io.seata.serializer.seata.SeataSerializer]
2020-04-06 22:14:03.931 INFO 99838 --- [ main] io.seata.core.rpc.netty.RmRpcClient : register RM success. server version:1.1.0,channel:[id: 0xe74ff08c, L:/10.37.129.2:61977 - R:/10.37.129.2:18091]
# 注册 Seata Resource Manager 到某个 Seata TC Server 成功
2020-04-06 22:14:03.940 INFO 99838 --- [ main] i.s.core.rpc.netty.NettyPoolableFactory : register success, cost 59 ms, version:1.1.0,role:RMROLE,channel:[id: 0xe74ff08c, L:/10.37.129.2:61977 - R:/10.37.129.2:18091]
# 连接到另一个 Seata TC Server 服务器
2020-04-06 22:14:03.941 INFO 99838 --- [ main] i.s.c.r.netty.NettyClientChannelManager : will connect to 10.37.129.2:28091
# 注册 Seata Resource Manager 到另一个 Seata TC Server 成功
2020-04-06 22:14:03.941 INFO 99838 --- [ main] io.seata.core.rpc.netty.RmRpcClient : RM will register :jdbc:mysql://127.0.0.1:3306/seata_order
2020-04-06 22:14:03.941 INFO 99838 --- [ main] i.s.core.rpc.netty.NettyPoolableFactory : NettyPool create channel to transactionRole:RMROLE,address:10.37.129.2:28091,msg:< RegisterRMRequest{resourceIds='jdbc:mysql://127.0.0.1:3306/seata_order', applicationId='order-service', transactionServiceGroup='order-service-group'} >
2020-04-06 22:14:03.945 INFO 99838 --- [ main] io.seata.core.rpc.netty.RmRpcClient : register RM success. server version:1.1.0,channel:[id: 0xa147094f, L:/10.37.129.2:61978 - R:/10.37.129.2:28091]
2020-04-06 22:14:03.945 INFO 99838 --- [ main] i.s.core.rpc.netty.NettyPoolableFactory : register success, cost 3 ms, version:1.1.0,role:RMROLE,channel:[id: 0xa147094f, L:/10.37.129.2:61978 - R:/10.37.129.2:28091]
# 因为 OrderServiceImpl 添加了 `@GlobalTransactional` 注解,所以创建其代理,用于全局事务。
2020-04-06 22:14:04.370 INFO 99838 --- [ main] i.s.s.a.GlobalTransactionScanner : Bean[cn.iocoder.springcloud.labx17.orderservice.service.impl.OrderServiceImpl] with name [orderServiceImpl] would use interceptor [io.seata.spring.annotation.GlobalTransactionalInterceptor]
执行 ProductServiceApplication 启动商品服务。相关的日志,胖友自己瞅瞅。
执行 AccountServiceApplication 启动账户服务。相关的日志,胖友自己瞅瞅。
3.5.1 正常流程
① 先查询下目前数据库的数据情况。如下图所示:
② 使用 Postman 模拟调用 http://127.0.0.1:8081/order/create 创建订单的接口,如下图所示:
此时,在控制台打印日志如下图所示:
- 订单服务:
- 商品服务:
- 账户服务:
再查询下目前数据库的数据情况。如下图所示:
3.5.2 异常流程
① 先查询下目前数据库的数据情况。如下图所示:
② 在 OrderServiceImpl 的
#createOrder(...)
方法上,打上断点如下图,方便我们看到
product
表的
balance
被减少:
友情提示:这里忘记截图了,稍后 IDEA 停留在该断点时,胖友可以去查询表,会发现
product
已经减少。
balance
③ 使用 Postman 模拟调用 http://127.0.0.1:8081/order/create 创建订单的接口,如下图所示:
此时,在控制台打印日志如下图所示:
- 订单服务:
- 商品服务:
- 账户服务:
再查询下目前数据库的数据情况。如下图所示: