Spring Cloud 中的分布式事务，附源码《一》

回复“1024”或“面试题”获取4T架构师资料

投稿作者：小东啊 微信公众号：小东IT技术分享| ID：xiaodong_tech

一. 需求

在开发我的开源项目 prex 时，加入工作流，解决工作流用户与当前系统用户同步问题时，涉及到远程调用操作两个数据库所产生的事务问题，比如系统用户在增加用户同步工作流用户时，系统用户添加成功，工作流用户没有添加成功，则造成数据不一致问题，本地事务无法回滚,那么则使用分布式事务解决方案。

开源项目:

https://gitee.com/kaiyuantuandui/prex

二. 什么是分布式事务?

指一次大的操作由不同的小操作组成的，这些小的操作分布在不同的服务器上，分布式事务需要保证这些小操作要么全部成功，要么全部失败。从本质上来说，分布式事务就是为了保证不同数据库的数据一致性。

通俗一点说就是单体应用被拆分成微服务应用，原来的一个模块被拆分成三个独立的应用，分别使用独立的数据源，业务操作需要调用三个服务来完成。

三. 分布式事务解决方案

分布式事务作为微服务应用中的大难题，在现有的解决方案中，个人认为 Seata 是目前最轻量的解决方案

四. Seata 是什么?

Seata 是一款开源的分布式事务解决方案，致力于提供高性能和简单易用的分布式事务服务。 Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式，为用户打造一站式的分布式解决方案。

AT 模式

前提

1、基于支持本地 ACID 事务的关系型数据库。

2、Java 应用，通过 JDBC 访问数据库。

整体机制

两阶段提交协议的演变：

一阶段：

业务数据和回滚日志记录在同一个本地事务中提交，释放本地锁和连接资源。

二阶段：

1、提交异步化，非常快速地完成。 2、回滚通过一阶段的回滚日志进行反向补偿。

写隔离

1、一阶段本地事务提交前，需要确保先拿到全局锁 。

2、拿不到 全局锁 ，不能提交本地事务。

3、拿 全局锁 的尝试被限制在一定范围内，超出范围将放弃，并回滚本地事务，释放本地锁。

以一个示例来说明：

两个全局事务 tx1 和 tx2，分别对 a 表的 m 字段进行更新操作，m 的初始值 1000。

tx1 先开始，开启本地事务，拿到本地锁，更新操作 m = 1000 - 100 = 900。本地事务提交前，先拿到该记录的 全局锁 ，本地提交释放本地锁。 tx2 后开始，开启本地事务，拿到本地锁，更新操作 m = 900 - 100 = 800。本地事务提交前，尝试拿该记录的 全局锁 ，tx1 全局提交前，该记录的全局锁被 tx1 持有，tx2 需要重试等待 全局锁 。

tx1 二阶段全局提交，释放 全局锁 。tx2 拿到 全局锁 提交本地事务。

如果 tx1 的二阶段全局回滚，则 tx1 需要重新获取该数据的本地锁，进行反向补偿的更新操作，实现分支的回滚。

此时，如果 tx2 仍在等待该数据的 全局锁，同时持有本地锁，则 tx1 的分支回滚会失败。分支的回滚会一直重试，直到 tx2 的 全局锁 等锁超时，放弃 全局锁 并回滚本地事务释放本地锁，tx1 的分支回滚最终成功。

因为整个过程 全局锁 在 tx1 结束前一直是被 tx1 持有的，所以不会发生 脏写 的问题。

读隔离

在数据库本地事务隔离级别 读已提交（Read Committed） 或以上的基础上，Seata（AT 模式）的默认全局隔离级别是 读未提交（Read Uncommitted） 。

如果应用在特定场景下，必需要求全局的 读已提交 ，目前 Seata 的方式是通过 SELECT FOR UPDATE 语句的代理。

SELECT FOR UPDATE 语句的执行会申请 全局锁 ，如果 全局锁 被其他事务持有，则释放本地锁（回滚 SELECT FOR UPDATE 语句的本地执行）并重试。这个过程中，查询是被 block 住的，直到 全局锁 拿到，即读取的相关数据是 已提交 的，才返回。

