分布式事务问题由来

分布式前

  • 单机单库没这个问题
  • 从1:1 -> 1:N -> N:N

单体应用被拆分成微服务应用，原来的三个模块被拆分成三个独立的应用,分别使用三个独立的数据源，业务操作需要调用三三 个服务来完成。此时每个服务内部的数据一致性由本地事务来保证， 但是全局的数据一致性问题没法保证。
image.png
一句话：一次业务操作需要跨多个数据源或需要跨多个系统进行远程调用，就会产生分布式事务问题。

Seata 介绍

是什么
Seata是一款开源的分布式事务解决方案，致力于在微服务架构下提供高性能和简单易用的分布式事务服务。
官方网址
能干嘛
一个典型的分布式事务过程
分布式事务处理过程的一ID+三组件模型：

  • Transaction ID XID 全局唯一的事务ID
  • 三组件概念
    • TC (Transaction Coordinator) - 事务协调者：维护全局和分支事务的状态，驱动全局事务提交或回滚。
    • TM (Transaction Manager) - 事务管理器：定义全局事务的范围：开始全局事务、提交或回滚全局事务。
    • RM (Resource Manager) - 资源管理器：管理分支事务处理的资源，与 TC 交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

处理过程：

  1. TM 向 TC 申请开启一个全局事务，全局事务创建成功并生成一个全局唯一的 XID；
  2. XID 在微服务调用链路的上下文中传播；
  3. RM 向 TC 注册分支事务，将其纳入 XID 对应全局事务的管辖；
  4. TM 向 TC 发起针对 XID 的全局提交或回滚决议；
  5. TC 调度 XID 下管辖的全部分支事务完成提交或回滚请求。

image.png

原理简介

2019 年 1 月份蚂蚁金服和阿里巴巴共同开源的分布式事务解决方案。
Simple Extensible Autonomous Transaction Architecture，简单可扩展自治事务框架。
2020起始，用 1.0 以后的版本。Alina Gingertail
image.png
分布式事务的执行流程

  • TM 开启分布式事务（TM 向 TC 注册全局事务记录） ;
  • 按业务场景，编排数据库、服务等事务内资源（RM 向 TC 汇报资源准备状态） ;
  • TM 结束分布式事务，事务一阶段结束（TM 通知 TC 提交/回滚分布式事务）;
  • TC 汇总事务信息，决定分布式事务是提交还是回滚；
  • TC 通知所有 RM 提交/回滚资源，事务二阶段结束。

AT模式如何做到对业务的无侵入

前提

  • 基于支持本地 ACID 事务的关系型数据库。
  • Java 应用，通过 JDBC 访问数据库。

整体机制 两阶段提交协议的演变：

  • 一阶段：业务数据和回滚日志记录在同一个本地事务中提交，释放本地锁和连接资源。
  • 二阶段：
    • 提交异步化，非常快速地完成。
    • 回滚通过一阶段的回滚日志进行反向补偿。

link

一阶段加载
在一阶段，Seata 会拦截“业务SQL”

  1. 解析SQL语义，找到“业务SQL” 要更新的业务数据，在业务数据被更新前，将其保存成”before image”
  2. 执行“业务SQL” 更新业务数据
  3. 在业务数据更新之后，其保存成”after image”，最后生成行锁。

以上操作全部在一个数据库事务内完成, 这样保证了一阶段操作的原子性。
image.png
二阶段提交
二阶段如果顺利提交的话，因为”业务SQL”在一阶段已经提交至数据库，所以 Seata 框架只需将一阶段保存的快照数据和行锁删掉，完成数据清理即可。
image.png
二阶段回滚
二阶段如果是回滚的话，Seata 就需要回滚一阶段已经执行的 “业务SQL”，还原业务数据。
回滚方式便是用”before image”还原业务数据；但在还原前要首先要校验脏写，对比“数据库当前业务数据”和”after image”。
如果两份数据完全一致就说明没有脏写， 可以还原业务数据，如果不一致就说明有脏写, 出现脏写就需要转人工处理。
image.png
补充
image.png

Seata-Server 安装

去哪下
发布说明: https://github.com/seata/seata/releases
怎么用
本地 @Transactional
全局 @GlobalTransactional
SEATA 的分布式交易解决方案
image.png
我们只需要使用一个 @GlobalTransactional 注解在业务方法上

安装
官网地址 - http://seata.io/zh-cn/
下载版本 - 0.9.0
seata-server-0.9.0.zip 解压到指定目录并修改 conf 目录下的 file.conf 配置文件
先备份原始 file.conf 文件
主要修改：自定义事务组名称 + 事务日志存储模式为db + 数据库连接信息
file.conf
service 模块

  1. service {
  2.     ##fsp_tx_group是自定义的
  3.     vgroup_mapping.my.test.tx_group="fsp_tx_group" 
  4.     default.grouplist = "127.0.0.1:8091"
  5.     enableDegrade = false
  6.     disable = false
  7.     max.commitretry.timeout= "-1"
  8.     max.ollbackretry.timeout= "-1"
  9. }

store 模块

  1. ## transaction log store
  2. store {
  3.     ## store mode: file, db
  4.     ## 改成db
  5.     mode = "db"
  7.     ## file store
  8.     file {
  9.         dir = "sessionStore"
  11.         # branch session size, if exceeded first try compress lockkey, still exceeded throws exceptions
  12.         max-branch-session-size = 16384
  13.         # globe session size, if exceeded throws exceptions
  14.         max-global-session-size = 512
  15.         # file buffer size, if exceeded allocate new buffer
  16.         file-write-buffer-cache-size = 16384
  17.         # when recover batch read size
  18.         session.reload.read_size= 100
  19.         # async, sync
  20.         flush-disk-mode = async
  21.     }
  22.     # database store
  23.     db {
  24.         ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
  25.         datasource = "dbcp"
  26.         ## mysql/oracle/h2/oceanbase etc.
  27.         ## 配置数据源
  28.         db-type = "mysql"
  29.         driver-class-name = "com.mysql.jdbc.Driver"
  30.         url = "jdbc:mysql://127.0.0.1:3306/seata"
  31.         user = "root"
  32.         password = "你自己密码"
  33.         min-conn= 1
  34.         max-conn = 3
  35.         global.table = "global_table"
  36.         branch.table = "branch_table"
  37.         lock-table = "lock_table"
  38.         query-limit = 100
  39.     }
  40. }

mysql5.7 数据库新建库 seata，在 seata 库里建表
建表 db_store.sql 在 \seata-server-0.9.0\seata\conf 目录里面

  1. -- the table to store GlobalSession data
  2. drop table if exists `global_table`;
  3. create table `global_table` (
  4.   `xid` varchar(128)  not null,
  5.   `transaction_id` bigint,
  6.   `status` tinyint not null,
  7.   `application_id` varchar(32),
  8.   `transaction_service_group` varchar(32),
  9.   `transaction_name` varchar(128),
  10.   `timeout` int,
  11.   `begin_time` bigint,
  12.   `application_data` varchar(2000),
  13.   `gmt_create` datetime,
  14.   `gmt_modified` datetime,
  15.   primary key (`xid`),
  16.   key `idx_gmt_modified_status` (`gmt_modified`, `status`),
  17.   key `idx_transaction_id` (`transaction_id`)
  18. );
  19. -- the table to store BranchSession data
  20. drop table if exists `branch_table`;
  21. create table `branch_table` (
  22.   `branch_id` bigint not null,
  23.   `xid` varchar(128) not null,
  24.   `transaction_id` bigint ,
  25.   `resource_group_id` varchar(32),
  26.   `resource_id` varchar(256) ,
  27.   `lock_key` varchar(128) ,
  28.   `branch_type` varchar(8) ,
  29.   `status` tinyint,
  30.   `client_id` varchar(64),
  31.   `application_data` varchar(2000),
  32.   `gmt_create` datetime,
  33.   `gmt_modified` datetime,
  34.   primary key (`branch_id`),
  35.   key `idx_xid` (`xid`)
  36. );
  37. -- the table to store lock data
  38. drop table if exists `lock_table`;
  39. create table `lock_table` (
  40.   `row_key` varchar(128) not null,
  41.   `xid` varchar(96),
  42.   `transaction_id` long ,
  43.   `branch_id` long,
  44.   `resource_id` varchar(256) ,
  45.   `table_name` varchar(32) ,
  46.   `pk` varchar(36) ,
  47.   `gmt_create` datetime ,
  48.   `gmt_modified` datetime,
  49.   primary key(`row_key`)
  50. );

修改 seata-server-0.9.0\seata\conf 目录下的 registry.conf 配置文件

  1. registry {
  2.   # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  3.   # 改用为nacos
  4.   type = "nacos"
  5.   nacos {
  6.       ## 加端口号
  7.     serverAddr = "localhost:8848"
  8.     namespace = ""
  9.     cluster = "default"
  10.   }
  11.   ...
  12. }

目的是：指明注册中心为 nacos，及修改 nacos 连接信息
先启动 Nacos 端口号 8848 nacos\bin\startup.cmd
再启动 seata-server - seata-server-0.9.0\seata\bin\seata-server.bat

示例

业务数据库准备

以下演示都需要先启动 Nacos 后启动 Seata，保证两个都 OK
分布式事务业务说明
这里我们会创建三个服务，一个订单服务，一个库存服务，一个账户服务。
当用户下单时，会在订单服务中创建一个订单，然后通过远程调用库存服务来扣减下单商品的库存，再通过远程调用账户服务来扣减用户账户里面的余额，最后在订单服务中修改订单状态为已完成。
该操作跨越三个数据库，有两次远程调用，很明显会有分布式事务问题。
一言蔽之，下订单—>扣库存—>减账户(余额)。
创建业务数据库

  • seata_ order：存储订单的数据库;
  • seata_ storage：存储库存的数据库;
  • seata_ account：存储账户信息的数据库。

建库 SQL

  1. CREATE DATABASE seata_order;
  2. CREATE DATABASE seata_storage;
  3. CREATE DATABASE seata_account;