出于总体性能上的考虑，Seata 目前的方案并没有对所有 SELECT 语句都进行代理，仅针对 FOR UPDATE 的 SELECT 语句。

工作机制

以一个示例来说明整个 AT 分支的工作过程。

业务表：product

Field	Type	Key
id	bigint(20)	PRI
name	varchar(100)
since	varchar(100)

AT 分支事务的业务逻辑：

update product set name = 'GTS' where name = 'TXC';

一阶段

过程：

1、解析 SQL：得到 SQL 的类型（UPDATE），表（product），条件（where name = 'TXC'）等相关的信息。

2、查询前镜像：根据解析得到的条件信息，生成查询语句，定位数据。

1. select id, name, since from product where name = 'TXC';

得到前镜像：

id name since 1 TXC 2014

3、执行业务 SQL：更新这条记录的 name 为 'GTS'。

4、查询后镜像：根据前镜像的结果，通过 主键 定位数据。

1. select id, name, since from product where id = 1`;

得到后镜像：

id name since 1 GTS 2014

5、插入回滚日志：把前后镜像数据以及业务 SQL 相关的信息组成一条回滚日志记录，插入到 UNDO_LOG 表中。

1. {

2. "branchId": 641789253,

3. "undoItems": [{

4. "afterImage": {

5. "rows": [{

6. "fields": [{

7. "name": "id",

8. "type": 4,

9. "value": 1

10. }, {

11. "name": "name",

12. "type": 12,

13. "value": "GTS"

14. }, {

15. "name": "since",

16. "type": 12,

17. "value": "2014"

18. }]

19. }],

20. "tableName": "product"

21. },

22. "beforeImage": {

23. "rows": [{

24. "fields": [{

25. "name": "id",

26. "type": 4,

27. "value": 1

28. }, {

29. "name": "name",

30. "type": 12,

31. "value": "TXC"

32. }, {

33. "name": "since",

34. "type": 12,

35. "value": "2014"

36. }]

37. }],

38. "tableName": "product"

39. },

40. "sqlType": "UPDATE"

41. }],

42. "xid": "xid:xxx"

43. }

6、提交前，向 TC 注册分支：申请 product 表中，主键值等于 1 的记录的 全局锁 。

7、本地事务提交：业务数据的更新和前面步骤中生成的 UNDO LOG 一并提交。

8、将本地事务提交的结果上报给 TC。

二阶段-回滚

1、收到 TC 的分支回滚请求，开启一个本地事务，执行如下操作。

2、通过 XID 和 Branch ID 查找到相应的 UNDO LOG 记录。

3、数据校验：拿 UNDO LOG 中的后镜与当前数据进行比较，如果有不同，说明数据被当前全局事务之外的动作做了修改。这种情况，需要根据配置策略来做处理，详细的说明在另外的文档中介绍。

根据 UNDO LOG 中的前镜像和业务 SQL 的相关信息生成并执行回滚的语句：

1. update product set name = 'TXC' where id = 1;

4、提交本地事务。并把本地事务的执行结果（即分支事务回滚的结果）上报给 TC。

二阶段-提交

1、收到 TC 的分支提交请求，把请求放入一个异步任务的队列中，马上返回提交成功的结果给 TC。

2、异步任务阶段的分支提交请求将异步和批量地删除相应 UNDO LOG 记录。

附录

回滚日志表

UNDO_LOG Table：不同数据库在类型上会略有差别。

以 MySQL 为例：

Field	Type
branch_id	bigint PK
xid	varchar(100)
context	varchar(128)
rollback_info	longblob
log_status	tinyint
log_created	datetime
log_modified	datetime

1. -- 注意此处0.7.0+ 增加字段 context

2. CREATE TABLE `undo_log` (

3. `id` bigint(20) NOT NULL AUTO_INCREMENT,

4. `branch_id` bigint(20) NOT NULL,

5. `xid` varchar(100) NOT NULL,

6. `context` varchar(128) NOT NULL,

7. `rollback_info` longblob NOT NULL,

8. `log_status` int(11) NOT NULL,

9. `log_created` datetime NOT NULL,

10. `log_modified` datetime NOT NULL,

11. PRIMARY KEY (`id`),

12. UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)

13. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

TCC 模式

回顾总览中的描述：一个分布式的全局事务，整体是 两阶段提交 的模型。全局事务是由若干分支事务组成的，分支事务要满足 两阶段提交 的模型要求，即需要每个分支事务都具备自己的：

一阶段 prepare 行为

二阶段 commit 或 rollback 行为

根据两阶段行为模式的不同，我们将分支事务划分为 Automatic (Branch) Transaction Mode 和 Manual (Branch) Transaction Mode.

AT 模式（参考链接 TBD）基于 支持本地 ACID 事务 的 关系型数据库：

一阶段 prepare 行为：在本地事务中，一并提交业务数据更新和相应回滚日志记录。

二阶段 commit 行为：马上成功结束，自动 异步批量清理回滚日志。

二阶段 rollback 行为：通过回滚日志，自动 生成补偿操作，完成数据回滚。

相应的，TCC 模式，不依赖于底层数据资源的事务支持：

一阶段 prepare 行为：调用 自定义 的 prepare 逻辑。

二阶段 commit 行为：调用 自定义 的 commit 逻辑。

二阶段 rollback 行为：调用 自定义 的 rollback 逻辑。

所谓 TCC 模式，是指支持把 自定义 的分支事务纳入到全局事务的管理中

Saga 模式

Saga 模式是 SEATA 提供的长事务解决方案，在 Saga 模式中，业务流程中每个参与者都提交本地事务，当出现某一个参与者失败则补偿前面已经成功的参与者，一阶段正向服务和二阶段补偿服务都由业务开发实现。

理论基础：Hector & Kenneth 发表论⽂ Sagas （1987）

适用场景：

1、业务流程长、业务流程多

2、参与者包含其它公司或遗留系统服务，无法提供 - TCC 模式要求的三个接口

优势：

1、一阶段提交本地事务，无锁，高性能

2、事件驱动架构，参与者可异步执行，高吞吐

3、补偿服务易于实现

缺点：

不保证隔离性（应对方案见用户文档）

五. 准备工作

1、这里我们使用 Nacos 作为注册中心，Nacos 的安装及使用可以参考

2、我们从官网下载 seata-server，这里下载的是 seata-server-0.9.0.zip，下载地址：https://github.com/seata/seata/releases
github 地址下载速度很慢,可以在公众号后台回复 seata安装包快速获取百度云下载链接

3、下载完成后解压 seata-server 安装包到指定目录

解压完成后我们得到了几个文件夹

1、bin

存放各个系统的 seata server 启动脚本

3、conf

存在 seata server 启动时所需要的配置信息、数据库模式下所需要的建表语句

3、lib

运行 seata server 所需要的依赖包列表

配置 Seata Server

seata server所有的配置都在 conf 文件夹内，该文件夹内有两个文件我们必须要详细介绍下。

seata server默认使用 file（文件方式）进行存储事务日志、事务运行信息，我们可以通过-m db 脚本参数的形式来指定，目前仅支持 file、db 这两种方式。

1、file.conf
该文件用于配置存储方式、透传事务信息的 NIO 等信息，默认对应 registry.conf 文件内的 file 方式配置

2、registry.conf

seata server 核心配置文件，可以通过该文件配置服务注册方式、配置读取方式。

注册方式目前支持 file 、nacos 、eureka、redis、zk、consul、etcd3、sofa 等方式，默认为 file，对应读取 file.conf 内的注册方式信息。

读取配置信息的方式支持 file、nacos 、apollo、zk、consul、etcd3 等方式，默认为 file，对应读取 file.conf 文件内的配置。

修改 conf 目录下的 file.conf 配置文件，主要修改自定义事务组名称，事务日志存储模式及数据库连接信息

1. transport {

2. ...省略

3. }

4. service {

5. #vgroup->rgroup

6. vgroup_mapping.prex_tx_group = "default" #修改事务组名称为：prex_tx_group，和客户端自定义的名称对应

7. #only support single node

8. default.grouplist = "127.0.0.1:8091"

9. #degrade current not support

10. enableDegrade = false

11. #disable

12. disable = false

13. #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent

14. max.commit.retry.timeout = "-1"

15. max.rollback.retry.timeout = "-1"

16. }

17. ## transaction log store

18. store {

19. ## store mode: file、db

20. mode = "db" #修改此处将事务信息存储到db数据库中

21. ## database store

22. db {

23. ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.

24. datasource = "druid"

25. ## mysql/oracle/h2/oceanbase etc.

26. db-type = "mysql"

27. driver-class-name = "com.mysql.jdbc.Driver"

28. url = "jdbc:mysql://localhost:3306/seat" #修改数据库连接地址

29. user = "root" #修改数据库用户名

30. password = "root" #修改数据库密码

31. min-conn = 1

32. max-conn = 3

33. global.table = "global_table"

34. branch.table = "branch_table"

35. lock-table = "lock_table"

36. query-limit = 100

37. }

38. }

说明:

1、存储事务日志可以使用 file 文件和 db 数据库两种方式

2、由于我们使用了 db 模式存储事务日志，所以我们需要创建一个 seat 数据库，建表 sql 在 seata-server 的/conf/db_store.sql 中

3、修改 conf 目录下的 registry.conf配置文件，指明注册中心为 nacos，及修改 nacos 连接信息即可；

1. registry {

2. # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa

3. type = "nacos"

4. nacos {

5. serverAddr = "localhost:8848"

6. namespace = ""

7. cluster = "default"

8. }

9. ... 省略

10. }

11. }

配置完成后启动 Seata

启动 seata server 的脚本位于 bin 文件内， Linux/Mac 环境使用 seata-server.sh 脚本启动，Windows 环境使用 seata-server.bat 脚本启动。

Linux/Mac启动方式示例如下所示：

1. nohup sh seata-server.sh -p 8091 -h 127.0.0.1 -m db &> seata.log &

通过 nohup 命令让 seata server 在系统后台运行。

脚本参数：

1、-p
指定启动 seata server 的端口号。

2、-h

指定 seata server 所绑定的主机，这里配置要注意指定的主机 IP 要与业务服务内的配置文件保持一致，如：-h 192.168.1.10，业务服务配置文件内应该配置 192.168.1.10，即使在同一台主机上也要保持一致。

3、-m
事务日志、事务执行信息存储的方式，目前支持 file（文件方式）、db（数据库方式，建表语句请查看 config/dbstore.sql、config/dbundo_log.sql）

查看启动日志

当我们看到-Server started 时并未发现其他错误信息，我们的 seata server 已经启动成功

六. 实战演示

让我们从一个微服务示例开始

用户购买商品的业务逻辑。整个业务逻辑由 3 个微服务提供支持：

1、仓储服务：对给定的商品扣除仓储数量。

2、订单服务：根据采购需求创建订单。

3、帐户服务：从用户帐户中扣除余额。

架构图

数据库

创建业务数据库

db-order：存储订单的数据库
db-storage：存储库存的数据库
db-account：存储账户信息的数据库

order 订单表:

1. DROP TABLE IF EXISTS `order`;

2. CREATE TABLE `order` (

3. `id` int(20) NOT NULL AUTO_INCREMENT COMMENT '主键Id',

4. `user_id` int(20) DEFAULT NULL COMMENT '用户Id',

5. `pay_money` decimal(11,0) DEFAULT NULL COMMENT '付款金额',

6. `product_id` int(20) DEFAULT NULL COMMENT '商品Id',

7. `status` int(11) DEFAULT NULL COMMENT '状态',

8. `count` int(11) DEFAULT NULL COMMENT '商品数量',

9. PRIMARY KEY (`id`) USING BTREE

10. ) ENGINE=InnoDB DEFAULT CHARSET=latin1 ROW_FORMAT=DYNAMIC COMMENT='订单表';

11. SET FOREIGN_KEY_CHECKS = 1;

product 商品表:

1. DROP TABLE IF EXISTS `product`;

2. CREATE TABLE `product` (

3. `id` int(20) NOT NULL COMMENT '主键',

4. `product_id` int(11) DEFAULT NULL COMMENT '商品Id',

5. `price` decimal(11,0) DEFAULT NULL COMMENT '价格',

6. `count` int(11) DEFAULT NULL COMMENT '库存数量',

7. PRIMARY KEY (`id`) USING BTREE

8. ) ENGINE=InnoDB DEFAULT CHARSET=latin1 ROW_FORMAT=DYNAMIC COMMENT='仓储服务';

9. -- ----------------------------

10. -- Records of product

11. -- ----------------------------

12. BEGIN;

13. INSERT INTO `product` VALUES (1, 1, 50, 100);

14. COMMIT;

15. SET FOREIGN_KEY_CHECKS = 1;

account 账户表:

1. DROP TABLE IF EXISTS `account`;

2. CREATE TABLE `account` (

3. `id` int(20) NOT NULL AUTO_INCREMENT COMMENT '主键Id',

4. `user_id` int(20) DEFAULT NULL COMMENT '用户Id',

5. `balance` decimal(11,0) DEFAULT NULL COMMENT '余额',

6. PRIMARY KEY (`id`) USING BTREE

7. ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=latin1 ROW_FORMAT=DYNAMIC;

8. -- ----------------------------

9. -- Records of account

10. -- ----------------------------

11. BEGIN;

12. INSERT INTO `account` VALUES (1, 1, 100);

13. COMMIT;

14. SET FOREIGN_KEY_CHECKS = 1;

创建日志回滚表

需要在每个数据库中创建日志回滚表，建表 sql 在 seata-server 的/conf/dbundolog.sql 中。

分布式事务问题产生

三个服务，一个订单服务，一个仓储服务，一个账户服务。当用户下单时，会在订单服务中创建一个订单，然后通过远程调用库存服务来扣减下单商品的库存，再通过远程调用账户服务来扣减用户账户里面的余额，最后在订单服务中修改订单状态为已完成。该操作跨越三个数据库，有两次远程调用，很明显会有分布式事务问题

工程结构

nacos-seata-account-server 账户服务
nacos-seata-order-server 订单服务
nacos-seata-storage-server 仓储服务

客户端配置

对 nacos-seata-account-server、nacos-seata-order-server 和 nacos-seata-storage-server 三个 seata 的客户端进行配置，它们配置大致相同，我们下面以 nacos-seata-account-server 的配置为例；

1、修改 application.yml 文件，自定义事务组的名称

1. spring:

2. cloud:

3. alibaba:

4. seata:

5. tx-service-group: prex_tx_group #自定义事务组名称需要与seata-server中的对应

2、添加并修改 file.conf 配置文件，主要是修改自定义事务组名称

1. service {

2. #vgroup->rgroup

3. vgroup_mapping.prex_tx_group = "default" #修改自定义事务组名称

4. #only support single node

5. default.grouplist = "127.0.0.1:8091"

6. #degrade current not support

7. enableDegrade = false

8. #disable

9. disable = false

10. #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent

11. max.commit.retry.timeout = "-1"

12. max.rollback.retry.timeout = "-1"

13. disableGlobalTransaction = false

14. }

添加并修改 registry.conf 配置文件，主要是将注册中心改为 nacos

1. registry {

2. # file 、nacos 、eureka、redis、zk

3. type = "nacos" #修改为nacos

4. nacos {

5. serverAddr = "localhost:8848" #修改为nacos的连接地址

6. namespace = ""

7. cluster = "default"

8. }

9. }

代码只展示核心代码 具体代码文章尾部链接

3、在启动类中取消数据源的自动创建

1. @EnableDiscoveryClient

2. @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)

3. @MapperScan("com.xd.example.seata.mapper")

4. public class NacosSeataAccountServerApplication {

5. public static void main(String[] args) {

6. SpringApplication.run(NacosSeataAccountServerApplication.class, args);

7. }

8. }

4、配置 MybatisPlus 使用 Seata 对数据源进行代理

MyBatisPlusConfig:

1. /**

2. * @Classname MyBatisPlusConfig

3. * @Description 配置MybatisPlus使用Seata对数据源进行代理

4. * @Author Created by Lihaodong (alias:小东啊) [e-mail]

5. * @Date 2019-11-25 11:21

6. * @Version 1.0

7. */

8. @Configuration

9. public class MyBatisPlusConfig {

10. @Value("${mybatis-plus.mapper-locations}")

11. private String mapperLocations;

12. /**

13. * @param sqlSessionFactory SqlSessionFactory

14. * @return SqlSessionTemplate

15. */

16. @Bean

17. public SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {

18. return new SqlSessionTemplate(sqlSessionFactory);

19. }

20. /**

21. * 从配置文件获取属性构造datasource，注意前缀，这里用的是druid，根据自己情况配置,

22. * 原生datasource前缀取"spring.datasource"

23. *

24. * @return

25. */

26. @Bean

27. @ConfigurationProperties(prefix = "spring.datasource.hikari")

28. public DataSource hikariDataSource() {

29. return new HikariDataSource();

30. }

31. /**

32. * 构造datasource代理对象，替换原来的datasource

33. *

34. * @param hikariDataSource

35. * @return

36. */

37. @Primary

38. @Bean("dataSource")

39. public DataSourceProxy dataSourceProxy(DataSource hikariDataSource) {

40. return new DataSourceProxy(hikariDataSource);

41. }

42. @Bean(name = "sqlSessionFactory")

43. public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception {

44. MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();

45. bean.setDataSource(dataSourceProxy);

46. ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();

47. bean.setMapperLocations(resolver.getResources(mapperLocations));

48. SqlSessionFactory factory = null;

49. try {

50. factory = bean.getObject();

51. } catch (Exception e) {

52. throw new RuntimeException(e);

53. }

54. return factory;

55. }

56. /**

57. * MP 自带分页插件

58. *

59. * @return

60. */

61. @Bean

62. public PaginationInterceptor paginationInterceptor() {

63. PaginationInterceptor page = new PaginationInterceptor();

64. page.setDialectType("mysql");

65. return page;

66. }

67. }

5、使用@GlobalTransactional 注解开启分布式事务

1. package com.xd.example.seata.service.impl;

2. import com.baomidou.mybatisplus.core.toolkit.Wrappers;

3. import com.xd.example.seata.domain.Order;

4. import com.xd.example.seata.mapper.OrderMapper;

5. import com.xd.example.seata.service.IOrderService;

6. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

7. import com.xd.example.seata.service.RemoteAccountService;

8. import com.xd.example.seata.service.RemoteStorageService;

9. import io.seata.core.context.RootContext;

10. import io.seata.spring.annotation.GlobalTransactional;

11. import lombok.extern.slf4j.Slf4j;

12. import org.springframework.beans.factory.annotation.Autowired;

13. import org.springframework.stereotype.Service;

14. /**

15. * <p>

16. * 订单表 服务实现类

17. * </p>

18. *

19. * @author lihaodong

20. * @since 2019-11-25

21. */

22. @Slf4j

23. @Service

24. public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements IOrderService {

25. @Autowired

26. private RemoteStorageService remoteStorageService;

27. @Autowired

28. private RemoteAccountService remoteAccountService;

29. @GlobalTransactional(rollbackFor = Exception.class)

30. @Override

31. public void createOrder(Order order) {

32. log.info("下单开始,用户:{},商品:{},数量:{},金额:{}", order.getUserId(), order.getProductId(), order.getCount(), order.getPayMoney());

33. //创建订单

34. order.setStatus(0);

35. boolean save = save(order);

36. log.info("保存订单{}", save ? "成功" : "失败");

37. log.info("当前 XID: {}", RootContext.getXID());

38. //远程调用库存服务扣减库存

39. log.info("扣减库存开始");

40. remoteStorageService.decrease(order.getProductId(), order.getCount());

41. log.info("扣减库存结束");

42. //远程调用账户服务扣减余额

43. log.info("扣减余额开始");

44. remoteAccountService.decrease(order.getUserId(), order.getPayMoney());

45. log.info("扣减余额结束");

46. //修改订单状态为已完成

47. log.info("修改订单状态开始");

48. update(Wrappers.<Order>lambdaUpdate().set(Order::getStatus, 1).eq(Order::getUserId, order.getUserId()));

49. log.info("修改订单状态结束");

50. log.info("下单结束");

51. }

52. }

七. 启动服务功能演示

1、分别运行 nacos-seata-order-server、nacos-seata-storage-server 和 nacos-seata-account-server 三个服务

可以看到 seata 注册成功

2、查询数据库初始数据信息

3、打开浏览器/Postman 调用接口进行下单操作:

http://localhost:8081/order/create?userId=1&productId=1&count=1&payMoney=50

结果:

查看控制台打印: 订单服务:

仓储服务:

账户服务:

4、再次数据库查询

5、我们在 nacos-seata-account-server 中制造一个超时异常后(其他异常也行)，调用下单接口

1. package com.xd.example.seata.service.impl;

2. import com.baomidou.mybatisplus.core.toolkit.Wrappers;

3. import com.xd.example.seata.domain.Account;

4. import com.xd.example.seata.mapper.AccountMapper;

5. import com.xd.example.seata.service.IAccountService;

6. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

7. import io.seata.core.context.RootContext;

8. import lombok.extern.slf4j.Slf4j;

9. import org.springframework.stereotype.Service;

10. import java.math.BigDecimal;

11. import java.util.Optional;

12. /**

13. * <p>

14. * 服务实现类

15. * </p>

16. *

17. * @author lihaodong

18. * @since 2019-11-25

19. */

20. @Slf4j

21. @Service

22. public class AccountServiceImpl extends ServiceImpl<AccountMapper, Account> implements IAccountService {

23. @Override

24. public boolean reduceBalance(Integer userId, BigDecimal balance) throws Exception {

25. log.info("当前 XID: {}", RootContext.getXID());

26. checkBalance(userId, balance);

27. log.info("开始扣减用户 {} 余额", userId);

28. //模拟超时异常

29. try {

30. Thread.sleep(10 * 1000);

31. } catch (InterruptedException e) {

32. e.printStackTrace();

33. }

34. Integer record = baseMapper.reduceBalance(userId, balance);

35. log.info("结束扣减用户 {} 余额结果:{}", userId, record > 0 ? "操作成功" : "扣减余额失败");

36. return record > 0;

37. }

38. private void checkBalance(Integer userId, BigDecimal price) throws Exception {

39. log.info("检查用户 {} 余额", userId);

40. Optional<Account> account = Optional.ofNullable(baseMapper.selectOne(Wrappers.<Account>lambdaQuery().eq(Account::getUserId, userId)));

41. if (account.isPresent()) {

42. BigDecimal balance = account.get().getBalance();

43. if (balance.compareTo(price) == -1) {

44. log.warn("用户 {} 余额不足，当前余额:{}", userId, balance);

45. throw new Exception("余额不足");

46. }

47. }

48. }

49. }

修改完会重启账户服务,再次发送请求

订单服务控制台:

可以看到订单正常,扣除库存正常,账户服务读取超时异常

6、发现下单后数据库数据并没有任何改变

7、我们在 seata-order-service 中注释掉@GlobalTransactional 来看看会发生什么

1. // @GlobalTransactional(name = "prex-create-order",rollbackFor = Exception.class)

2. @Override

3. public void createOrder(Order order) {

4. log.info("当前 XID: {}", RootContext.getXID());

5. log.info("下单开始,用户:{},商品:{},数量:{},金额:{}", order.getUserId(), order.getProductId(), order.getCount(), order.getPayMoney());

6. //创建订单

7. order.setStatus(0);

8. boolean save = save(order);

9. log.info("保存订单{}", save ? "成功" : "失败");

10. ... 省略代码

11. }

保存重启订单服务,再次请求接口

由于 nacos-seata-account-server 的超时会导致当库存和账户金额扣减后订单状态并没有设置为已经完成

八. Seata 事务分组

下一篇更新

Seata 分布式事务原理解释

下一篇更新

项目源码地址

https://gitee.com/li_haodong/SpringCloudAlibabaLearn

参考资料: http://seata.io/zh-cn

近期技术热文