按照上述3库分别建对应业务表

  • seata_order库下建t_order表

    1. CREATE TABLE t_order (
    2. `id` BIGINT(11) NOT NULL AUTO_INCREMENT PRIMARY KEY,
    3. `user_id` BIGINT(11) DEFAULT NULL COMMENT '用户id',
    4. `product_id` BIGINT(11) DEFAULT NULL COMMENT '产品id',
    5. `count` INT(11) DEFAULT NULL COMMENT '数量',
    6. `money` DECIMAL(11,0) DEFAULT NULL COMMENT '金额',
    7. `status` INT(1) DEFAULT NULL COMMENT '订单状态: 0:创建中; 1:已完结'
    8. ) ENGINE=INNODB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
    9. SELECT * FROM t_order;

  • seata_storage 库下建 t_storage 表

    1. CREATE TABLE t_storage (
    2. `id` BIGINT(11) NOT NULL AUTO_INCREMENT PRIMARY KEY,
    3. `product_id` BIGINT(11) DEFAULT NULL COMMENT '产品id',
    4. `total` INT(11) DEFAULT NULL COMMENT '总库存',
    5. `used` INT(11) DEFAULT NULL COMMENT '已用库存',
    6. `residue` INT(11) DEFAULT NULL COMMENT '剩余库存'
    7. ) ENGINE=INNODB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
    8. INSERT INTO seata_storage.t_storage(`id`, `product_id`, `total`, `used`, `residue`)
    9. VALUES ('1', '1', '100', '0','100');
    10. SELECT * FROM t_storage;

  • seata_account 库下建 t_account 表

    1. CREATE TABLE t_account(
    2. `id` BIGINT(11) NOT NULL AUTO_INCREMENT PRIMARY KEY COMMENT 'id',
    3. `user_id` BIGINT(11) DEFAULT NULL COMMENT '用户id',
    4. `total` DECIMAL(10,0) DEFAULT NULL COMMENT '总额度',
    5. `used` DECIMAL(10,0) DEFAULT NULL COMMENT '已用余额',
    6. `residue` DECIMAL(10,0) DEFAULT '0' COMMENT '剩余可用额度'
    7. ) ENGINE=INNODB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
    8. INSERT INTO seata_account.t_account(`id`, `user_id`, `total`, `used`, `residue`)
    9. VALUES ('1', '1', '1000', '0', '1000');
    10. SELECT * FROM t_account;

    按照上述3库分别建对应的回滚日志表

  • 订单-库存-账户3个库下都需要建各自的回滚日志表

  • \seata-server-0.9.0\seata\conf 目录下的 db_ undo_ log.sql

  • 建表 SQL

    1. -- the table to store seata xid data
    2. -- 0.7.0+ add context
    3. -- you must to init this sql for you business databese. the seata server not need it.
    4. -- 此脚本必须初始化在你当前的业务数据库中，用于AT 模式XID记录。与server端无关（注：业务数据库）
    5. -- 注意此处0.3.0+ 增加唯一索引 ux_undo_log
    6. drop table `undo_log`;
    7. CREATE TABLE `undo_log` (
    8. `id` bigint(20) NOT NULL AUTO_INCREMENT,
    9. `branch_id` bigint(20) NOT NULL,
    10. `xid` varchar(100) NOT NULL,
    11. `context` varchar(128) NOT NULL,
    12. `rollback_info` longblob NOT NULL,
    13. `log_status` int(11) NOT NULL,
    14. `log_created` datetime NOT NULL,
    15. `log_modified` datetime NOT NULL,
    16. `ext` varchar(100) DEFAULT NULL,
    17. PRIMARY KEY (`id`),
    18. UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
    19. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

    Order-Module 搭建

    下订单 -> 减库存 -> 扣余额 -> 改（订单）状态
    seata-order-service2001
    POM

    1. <?xml version="1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.   <parent>
    6.       <artifactId>cloud2020</artifactId>
    7.       <groupId>com.atguigu.springcloud</groupId>
    8.       <version>1.0-SNAPSHOT</version>
    9.   </parent>
    10.   <modelVersion>4.0.0</modelVersion>
    11.   <artifactId>seata-order-service2001</artifactId>
    12.   <dependencies>
    13.       <!--nacos-->
    14.       <dependency>
    15.           <groupId>com.alibaba.cloud</groupId>
    16.           <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
    17.       </dependency>
    18.       <!--seata-->
    19.       <dependency>
    20.           <groupId>com.alibaba.cloud</groupId>
    21.           <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
    22.           <exclusions>
    23.               <exclusion>
    24.                   <artifactId>seata-all</artifactId>
    25.                   <groupId>io.seata</groupId>
    26.               </exclusion>
    27.           </exclusions>
    28.       </dependency>
    29.       <dependency>
    30.           <groupId>io.seata</groupId>
    31.           <artifactId>seata-all</artifactId>
    32.           <version>0.9.0</version>
    33.       </dependency>
    34.       <!--feign-->
    35.       <dependency>
    36.           <groupId>org.springframework.cloud</groupId>
    37.           <artifactId>spring-cloud-starter-openfeign</artifactId>
    38.       </dependency>
    39.       <!--web-actuator-->
    40.       <dependency>
    41.           <groupId>org.springframework.boot</groupId>
    42.           <artifactId>spring-boot-starter-web</artifactId>
    43.       </dependency>
    44.       <dependency>
    45.           <groupId>org.springframework.boot</groupId>
    46.           <artifactId>spring-boot-starter-actuator</artifactId>
    47.       </dependency>
    48.       <!--mysql-druid-->
    49.       <dependency>
    50.           <groupId>mysql</groupId>
    51.           <artifactId>mysql-connector-java</artifactId>
    52.           <version>5.1.37</version>
    53.       </dependency>
    54.       <dependency>
    55.           <groupId>com.alibaba</groupId>
    56.           <artifactId>druid-spring-boot-starter</artifactId>
    57.           <version>1.1.10</version>
    58.       </dependency>
    59.       <dependency>
    60.           <groupId>org.mybatis.spring.boot</groupId>
    61.           <artifactId>mybatis-spring-boot-starter</artifactId>
    62.           <version>2.0.0</version>
    63.       </dependency>
    64.       <dependency>
    65.           <groupId>org.springframework.boot</groupId>
    66.           <artifactId>spring-boot-starter-test</artifactId>
    67.           <scope>test</scope>
    68.       </dependency>
    69.       <dependency>
    70.           <groupId>org.projectlombok</groupId>
    71.           <artifactId>lombok</artifactId>
    72.           <optional>true</optional>
    73.       </dependency>
    74.   </dependencies>
    75. </project>

    配置文件
    YML

    1. server:
    2. port: 2001
    3. spring:
    4. application:
    5.   name: seata-order-service
    6. cloud:
    7.   alibaba:
    8.     seata:
    9.       #自定义事务组名称需要与seata-server中的对应
    10.       tx-service-group: fsp_tx_group
    11.   nacos:
    12.     discovery:
    13.       server-addr: localhost:8848
    14. datasource:
    15.   driver-class-name: com.mysql.jdbc.Driver
    16.   url: jdbc:mysql://localhost:3306/seata_order
    17.   username: root
    18.   password: 123456
    19. feign:
    20. hystrix:
    21.   enabled: false
    22. logging:
    23. level:
    24.   io:
    25.     seata: info
    26. mybatis:
    27. mapperLocations: classpath:mapper/*.xml

    file.conf

    1. transport {
    2. # tcp udt unix-domain-socket
    3. type = "TCP"
    4. #NIO NATIVE
    5. server = "NIO"
    6. #enable heartbeat
    7. heartbeat = true
    8. #thread factory for netty
    9. thread-factory {
    10.   boss-thread-prefix = "NettyBoss"
    11.   worker-thread-prefix = "NettyServerNIOWorker"
    12.   server-executor-thread-prefix = "NettyServerBizHandler"
    13.   share-boss-worker = false
    14.   client-selector-thread-prefix = "NettyClientSelector"
    15.   client-selector-thread-size = 1
    16.   client-worker-thread-prefix = "NettyClientWorkerThread"
    17.   # netty boss thread size,will not be used for UDT
    18.   boss-thread-size = 1
    19.   #auto default pin or 8
    20.   worker-thread-size = 8
    21. }
    22. shutdown {
    23.   # when destroy server, wait seconds
    24.   wait = 3
    25. }
    26. serialization = "seata"
    27. compressor = "none"
    28. }
    29. service {
    30. vgroup_mapping.fsp_tx_group = "default" #修改自定义事务组名称
    31. default.grouplist = "127.0.0.1:8091"
    32. enableDegrade = false
    33. disable = false
    34. max.commit.retry.timeout = "-1"
    35. max.rollback.retry.timeout = "-1"
    36. disableGlobalTransaction = false
    37. }
    38. client {
    39. async.commit.buffer.limit = 10000
    40. lock {
    41.   retry.internal = 10
    42.   retry.times = 30
    43. }
    44. report.retry.count = 5
    45. tm.commit.retry.count = 1
    46. tm.rollback.retry.count = 1
    47. }
    48. ## transaction log store
    49. store {
    50. ## store mode: file、db
    51. mode = "db"
    52. ## file store
    53. file {
    54.   dir = "sessionStore"
    55.   # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
    56.   max-branch-session-size = 16384
    57.   # globe session size , if exceeded throws exceptions
    58.   max-global-session-size = 512
    59.   # file buffer size , if exceeded allocate new buffer
    60.   file-write-buffer-cache-size = 16384
    61.   # when recover batch read size
    62.   session.reload.read_size = 100
    63.   # async, sync
    64.   flush-disk-mode = async
    65. }
    66. ## database store
    67. db {
    68.   ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    69.   datasource = "dbcp"
    70.   ## mysql/oracle/h2/oceanbase etc.
    71.   db-type = "mysql"
    72.   driver-class-name = "com.mysql.jdbc.Driver"
    73.   url = "jdbc:mysql://127.0.0.1:3306/seata"
    74.   user = "root"
    75.   password = "123456"
    76.   min-conn = 1
    77.   max-conn = 3
    78.   global.table = "global_table"
    79.   branch.table = "branch_table"
    80.   lock-table = "lock_table"
    81.   query-limit = 100
    82. }
    83. }
    84. lock {
    85. ## the lock store mode: local、remote
    86. mode = "remote"
    87. local {
    88.   ## store locks in user's database
    89. }
    90. remote {
    91.   ## store locks in the seata's server
    92. }
    93. }
    94. recovery {
    95. #schedule committing retry period in milliseconds
    96. committing-retry-period = 1000
    97. #schedule asyn committing retry period in milliseconds
    98. asyn-committing-retry-period = 1000
    99. #schedule rollbacking retry period in milliseconds
    100. rollbacking-retry-period = 1000
    101. #schedule timeout retry period in milliseconds
    102. timeout-retry-period = 1000
    103. }
    104. transaction {
    105. undo.data.validation = true
    106. undo.log.serialization = "jackson"
    107. undo.log.save.days = 7
    108. #schedule delete expired undo_log in milliseconds
    109. undo.log.delete.period = 86400000
    110. undo.log.table = "undo_log"
    111. }
    112. ## metrics settings
    113. metrics {
    114. enabled = false
    115. registry-type = "compact"
    116. # multi exporters use comma divided
    117. exporter-list = "prometheus"
    118. exporter-prometheus-port = 9898
    119. }
    120. support {
    121. ## spring
    122. spring {
    123.   # auto proxy the DataSource bean
    124.   datasource.autoproxy = false
    125. }
    126. }

    registry.conf

    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. eureka {
    10.   serviceUrl = "http://localhost:8761/eureka"
    11.   application = "default"
    12.   weight = "1"
    13. }
    14. redis {
    15.   serverAddr = "localhost:6379"
    16.   db = "0"
    17. }
    18. zk {
    19.   cluster = "default"
    20.   serverAddr = "127.0.0.1:2181"
    21.   session.timeout = 6000
    22.   connect.timeout = 2000
    23. }
    24. consul {
    25.   cluster = "default"
    26.   serverAddr = "127.0.0.1:8500"
    27. }
    28. etcd3 {
    29.   cluster = "default"
    30.   serverAddr = "http://localhost:2379"
    31. }
    32. sofa {
    33.   serverAddr = "127.0.0.1:9603"
    34.   application = "default"
    35.   region = "DEFAULT_ZONE"
    36.   datacenter = "DefaultDataCenter"
    37.   cluster = "default"
    38.   group = "SEATA_GROUP"
    39.   addressWaitTime = "3000"
    40. }
    41. file {
    42.   name = "file.conf"
    43. }
    44. }
    45. config {
    46. # file、nacos 、apollo、zk、consul、etcd3
    47. type = "file"
    48. nacos {
    49.   serverAddr = "localhost"
    50.   namespace = ""
    51. }
    52. consul {
    53.   serverAddr = "127.0.0.1:8500"
    54. }
    55. apollo {
    56.   app.id = "seata-server"
    57.   apollo.meta = "http://192.168.1.204:8801"
    58. }
    59. zk {
    60.   serverAddr = "127.0.0.1:2181"
    61.   session.timeout = 6000
    62.   connect.timeout = 2000
    63. }
    64. etcd3 {
    65.   serverAddr = "http://localhost:2379"
    66. }
    67. file {
    68.   name = "file.conf"
    69. }
    70. }

    domain

    1. import lombok.AllArgsConstructor;
    2. import lombok.Data;
    3. import lombok.NoArgsConstructor;
    4. @Data
    5. @AllArgsConstructor
    6. @NoArgsConstructor
    7. public class CommonResult<T> {
    9.   private Integer code;
    10.   private String  message;
    11.   private T       data;
    13.   public CommonResult(Integer code, String message) {
    14.       this(code,message,null);
    15.   }
    16. }

    ```java package com.atguigu.springcloud.alibaba.domain; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import java.math.BigDecimal;

@Data @AllArgsConstructor @NoArgsConstructor public class Order { private Long id; private Long userId; private Long productId; private Integer count; private BigDecimal money; private Integer status; //订单状态：0：创建中；1：已完结 }

  1. Dao 接口及实现
  2. ```java
  3. import com.atguigu.springcloud.alibaba.domain.Order;
  4. import org.apache.ibatis.annotations.Mapper;
  5. import org.apache.ibatis.annotations.Param;
  7. @Mapper
  8. public interface OrderDao {
  9.     //1 新建订单
  10.     void create(Order order);
  11.     //2 修改订单状态，从零改为1
  12.     void update(@Param("userId") Long userId,@Param("status") Integer status);
  13. }
  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
  3. <mapper namespace="com.atguigu.springcloud.alibaba.dao.OrderDao">
  4.     <resultMap id="BaseResultMap" type="com.atguigu.springcloud.alibaba.domain.Order">
  5.         <id column="id" property="id" jdbcType="BIGINT"/>
  6.         <result column="user_id" property="userId" jdbcType="BIGINT"/>
  7.         <result column="product_id" property="productId" jdbcType="BIGINT"/>
  8.         <result column="count" property="count" jdbcType="INTEGER"/>
  9.         <result column="money" property="money" jdbcType="DECIMAL"/>
  10.         <result column="status" property="status" jdbcType="INTEGER"/>
  11.     </resultMap>
  12.     <insert id="create">
  13.         insert into t_order (id,user_id,product_id,count,money,status)
  14.         values (null,#{userId},#{productId},#{count},#{money},0);
  15.     </insert>
  16.     <update id="update">
  17.         update t_order set status = 1
  18.         where user_id=#{userId} and status = #{status};
  19.     </update>
  20. </mapper>

Service接口及实现

  • OrderService
    • OrderServiceImpl
  • StorageService
  • AccountService ```java import com.atguigu.springcloud.alibaba.domain.Order;

public interface OrderService { void create(Order order); }

  1. ```java
  2. import com.atguigu.springcloud.alibaba.domain.CommonResult;
  3. import org.springframework.cloud.openfeign.FeignClient;
  4. import org.springframework.web.bind.annotation.PostMapping;
  5. import org.springframework.web.bind.annotation.RequestParam;
  6. import java.math.BigDecimal;
  8. @FeignClient(value = "seata-storage-service")
  9. public interface StorageService {
  11.     @PostMapping(value = "/storage/decrease")
  12.     CommonResult decrease(@RequestParam("productId") Long productId, @RequestParam("count") Integer count);
  13. }
  1. import com.atguigu.springcloud.alibaba.domain.CommonResult;
  2. import org.springframework.cloud.openfeign.FeignClient;
  3. import org.springframework.web.bind.annotation.PostMapping;
  4. import org.springframework.web.bind.annotation.RequestParam;
  5. import java.math.BigDecimal;
  7. @FeignClient(value = "seata-account-service")
  8. public interface AccountService {
  10.     @PostMapping(value = "/account/decrease")
  11.     CommonResult decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money);
  12. }
  1. import com.atguigu.springcloud.alibaba.dao.OrderDao;
  2. import com.atguigu.springcloud.alibaba.domain.Order;
  3. import com.atguigu.springcloud.alibaba.service.AccountService;
  4. import com.atguigu.springcloud.alibaba.service.OrderService;
  5. import com.atguigu.springcloud.alibaba.service.StorageService;
  6. import io.seata.spring.annotation.GlobalTransactional;
  7. import lombok.extern.slf4j.Slf4j;
  8. import org.springframework.stereotype.Service;
  9. import javax.annotation.Resource;
  11. @Service
  12. @Slf4j
  13. public class OrderServiceImpl implements OrderService {
  15.     @Resource
  16.     private OrderDao orderDao;
  18.     @Resource
  19.     private StorageService storageService;
  21.     @Resource
  22.     private AccountService accountService;
  24.     /**
  25.      * 创建订单->调用库存服务扣减库存->调用账户服务扣减账户余额->修改订单状态
  26.      * 简单说：下订单->扣库存->减余额->改状态
  27.      */
  28.     @Override
  29.     //@GlobalTransactional(name = "fsp-create-order",rollbackFor = Exception.class)
  30.     public void create(Order order) {
  32.         log.info("----->开始新建订单");
  34.         //1 新建订单
  35.         orderDao.create(order);
  37.         //2 扣减库存
  38.         log.info("----->订单微服务开始调用库存，做扣减Count");
  39.         storageService.decrease(order.getProductId(),order.getCount());
  40.         log.info("----->订单微服务开始调用库存，做扣减end");
  42.         //3 扣减账户
  43.         log.info("----->订单微服务开始调用账户，做扣减Money");
  44.         accountService.decrease(order.getUserId(),order.getMoney());
  45.         log.info("----->订单微服务开始调用账户，做扣减end");
  47.         //4 修改订单状态，从零到1,1代表已经完成
  48.         log.info("----->修改订单状态开始");
  49.         orderDao.update(order.getUserId(),0);
  50.         log.info("----->修改订单状态结束");
  52.         log.info("----->下订单结束了，O(∩_∩)O哈哈~");
  53.     }
  54. }

Controller

  1. import com.atguigu.springcloud.alibaba.domain.CommonResult;
  2. import com.atguigu.springcloud.alibaba.domain.Order;
  3. import com.atguigu.springcloud.alibaba.service.OrderService;
  4. import org.springframework.web.bind.annotation.GetMapping;
  5. import org.springframework.web.bind.annotation.RestController;
  6. import javax.annotation.Resource;
  7. @RestController
  8. public class OrderController {
  10.     @Resource
  11.     private OrderService orderService;
  13.     @GetMapping("/order/create")
  14.     public CommonResult create(Order order) {
  15.         orderService.create(order);
  16.         return new CommonResult(200,"订单创建成功");
  17.     }
  18. }

Config配置

  • MyBatisConfig
  • DataSourceProxyConfig ```java import org.mybatis.spring.annotation.MapperScan; import org.springframework.context.annotation.Configuration;

@Configuration @MapperScan({“com.atguigu.springcloud.alibaba.dao”}) public class MyBatisConfig { }

  1. ```java
  2. import com.alibaba.druid.pool.DruidDataSource;
  3. import io.seata.rm.datasource.DataSourceProxy;
  4. import org.apache.ibatis.session.SqlSessionFactory;
  5. import org.mybatis.spring.SqlSessionFactoryBean;
  6. import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
  7. import org.springframework.beans.factory.annotation.Value;
  8. import org.springframework.boot.context.properties.ConfigurationProperties;
  9. import org.springframework.context.annotation.Bean;
  10. import org.springframework.context.annotation.Configuration;
  11. import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
  12. import javax.sql.DataSource;
  14. /**
  15. * 使用Seata对数据源进行代理
  16. */
  17. @Configuration
  18. public class DataSourceProxyConfig {
  20.     @Value("${mybatis.mapperLocations}")
  21.     private String mapperLocations;
  23.     @Bean
  24.     @ConfigurationProperties(prefix = "spring.datasource")
  25.     public DataSource druidDataSource(){
  26.         return new DruidDataSource();
  27.     }
  29.     @Bean
  30.     public DataSourceProxy dataSourceProxy(DataSource dataSource) {
  31.         return new DataSourceProxy(dataSource);
  32.     }
  34.     @Bean
  35.     public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception {
  36.         SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
  37.         sqlSessionFactoryBean.setDataSource(dataSourceProxy);
  38.         sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(mapperLocations));
  39.         sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory());
  40.         return sqlSessionFactoryBean.getObject();
  41.     }
  42. }

主启动

  1. import org.springframework.boot.SpringApplication;
  2. import org.springframework.boot.autoconfigure.SpringBootApplication;
  3. import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
  4. import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
  5. import org.springframework.cloud.openfeign.EnableFeignClients;
  7. @EnableDiscoveryClient
  8. @EnableFeignClients
  9. //取消数据源的自动创建，而是使用自己定义的
  10. @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
  11. public class SeataOrderMainApp2001 {
  13.     public static void main(String[] args) {
  14.         SpringApplication.run(SeataOrderMainApp2001.class, args);
  15.     }
  16. }

Storage-Module

与seata-order-service2001模块大致相同
seata- storage - service2002
POM（与seata-order-service2001模块大致相同）
YML

  1. server:
  2.   port: 2002
  3. spring:
  4.   application:
  5.     name: seata-storage-service
  6.   cloud:
  7.     alibaba:
  8.       seata:
  9.         tx-service-group: fsp_tx_group
  10.     nacos:
  11.       discovery:
  12.         server-addr: localhost:8848
  13.   datasource:
  14.     driver-class-name: com.mysql.jdbc.Driver
  15.     url: jdbc:mysql://localhost:3306/seata_storage
  16.     username: root
  17.     password: 123456
  18. logging:
  19.   level:
  20.     io:
  21.       seata: info
  22. mybatis:
  23.   mapperLocations: classpath:mapper/*.xml

file.conf（与seata-order-service2001模块大致相同）
registry.conf（与seata-order-service2001模块大致相同）
domain

  1. import lombok.Data;
  2. @Data
  3. public class Storage {
  4.     private Long id;
  5.     /**
  6.      * 产品id
  7.      */
  8.     private Long productId;
  9.     /**
  10.      * 总库存
  11.      */
  12.     private Integer total;
  13.     /**
  14.      * 已用库存
  15.      */
  16.     private Integer used;
  17.     /**
  18.      * 剩余库存
  19.      */
  20.     private Integer residue;
  21. }

CommonResult（与seata-order-service2001模块大致相同）
Dao 接口及实现

  1. import org.apache.ibatis.annotations.Mapper;
  2. import org.apache.ibatis.annotations.Param;
  4. @Mapper
  5. public interface StorageDao {
  6.     //扣减库存
  7.     void decrease(@Param("productId") Long productId, @Param("count") Integer count);
  8. }
  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
  3. <mapper namespace="com.atguigu.springcloud.alibaba.dao.StorageDao">
  4.     <resultMap id="BaseResultMap" type="com.atguigu.springcloud.alibaba.domain.Storage">
  5.         <id column="id" property="id" jdbcType="BIGINT"/>
  6.         <result column="product_id" property="productId" jdbcType="BIGINT"/>
  7.         <result column="total" property="total" jdbcType="INTEGER"/>
  8.         <result column="used" property="used" jdbcType="INTEGER"/>
  9.         <result column="residue" property="residue" jdbcType="INTEGER"/>
  10.     </resultMap>
  11.     <update id="decrease">
  12.         UPDATE
  13.             t_storage
  14.         SET
  15.             used = used + #{count},residue = residue - #{count}
  16.         WHERE
  17.             product_id = #{productId}
  18.     </update>
  19. </mapper>

Service接口及实现

  1. public interface StorageService {
  2.     /**
  3.     * 扣减库存
  4.     */
  5.     void decrease(Long productId, Integer count);
  6. }
  1. import com.atguigu.springcloud.alibaba.dao.StorageDao;
  2. import com.atguigu.springcloud.alibaba.service.StorageService ;
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.stereotype.Service;
  7. import javax.annotation.Resource;
  9. @Service
  10. public class StorageServiceImpl implements StorageService {
  12.     private static final Logger LOGGER = LoggerFactory.getLogger(StorageServiceImpl.class);
  14.     @Resource
  15.     private StorageDao storageDao;
  17.     /**
  18.      * 扣减库存
  19.      */
  20.     @Override
  21.     public void decrease(Long productId, Integer count) {
  22.         LOGGER.info("------->storage-service中扣减库存开始");
  23.         storageDao.decrease(productId,count);
  24.         LOGGER.info("------->storage-service中扣减库存结束");
  25.     }
  26. }

Controller

  1. import com.atguigu.springcloud.alibaba.domain.CommonResult ;
  2. import com.atguigu.springcloud.alibaba.service.StorageService ;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.web.bind.annotation.RequestMapping;
  5. import org.springframework.web.bind.annotation.RestController;
  7. @RestController
  8. public class StorageController {
  10.     @Autowired
  11.     private StorageService storageService;
  13.     /**
  14.      * 扣减库存
  15.      */
  16.     @RequestMapping("/storage/decrease")
  17.     public CommonResult decrease(Long productId, Integer count) {
  18.         storageService.decrease(productId, count);
  19.         return new CommonResult(200,"扣减库存成功！");
  20.     }
  21. }

Config 配置（与seata-order-service2001模块大致相同）
主启动（与seata-order-service2001模块大致相同）

Account-Module

与 seata-order-service2001 模块大致相同
seata- account- service2003
POM（与seata-order-service2001模块大致相同）
YML

  1. server:
  2.   port: 2003
  3. spring:
  4.   application:
  5.     name: seata-account-service
  6.   cloud:
  7.     alibaba:
  8.       seata:
  9.         tx-service-group: fsp_tx_group
  10.     nacos:
  11.       discovery:
  12.         server-addr: localhost:8848
  13.   datasource:
  14.     driver-class-name: com.mysql.jdbc.Driver
  15.     url: jdbc:mysql://localhost:3306/seata_account
  16.     username: root
  17.     password: 123456
  18. feign:
  19.   hystrix:
  20.     enabled: false
  21. logging:
  22.   level:
  23.     io:
  24.       seata: info
  25. mybatis:
  26.   mapperLocations: classpath:mapper/*.xml

file.conf（与seata-order-service2001模块大致相同）
registry.conf（与seata-order-service2001模块大致相同）
domain

  1. import lombok.AllArgsConstructor;
  2. import lombok.Data;
  3. import lombok.NoArgsConstructor;
  4. import java.math.BigDecimal;
  6. @Data
  7. @AllArgsConstructor
  8. @NoArgsConstructor
  9. public class Account {
  10.     private Long id;
  11.     /**
  12.      * 用户id
  13.      */
  14.     private Long userId;
  15.     /**
  16.      * 总额度
  17.      */
  18.     private BigDecimal total;
  19.     /**
  20.      * 已用额度
  21.      */
  22.     private BigDecimal used;
  23.     /**
  24.      * 剩余额度
  25.      */
  26.     private BigDecimal residue;
  27. }

CommonResult（与seata-order-service2001模块大致相同）
Dao 接口及实现

  1. import org.apache.ibatis.annotations.Mapper;
  2. import org.apache.ibatis.annotations.Param;
  3. import org.springframework.stereotype.Component;
  4. import org.springframework.stereotype.Repository;
  5. import java.math.BigDecimal;
  7. @Mapper
  8. public interface AccountDao {
  9.     /**
  10.     * 扣减账户余额
  11.     */
  12.     void decrease(@Param("userId") Long userId, @Param("money") BigDecimal money);
  13. }
  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
  3. <mapper namespace="com.atguigu.springcloud.alibaba.dao.AccountDao">
  4.     <resultMap id="BaseResultMap" type="com.atguigu.springcloud.alibaba.domain.Account">
  5.         <id column="id" property="id" jdbcType="BIGINT"/>
  6.         <result column="user_id" property="userId" jdbcType="BIGINT"/>
  7.         <result column="total" property="total" jdbcType="DECIMAL"/>
  8.         <result column="used" property="used" jdbcType="DECIMAL"/>
  9.         <result column="residue" property="residue" jdbcType="DECIMAL"/>
  10.     </resultMap>
  11.     <update id="decrease">
  12.         UPDATE t_account
  13.         SET
  14.           residue = residue - #{money},used = used + #{money}
  15.         WHERE
  16.           user_id = #{userId};
  17.     </update>
  18. </mapper>

Service接口及实现

  1. import org.springframework.web.bind.annotation.RequestParam;
  2. import org.springframework.web.bind.annotation.ResponseBody;
  3. import java.math.BigDecimal;
  4. public interface AccountService {
  6.     /**
  7.      * 扣减账户余额
  8.      * @param userId 用户id
  9.      * @param money 金额
  10.      */
  11.     void decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money);
  12. }
  1. import com.atguigu.springcloud.alibaba.dao.AccountDao;
  2. import com.atguigu.springcloud.alibaba.service.AccountService ;
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.stereotype.Service;
  7. import javax.annotation.Resource;
  8. import java.math.BigDecimal;
  9. import java.util.concurrent.TimeUnit;
  11. /**
  12.  */
  13. @Service
  14. public class AccountServiceImpl implements AccountService {
  16.     private static final Logger LOGGER = LoggerFactory.getLogger(AccountServiceImpl.class);
  18.     @Resource
  19.     AccountDao accountDao;
  21.     /**
  22.      * 扣减账户余额
  23.      */
  24.     @Override
  25.     public void decrease(Long userId, BigDecimal money) {
  26.         LOGGER.info("------->account-service中扣减账户余额开始");
  27.         accountDao.decrease(userId,money);
  28.         LOGGER.info("------->account-service中扣减账户余额结束");
  29.     }
  30. }

Controller

  1. import com.atguigu.springcloud.alibaba.domain.CommonResult ;
  2. import com.atguigu.springcloud.alibaba.service.AccountService ;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.web.bind.annotation.PostMapping;
  5. import org.springframework.web.bind.annotation.RequestMapping;
  6. import org.springframework.web.bind.annotation.RequestParam;
  7. import org.springframework.web.bind.annotation.RestController;
  8. import javax.annotation.Resource;
  9. import java.math.BigDecimal;
  10. @RestController
  11. public class AccountController {
  13.     @Resource
  14.     AccountService accountService;
  16.     /**
  17.     * 扣减账户余额
  18.     */
  19.     @RequestMapping("/account/decrease")
  20.     public CommonResult decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money){
  21.         accountService.decrease(userId,money);
  22.         return new CommonResult(200,"扣减账户余额成功！");
  23.     }
  24. }

Config 配置（与seata-order-service2001模块大致相同）
主启动（与seata-order-service2001模块大致相同）

@GlobalTransactional 验证

下订单 -> 减库存 -> 扣余额 -> 改（订单）状态
数据库初始情况：
image.png
正常下单 - http://localhost:2001/order/create?userId=1&productId=1&count=10&money=100
数据库正常下单后状况：
image.png

超时异常，没加 **@GlobalTransactional**
模拟 AccountServiceImpl 添加超时

  1. @Service
  2. public class AccountServiceImpl implements AccountService {
  4.     private static final Logger LOGGER = LoggerFactory.getLogger(AccountServiceImpl.class);
  6.     @Resource
  7.     AccountDao accountDao;
  9.     /**
  10.      * 扣减账户余额
  11.      */
  12.     @Override
  13.     public void decrease(Long userId, BigDecimal money) {
  14.         LOGGER.info("------->account-service中扣减账户余额开始");
  15.         //模拟超时异常，全局事务回滚
  16.         //暂停几秒钟线程
  17.         try { TimeUnit.SECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); }
  18.         accountDao.decrease(userId,money);
  19.         LOGGER.info("------->account-service中扣减账户余额结束");
  20.     }
  21. }

另外，OpenFeign 的调用默认时间是 1s 以内，所以最后会抛异常。
数据库情况
image.png
故障情况

  • 当库存和账户金额扣减后，订单状态并没有设置为已经完成，没有从零改为1
  • 而且由于 feign 的重试机制，账户余额还有可能被多次扣减

超时异常，加了 **@GlobalTransactional**
用 @GlobalTransactional 标注 OrderServiceImpl 的 create() 方法。

  1. @Service
  2. @Slf4j
  3. public class OrderServiceImpl implements OrderService {
  5.     ...
  6.     /**
  7.      * 创建订单->调用库存服务扣减库存->调用账户服务扣减账户余额->修改订单状态
  8.      * 简单说：下订单->扣库存->减余额->改状态
  9.      */
  10.     @Override
  11.     //rollbackFor = Exception.class表示对任意异常都进行回滚
  12.     @GlobalTransactional(name = "fsp-create-order",rollbackFor = Exception.class)
  13.     public void create(Order order) {
  14.         ...
  15.     }
  16. }

还是模拟AccountServiceImpl添加超时，下单后数据库数据并没有任何改变，记录都添加不进来，达到出异常，数据库回滚的效果。