1. Seata产品模块

Seata中有三⼤模块，分别是TM、RM和TC。其中TM和RM是作为Seata的客户端与业务系统集成在⼀起，TC作为Seata的服务端独⽴部署。

  • TC (Transaction Coordinator) -事务协调者

维护全局和分⽀事务的状态，驱动全局事务提交或回滚。

  • TM (Transaction Manager) -事务管理器

定义全局事务的范围：开始全局事务、提交或回滚全局事务。

  • RM (Resource Manager) -资源管理器

管理分⽀事务处理的资源，与TC交谈以注册分⽀事务和报告分⽀事务的状态，并驱动分⽀事务提交或回滚。

Seata分布式事务框架-copy - 图1

在Seata(基于2PC)中，分布式事务的执⾏流程：

  • TM开启分布式事务, TM会 向TC注册全局事务记录；
  • 操作具体业务模块的数据库操作之前, RM会向TC注册分⽀事务；
  • 当业务操作完事后.TM会通知TC提交/回滚分布式事务；
  • TC汇总事务信息，决定分布式事务是提交还是回滚；
  • TC通知所有RM提交/回滚 资源，事务⼆阶段结束。

2. Seata-AT模式

2.1 案例引⼊及问题剖析

  1. 导⼊lagou_parent⼯程
    image.png
    2. 执⾏初始化SQL脚本,⾸先创建4个数据库
    seata_bussiness/seata_order/seata_points/seata_storage,在各⾃数据库执⾏SQL脚本
    seata_order数据库 ```sql */ SET NAMES utf8mb4; SET FOREIGN_KEY_CHECKS = 0;

— Table structure for t_order

DROP TABLE IF EXISTS t_order; CREATE TABLE t_order ( id bigint(0) NOT NULL COMMENT ‘订单id’, goods_Id int(0) NULL DEFAULT NULL COMMENT ‘商品ID’, num int(0) NULL DEFAULT NULL COMMENT ‘商品数量’, money decimal(10, 0) NULL DEFAULT NULL COMMENT ‘商品总⾦额’, create_time datetime(0) NULL DEFAULT NULL COMMENT ‘订单创建时间’, username varchar(50) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT ‘⽤户名称’, PRIMARY KEY (id) USING BTREE ) ENGINE = InnoDB AUTO_INCREMENT = 49 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;

— Records of t_order

SET FOREIGN_KEY_CHECKS = 1;

  1. seata_points数据库
  2. ```sql
  3. SET NAMES utf8mb4;
  4. SET FOREIGN_KEY_CHECKS = 0;
  5. -- ----------------------------
  6. -- Table structure for t_points
  7. -- ----------------------------
  8. DROP TABLE IF EXISTS `t_points`;
  9. CREATE TABLE `t_points` (
  10.  `id` int(0) NOT NULL AUTO_INCREMENT COMMENT '积分ID',
  11.  `username` varchar(50) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL
  12. COMMENT '⽤户名',
  13.  `points` int(0) NULL DEFAULT NULL COMMENT '⽤户积分',
  14.  PRIMARY KEY (`id`) USING BTREE
  15. ) ENGINE = InnoDB AUTO_INCREMENT = 3 CHARACTER SET = utf8 COLLATE =
  16. utf8_bin ROW_FORMAT = Dynamic;
  17. -- ----------------------------
  18. -- Records of t_points
  19. -- ----------------------------
  20. SET FOREIGN_KEY_CHECKS = 1;

seata_storage数据库

  1. SET NAMES utf8mb4;
  2. SET FOREIGN_KEY_CHECKS = 0;
  3. -- ----------------------------
  4. -- Table structure for t_storage
  5. -- ----------------------------
  6. DROP TABLE IF EXISTS `t_storage`;
  7. CREATE TABLE `t_storage` (
  8.  `id` int(0) NOT NULL AUTO_INCREMENT COMMENT '库存ID',
  9.  `goods_id` int(0) NULL DEFAULT NULL COMMENT '商品ID',
  10.   `storage` int(0) NULL DEFAULT NULL COMMENT '库存量',
  11.  PRIMARY KEY (`id`) USING BTREE
  12. ) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8 COLLATE =
  13. utf8_bin ROW_FORMAT = Dynamic;
  14. -- ----------------------------
  15. -- Records of t_storage
  16. -- ----------------------------
  17. INSERT INTO `t_storage` VALUES (1, 1, 100);
  18. SET FOREIGN_KEY_CHECKS = 1;
  1. 案例测试

**依次将4个服务启动.和nacos服务

**
访问路径为:

http://localhost:8000/test1 正常访问数据分别⼊库

http://localhost:8000/test2访问出错库存不⾜,导致服务调⽤失败.则观察数据库, 经发现订单与积

分数据库都已改变,⽽库存数据库没有减少库存, 所以不满⾜事务的特性.
image.png
image.png
image.png

2.2 AT模式介绍

AT 模式是⼀种⽆侵⼊的分布式事务解决⽅案。在 AT 模式下，⽤户只需关注⾃⼰的“业务 SQL”，⽤户的 “业务 SQL” 作为⼀阶段，Seata 框架会⾃动⽣成事务的⼆阶段提交和回滚操作。
image.png

2.3 AT模式原理

在介绍AT 模式的时候它是⽆侵⼊的分布式事务解决⽅案, 那么如何做到对业务的⽆侵⼊的呢?
1. ⼀阶段
在⼀阶段，Seata 会拦截“业务 SQL”，⾸先解析 SQL 语义，找到“业务 SQL”要更新的业务数据，在业务数据被更新前，将其保存成“before image”，然后执⾏“业务 SQL”更新业务数据，在业务数据更新之后，再将其保存成“after image”，最后⽣成⾏锁。以上操作全部在⼀个数据库事务内完成，这样保证了⼀阶段操作的原⼦性。

image.png
2. ⼆阶段

  • 提交
    ⼆阶段如果是提交的话，因为“业务 SQL”在⼀阶段已经提交⾄数据库， 所以 Seata 框架只需将⼀阶段保存的快照数据和⾏锁删掉，完成数据清理即可。

image.png

  • 回滚
    ⼆阶段如果是回滚的话，Seata 就需要回滚⼀阶段已经执⾏的“业务 SQL”，还原业务数据。回滚⽅式便是⽤“before image”还原业务数据；但在还原前要⾸先要校验脏写，对⽐“数据库当前业务数据”和 “after image”，如果两份数据完全⼀致就说明没有脏写，可以还原业务数据，如果不⼀致就说明有脏写，出现脏写就需要转⼈⼯处理。


image.png
AT 模式的⼀阶段、⼆阶段提交和回滚均由 Seata 框架⾃动⽣成，⽤户只需编写“业务SQL”，便能轻松接⼊分布式事务，AT 模式是⼀种对业务⽆任何侵⼊的分布式事务解决⽅案。

2.4 AT模式改造案例

2.4.1 Seata Server - TC全局事务协调器

介绍了 seata 事务的三个模块：TC（事务协调器）、TM（事务管理器）和RM（资源管理器），其中 TM 和 RM 是嵌⼊在业务应⽤中的，⽽ TC 则是⼀个独⽴服务。
Seata Server 就是 TC，直接从官⽅仓库下载启动即可，下载地址：https://github.com/seata/seata/releases

  1. registry.conf

    Seata Server 要向注册中⼼进⾏注册，这样，其他服务就可以通过注册中⼼去发现 Seata Server，与 Seata Server 进⾏通信。

Seata ⽀持多款注册中⼼服务：nacos 、eureka、redis、zk、consul、etcd3、sofa。
我们项⽬中要使⽤ nacos注册中⼼，nacos服务的连接地址、注册的服务名，这需要在seata/conf/registry.conf⽂件中进⾏配置：

  1. registry {
  2.   # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  3.   type = "nacos"
  4.   loadBalance = "RandomLoadBalance"
  5.   loadBalanceVirtualNodes = 10
  7.   nacos {
  8.     application = "seata-server"
  9.     serverAddr = "106.75.226.220:8848"
  10.     group = "SEATA_GROUP"
  11.     namespace = ""
  12.     cluster = "default"
  13.     username = "nacos"
  14.     password = "nacos"
  15.   }
  16.   eureka {
  17.     serviceUrl = "http://localhost:8761/eureka"
  18.     application = "default"
  19.     weight = "1"
  20.   }
  21.   redis {
  22.     serverAddr = "localhost:6379"
  23.     db = 0
  24.     password = ""
  25.     cluster = "default"
  26.     timeout = 0
  27.   }
  28.   zk {
  29.     cluster = "default"
  30.     serverAddr = "127.0.0.1:2181"
  31.     sessionTimeout = 6000
  32.     connectTimeout = 2000
  33.     username = ""
  34.     password = ""
  35.   }
  36.   consul {
  37.     cluster = "default"
  38.     serverAddr = "127.0.0.1:8500"
  39.   }
  40.   etcd3 {
  41.     cluster = "default"
  42.     serverAddr = "http://localhost:2379"
  43.   }
  44.   sofa {
  45.     serverAddr = "127.0.0.1:9603"
  46.     application = "default"
  47.     region = "DEFAULT_ZONE"
  48.     datacenter = "DefaultDataCenter"
  49.     cluster = "default"
  50.     group = "SEATA_GROUP"
  51.     addressWaitTime = "3000"
  52.   }
  53.   file {
  54.     name = "file.conf"
  55.   }
  56. }
  58. config {
  59.   # file、nacos 、apollo、zk、consul、etcd3
  60.   type = "nacos"
  62.   nacos {
  63.     serverAddr = "106.75.226.220:8848"
  64.     namespace = ""
  65.     group = "SEATA_GROUP"
  66.     username = "nacos"
  67.     password = "nacos"
  68.   }
  69.   consul {
  70.     serverAddr = "127.0.0.1:8500"
  71.   }
  72.   apollo {
  73.     appId = "seata-server"
  74.     apolloMeta = "http://192.168.1.204:8801"
  75.     namespace = "application"
  76.     apolloAccesskeySecret = ""
  77.   }
  78.   zk {
  79.     serverAddr = "127.0.0.1:2181"
  80.     sessionTimeout = 6000
  81.     connectTimeout = 2000
  82.     username = ""
  83.     password = ""
  84.   }
  85.   etcd3 {
  86.     serverAddr = "http://localhost:2379"
  87.   }
  88.   file {
  89.     name = "file.conf"
  90.   }
  91. }
  1. 向nacos中添加配置信息
    下载配置config.txt https://github.com/seata/seata/tree/develop/script/config-center
    image.png
    https://seata.io/zh-cn/docs/user/configurations.html针对每个⼀项配置介绍
  • 将config.txt⽂件放⼊seata⽬录下⾯
  • 修改config.txt信息

Server端存储的模式（store.mode）现有file,db,redis三种。主要存储全局事务会话信息,分⽀事务信息, 锁记录表信息,seata-server默认是file模式。file只能⽀持单机模式, 如果想要⾼可⽤模式的话可以切换db或者redis. 为了⽅便查看全局事务会话信息本次课程采⽤db数据库模式
存储模式

  1. store.mode=db

mysql数据库连接信息

  1. store.db.datasource=druid
  2. store.db.dbType=mysql
  3. store.db.driverClassName=com.mysql.jdbc.Driver
  4. store.db.url=jdbc:mysql://127.0.0.1:3306/seata?useUnicode=true
  5. store.db.user=root
  6. store.db.password=root
  7. store.db.minConn=5
  8. store.db.maxConn=30
  9. store.db.globalTable=global_table
  10. store.db.branchTable=branch_table
  11. store.db.queryLimit=100
  12. store.db.lockTable=lock_table
  13. store.db.maxWait=5000

注意1: 需要创建seata数据库
image.png
注意2: 需要创建global_table/branch_table/lock_table三张表,seata1.0以上就不⾃带数据库⽂件了，要⾃⼰去github下载，https://github.com/seata/seata/tree/develop/scrip![image.png](https://cdn.nlark.com/yuque/0/2021/png/12497888/1619059449220-43720d21-e0ef-4b14-936d-df874a88860a.png#clientId=ue7856afe-3e47-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=328&id=u9199c376&margin=%5Bobject%20Object%5D&name=image.png&originHeight=328&originWidth=649&originalType=binary&ratio=1&rotation=0&showTitle=false&size=23493&status=done&style=none&taskId=u2610e3b5-fd5d-4ebf-a6eb-4510ffc88aa&title=&width=649)

  1. -- -------------------------------- The script used when storeMode is 'db' --------------------------------
  2. -- the table to store GlobalSession data
  3. CREATE TABLE IF NOT EXISTS `global_table`
  4. (
  5.     `xid`                       VARCHAR(128) NOT NULL,
  6.     `transaction_id`            BIGINT,
  7.     `status`                    TINYINT      NOT NULL,
  8.     `application_id`            VARCHAR(32),
  9.     `transaction_service_group` VARCHAR(32),
  10.     `transaction_name`          VARCHAR(128),
  11.     `timeout`                   INT,
  12.     `begin_time`                BIGINT,
  13.     `application_data`          VARCHAR(2000),
  14.     `gmt_create`                DATETIME,
  15.     `gmt_modified`              DATETIME,
  16.     PRIMARY KEY (`xid`),
  17.     KEY `idx_gmt_modified_status` (`gmt_modified`, `status`),
  18.     KEY `idx_transaction_id` (`transaction_id`)
  19. ) ENGINE = InnoDB
  20.   DEFAULT CHARSET = utf8;
  22. -- the table to store BranchSession data
  23. CREATE TABLE IF NOT EXISTS `branch_table`
  24. (
  25.     `branch_id`         BIGINT       NOT NULL,
  26.     `xid`               VARCHAR(128) NOT NULL,
  27.     `transaction_id`    BIGINT,
  28.     `resource_group_id` VARCHAR(32),
  29.     `resource_id`       VARCHAR(256),
  30.     `branch_type`       VARCHAR(8),
  31.     `status`            TINYINT,
  32.     `client_id`         VARCHAR(64),
  33.     `application_data`  VARCHAR(2000),
  34.     `gmt_create`        DATETIME(6),
  35.     `gmt_modified`      DATETIME(6),
  36.     PRIMARY KEY (`branch_id`),
  37.     KEY `idx_xid` (`xid`)
  38. ) ENGINE = InnoDB
  39.   DEFAULT CHARSET = utf8;
  41. -- the table to store lock data
  42. CREATE TABLE IF NOT EXISTS `lock_table`
  43. (
  44.     `row_key`        VARCHAR(128) NOT NULL,
  45.     `xid`            VARCHAR(128),
  46.     `transaction_id` BIGINT,
  47.     `branch_id`      BIGINT       NOT NULL,
  48.     `resource_id`    VARCHAR(256),
  49.     `table_name`     VARCHAR(32),
  50.     `pk`             VARCHAR(36),
  51.     `gmt_create`     DATETIME,
  52.     `gmt_modified`   DATETIME,
  53.     PRIMARY KEY (`row_key`),
  54.     KEY `idx_branch_id` (`branch_id`)
  55. ) ENGINE = InnoDB
  56.   DEFAULT CHARSET = utf8;
  1. 启动seata-server
    image.png
    观察nacos服务列表
    image.png

2.4.2 TM/RM端整合Seata

AT 模式在RM端需要 UNDO_LOG 表,来记录每个RM的事务信息,主要包含数据修改前,后的相关信息,⽤于回滚处理,所以在所有数据库中分别执⾏

  1. -- 注意此处0.3.0+ 增加唯⼀索引 ux_undo_log
  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.  `ext` varchar(100) DEFAULT NULL,
  12.  PRIMARY KEY (`id`),
  13.  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
  14. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

TM/RM端整合seata⼀共有五个步骤:
image.png
事务分组:
image.png
RM(事务管理器)端整合Seata与TM(事务管理器)端步骤类似,只不过不需要在⽅法添加@GlobalTransactional注解,针对我们⼯程lagou_bussiness是事务的发起者,所以是TM端,其它⼯程为RM端. 所以我们只需要在lagou_common_db完成前4步骤即可

  1. ⼯程中添加Seata依赖
    lagou_parent添加seata依赖管理,⽤于seata的版本锁定
    1. <dependencyManagement>
    2. <dependencies>
    3. <!--spring cloud依赖管理，引⼊了Spring Cloud的版本-->
    4. <dependency>
    5. <groupId>org.springframework.cloud</groupId>
    6. <artifactId>spring-cloud-dependencies</artifactId>
    7. <version>Greenwich.RELEASE</version>
    8. <type>pom</type>
    9. <scope>import</scope>
    10. </dependency>
    11. <!--SCA -->
    12. <dependency>
    13. <groupId>mysql</groupId>
    14. <artifactId>mysql-connector-java</artifactId>
    15. <version>5.1.47</version>
    16. </dependency>
    17. <!--SCA -->
    18. <dependency>
    19. <groupId>com.alibaba.cloud</groupId>
    20. <artifactId>spring-cloud-alibaba-dependencies</artifactId>
    21. <version>2.1.0.RELEASE</version>
    22. <type>pom</type>
    23. <scope>import</scope>
    24. </dependency>
    25. <!--SCA -->
    26. <!--seata版本管理, ⽤于锁定⾼版本的seata -->
    27. <dependency>
    28. <groupId>io.seata</groupId>
    29. <artifactId>seata-all</artifactId>
    30. <version>1.3.0</version>
    31. </dependency>
    32. </dependencies>
    33. </dependencyManagement>
    在lagou_common_db⼯程添加seata依赖
    1. <!--seata依赖-->
    2. <dependency>
    3. <groupId>com.alibaba.cloud</groupId>
    4. <artifactId>spring-cloud-alibaba-seata</artifactId>
    5. <!--排除低版本seata依赖-->
    6. <exclusions>
    7. <exclusion>
    8. <groupId>io.seata</groupId>
    9. <artifactId>seata-all</artifactId>
    10. </exclusion>
    11. </exclusions>
    12. </dependency>
    13. <!--添加⾼版本seata依赖-->
    14. <dependency>
    15. <groupId>io.seata</groupId>
    16. <artifactId>seata-all</artifactId>
    17. <version>1.3.0</version>
    18. </dependency>

  2. 在common⼯程添加registry.conf依赖 ```xml registry {

    file 、nacos 、eureka、redis、zk、consul、etcd3、sofa

    type = “nacos” loadBalance = “RandomLoadBalance” loadBalanceVirtualNodes = 10

    nacos { application = “seata-server” serverAddr = “106.75.226.220:8848” group = “SEATA_GROUP” namespace = “” cluster = “default” username = “nacos” password = “nacos” } eureka { serviceUrl = “http://localhost:8761/eureka“ application = “default” weight = “1” } redis { serverAddr = “localhost:6379” db = 0 password = “” cluster = “default” timeout = 0 } zk { cluster = “default” serverAddr = “127.0.0.1:2181” sessionTimeout = 6000 connectTimeout = 2000 username = “” password = “” } consul { cluster = “default” serverAddr = “127.0.0.1:8500” } etcd3 { cluster = “default” serverAddr = “http://localhost:2379“ } sofa { serverAddr = “127.0.0.1:9603” application = “default” region = “DEFAULT_ZONE” datacenter = “DefaultDataCenter” cluster = “default” group = “SEATA_GROUP” addressWaitTime = “3000” } file { name = “file.conf” } }

config {

file、nacos 、apollo、zk、consul、etcd3

type = “nacos”

nacos { serverAddr = “106.75.226.220:8848” namespace = “” group = “SEATA_GROUP” username = “nacos” password = “nacos” } consul { serverAddr = “127.0.0.1:8500” } apollo { appId = “seata-server” apolloMeta = “http://192.168.1.204:8801“ namespace = “application” apolloAccesskeySecret = “” } zk { serverAddr = “127.0.0.1:2181” sessionTimeout = 6000 connectTimeout = 2000 username = “” password = “” } etcd3 { serverAddr = “http://localhost:2379“ } file { name = “file.conf” } }

  1. 3. 添加公共配置
  2. ```xml
  3. spring.cloud.alibaba.seata.tx-service-group=my_test_tx_group  //分组名称为nacos上配置文件的名称
  4. logging.level.io.seata=debug

image.png
image.png
指定值为default,对应的配置为：
image.png
4. 在每个模块下引⼊公共配置⽂件

  1. profiles
  2.  active: seata
  1. 编译数据源代理
    1. package com.lagou.common_db;
    2. import com.alibaba.druid.pool.DruidDataSource;
    3. import io.seata.rm.datasource.DataSourceProxy;
    4. import
    5. org.springframework.boot.context.properties.ConfigurationProperties;
    6. import org.springframework.context.annotation.Bean;
    7. import org.springframework.context.annotation.Configuration;
    8. import org.springframework.context.annotation.Primary;
    9. import javax.sql.DataSource;
    10. @Configuration
    11. public class DataSourceConfiguration {
    12. /**
    13. * 使⽤druid连接池
    14. *
    15. * @return
    16. */
    17. @Bean
    18. @ConfigurationProperties(prefix = "spring.datasource")
    19. public DataSource druidDataSource() {
    20. return new DruidDataSource();
    21. }
    22. /**
    23. * 设置数据源代理-,完成分⽀事务注册/事务提交与回滚等操作
    24. * @param druidDataSource
    25. * @return
    26. */
    27. @Primary //设置⾸选数据源对象
    28. @Bean("dataSource")
    29. public DataSourceProxy dataSource(DataSource druidDataSource) {
    30. return new DataSourceProxy(druidDataSource);
    31. }
    32. }
    启动扫描配置类,分别加载每个⼯程的启动类中
    1. @SpringBootApplication(exclude = DataSourceAutoConfiguration.class,
    2. scanBasePackages = "com.lagou")
  2. 添加注解@GlobalTransactional ```java /**
  • 商品销售 *
  • @param goodsId 商品id
  • @param num 销售数量
  • @param username ⽤户名
  • @param money ⾦额 */ @GlobalTransactional(name = “sale”, timeoutMills = 100000, rollbackFor =Exception.class) //@Transactional public void sale(Integer goodsId, Integer num, Double money, String username) { //创建订单 orderServiceFeign.addOrder(idWorker.nextId(),goodsId, num, money,username); //增加积分 pointsServiceFeign.increase(username, (int) (money / 10)); //扣减库存 storageServiceFeign.decrease(goodsId, num); } ```

    3.Seata-TCC模式

    3.1 TCC模式介绍

    Seata 开源了 TCC 模式，该模式由蚂蚁⾦服贡献。TCC 模式需要⽤户根据⾃⼰的业务场景实现Try、Confirm 和 Cancel 三个操作；事务发起⽅在⼀阶段 执⾏ Try ⽅式，在⼆阶段提交执⾏ Confirm⽅法，⼆阶段回滚执⾏ Cancel ⽅法。

image.png

TCC 三个⽅法描述：

  • Try：资源的检测和预留；
  • Confirm：执⾏的业务操作提交；要求 Try 成功 Confirm ⼀定要能成功；
  • Cancel：预留资源释放。

**业务模型分 2 阶段设计：

**
⽤户接⼊ TCC ，最重要的是考虑如何将⾃⼰的业务模型拆成两阶段来实现。
以“扣钱”场景为例，在接⼊ TCC 前，对 A 账户的扣钱，只需⼀条更新账户余额的 SQL 便能完成；
但是在接⼊ TCC 之后，⽤户就需要考虑如何将原来⼀步就能完成的扣钱操作，拆成两阶段，实现成三个⽅法，并且保证⼀阶段 Try 成功的话 ⼆阶段 Confirm ⼀定能成功。
image.png

Try ⽅法作为⼀阶段准备⽅法，需要做资源的检查和预留。在扣钱场景下，Try 要做的事情是就是检查账户余额是否充⾜，预留转账资⾦，预留的⽅式就是冻结 A 账户的 转账资⾦。Try ⽅法执⾏之后，
账号 A 余额虽然还是 100，但是其中 30 元已经被冻结了，不能被其他事务使⽤。⼆阶段 Confirm ⽅法执⾏真正的扣钱操作。Confirm 会使⽤ Try 阶段冻结的资⾦，执⾏账号扣款。
Confirm ⽅法执⾏之后，账号 A 在⼀阶段中冻结的 30 元已经被扣除，账号 A 余额变成 70 元 。
如果⼆阶段是回滚的话，就需要在 Cancel ⽅法内释放⼀阶段 Try 冻结的 30 元，使账号 A 的回到初始状态，100 元全部可⽤。

⽤户接⼊ TCC 模式，最重要的事情就是考虑如何将业务模型拆成 2 阶段，实现成 TCC 的 3 个⽅法，并且保证 Try 成功 Confirm ⼀定能成功。相对于 AT 模式，TCC 模式对业务代码有⼀定的侵⼊性，但是 TCC 模式⽆ AT 模式的全局⾏锁，TCC 性能会⽐ AT 模式⾼很多。

3.2 TCC模式改造案例

3.2.1 RM端改造


针对RM端,实现起来需要完成try/commit/rollback的实现,所以步骤相对较多但是前三步骤和AT模式⼀样
image.png

  1. 修改数据库表结构,增加预留检查字段,⽤于提交和回滚
    1. ALTER TABLE `seata_order`.`t_order` ADD COLUMN `status` int(0) NULL
    2. COMMENT '订单状态-0不可⽤,事务未提交 , 1-可⽤,事务提交' ;
    3. ALTER TABLE `seata_points`.`t_points` ADD COLUMN `frozen_points` int(0)
    4. NULL DEFAULT 0 COMMENT '冻结积分' AFTER `points`;
    5. ALTER TABLE `seata_storage`.`t_storage` ADD COLUMN `frozen_storage` int(0)
    6. NULL DEFAULT 0 COMMENT '冻结库存' AFTER `goods_id`;
  2. lagou_order⼯程改造
    接⼝ ```sql import com.lagou.order.entity.Order; import io.seata.rm.tcc.api.BusinessActionContext; import io.seata.rm.tcc.api.BusinessActionContextParameter; import io.seata.rm.tcc.api.LocalTCC; import io.seata.rm.tcc.api.TwoPhaseBusinessAction; /**
  • @LocalTCC 该注解需要添加到上⾯描述的接⼝上，表示实现该接⼝的类被 seata 来管 理，seata 根据事务的状态，
  • ⾃动调⽤我们定义的⽅法，如果没问题则调⽤ Commit ⽅法，否则调⽤ Rollback ⽅ 法。 / @LocalTCC public interface OrderService extends IService { /*
    • @TwoPhaseBusinessAction 描述⼆阶段提交
    • name: 为 tcc⽅法的 bean 名称，需要全局唯⼀，⼀般写⽅法名即可
    • commitMethod: Commit⽅法的⽅法名
    • rollbackMethod:Rollback⽅法的⽅法名
    • @BusinessActionContextParamete 该注解⽤来修饰 Try⽅法的⼊参，
    • 被修饰的⼊参可以在 Commit ⽅法和 Rollback ⽅法中通过 BusinessActionContext 获取。 */ @TwoPhaseBusinessAction(name = “addTcc”, commitMethod = “addCommit”, rollbackMethod = “addRollBack”) void add(@BusinessActionContextParameter(paramName = “order”) Order order); public boolean addCommit(BusinessActionContext context); public boolean addRollBack(BusinessActionContext context); }
      1. 实现类
      2. ```java
      3. import java.util.Date;
      4. @Slf4j
      5. @Service
      6. public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order>
      7. implements OrderService {
      8. @Override
      9. public void add(Order order) {
      10. order.setCreateTime(new Date());//设置订单创建时间
      11. order.setStatus(0);//try阶段-预检查
      12. this.save(order);//保存订单
      13. }
      14. @Override
      15. public boolean addCommit(BusinessActionContext context) {
      16. Order order =
      17. JSON.parseObject(context.getActionContext("order").toString(),
      18. Order.class);
      19. order = this.getById(order.getId());
      20. if (order != null) {
      21. order.setStatus(1);//commit阶段-提交事务
      22. this.saveOrUpdate(order);//修改订单
      23. }
      24. log.info("--------->xid=" + context.getXid() + " 提交成功!");
      25. return true;
      26. }
      27. @Override
      28. public boolean addRollBack(BusinessActionContext context) {
      29. Order order =
      30. JSON.parseObject(context.getActionContext("order").toString(),
      31. Order.class);
      32. order = this.getById(order.getId());
      33. if (order != null) {
      34. this.removeById(order.getId());//删除订单
      35. }
      36. log.info("--------->xid=" + context.getXid() + " 回滚成功!");
      37. return true;
      38. }
      39. }
  1. lagou_points⼯程改造
    接⼝改造
    image.png
    实现类改造 ```java import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.lagou.points.mapper.PointsMapper; import com.lagou.points.entity.Points; import com.lagou.points.service.PointsService; import io.seata.rm.tcc.api.BusinessActionContext; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; /**
  • 会员积分服务 / @Slf4j @Service public class PointsServiceImpl extends ServiceImpl implements PointsService { /*
    • 会员增加积分 *
    • @param username ⽤户名
    • @param points 增加的积分
    • @return 积分对象 */ public void increase(String username, Integer points) { QueryWrapper wrapper = new QueryWrapper(); wrapper.lambda().eq(Points::getUsername, username); Points userPoints = this.getOne(wrapper); if (userPoints == null) { userPoints = new Points(); userPoints.setUsername(username); //userPoints.setPoints(points); 不直接增加积分 userPoints.setFrozenPoints(points);//设置冻结积分 this.save(userPoints); } else { userPoints.setFrozenPoints(points);//设置冻结积分 this.saveOrUpdate(userPoints); } } @Override public boolean increaseCommit(BusinessActionContext context) { //查询⽤户积分 QueryWrapper wrapper = new QueryWrapper(); wrapper.lambda().eq(Points::getUsername, context.getActionContext(“username”)); Points userPoints = this.getOne(wrapper); if (userPoints != null) { //增加⽤户积分 userPoints.setPoints(userPoints.getPoints() + userPoints.getFrozenPoints()); //冻结积分清零 userPoints.setFrozenPoints(0); this.saveOrUpdate(userPoints); } log.info(“————->xid=” + context.getXid() + “ 提交成功!”); return true; } @Override public boolean increaseRollback(BusinessActionContext context) { //查询⽤户积分 QueryWrapper wrapper = new QueryWrapper(); wrapper.lambda().eq(Points::getUsername, context.getActionContext(“username”)); Points userPoints = this.getOne(wrapper); if (userPoints != null) { //冻结积分清零 userPoints.setFrozenPoints(0); this.saveOrUpdate(userPoints); } log.info(“————->xid=” + context.getXid() + “ 回滚成功!”); return true; } } ```
  1. lagou_stroage⼯程改造

接⼝改造
image.png
实现类改造

  1. /**
  2. * 仓库服务
  3. */
  4. @Slf4j
  5. @Service
  6. public class StorageServiceImpl extends ServiceImpl<StorageMapper,
  7. Storage> implements StorageService {
  8.  /**
  9.  * 减少库存
  10.  *
  11.  * @param goodsId 商品ID
  12.  * @param quantity 减少数量
  13.  * @return 库存对象
  14.  */
  15.  public void decrease(Integer goodsId, Integer quantity) {
  16.  QueryWrapper<Storage> wrapper = new QueryWrapper<Storage>();
  17.  wrapper.lambda().eq(Storage::getGoodsId, goodsId);
  18.  Storage goodsStorage = this.getOne(wrapper);
  19.  if (goodsStorage.getStorage() >= quantity) {
  20.  //goodsStorage.setStorage(goodsStorage.getStorage() -
  21. quantity);
  22.  //设置冻结库存
  23.  goodsStorage.setFrozenStorage(quantity);
  24.  } else {
  25.  throw new RuntimeException(goodsId + "库存不⾜,⽬前剩余库存:"
  26. + goodsStorage.getStorage());
  27.  }
  28.  this.saveOrUpdate(goodsStorage);
  29.  }
  30.  @Override
  31.  public boolean decreaseCommit(BusinessActionContext context) {
  32.  QueryWrapper<Storage> wrapper = new QueryWrapper<Storage>();
  33.  wrapper.lambda().eq(Storage::getGoodsId,
  34. context.getActionContext("goodsId"));
  35.  Storage goodsStorage = this.getOne(wrapper);
  36.  if (goodsStorage != null) {
  37.  //扣减库存
  38.  goodsStorage.setStorage(goodsStorage.getStorage() -
  39. goodsStorage.getFrozenStorage());
  40.      //冻结库存清零
  41.  goodsStorage.setFrozenStorage(0);
  42.  this.saveOrUpdate(goodsStorage);
  43.  }
  44.  log.info("--------->xid=" + context.getXid() + " 提交成功!");
  45.  return true;
  46.  }
  47.  @Override
  48.  public boolean decreaseRollback(BusinessActionContext context) {
  49.  QueryWrapper<Storage> wrapper = new QueryWrapper<Storage>();
  50.  wrapper.lambda().eq(Storage::getGoodsId,
  51. context.getActionContext("goodsId"));
  52.  Storage goodsStorage = this.getOne(wrapper);
  53.  if (goodsStorage != null) {
  54.  //冻结库存清零
  55.  goodsStorage.setFrozenStorage(0);
  56.  this.saveOrUpdate(goodsStorage);
  57.  }
  58.  log.info("--------->xid=" + context.getXid() + " 回滚成功!");
  59.  return true;
  60.  }
  61. }

3.2.2 TM端改造

针对我们⼯程lagou_bussiness是事务的发起者,所以是TM端,其它⼯程为RM端. 所以我们只需要在
lagou_common_db完成即可,因为lagou_bussiness⽅法⾥⾯没有对数据库操作.所以只需要将之前AT模式的代理数据源去掉即可.注意:如果lagou_bussiness也对数据库操作了.也需要完成
try/commit/rollback的实现.
image.png
代码实现:
image.png

4.Seata-Saga模式简介与三种模式对⽐

4.1 Saga模式简单介绍

Saga 模式是 Seata 开源的⻓事务解决⽅案，将由蚂蚁⾦服主要贡献。在 Saga 模式下，分布式事务内有多个参与者，每⼀个参与者都是⼀个冲正补偿服务，需要⽤户根据业务场景实现其正向操作和逆向回滚操作。

分布式事务执⾏过程中，依次执⾏各参与者的正向操作，如果所有正向操作均执⾏成功，那么分布式事务提交。如果任何⼀个正向操作执⾏失败，那么分布式事务会去退回去执⾏前⾯各参与者的逆向回滚操作，回滚已提交的参与者，使分布式事务回到初始状态。
image.png
**适⽤场景：

**
业务流程⻓、业务流程多
参与者包含第三⽅公司或遗留系统服务，⽆法提供 TCC 模式要求的三个接⼝
典型业务系统：如⾦融⽹络（与外部⾦融机构对接）、互联⽹微贷、渠道整合等业务系统

4.2 三种模式对⽐

image.png

**AT 模式是⽆侵⼊的分布式事务解决⽅案，适⽤于不希望对业务进⾏改造的场景，⼏乎0学习成本。


TCC 模式是⾼性能分布式事务解决⽅案，适⽤于核⼼系统等对性能有很⾼要求的场景。
**Saga 模式是⻓事务解决⽅案，适⽤于业务流程⻓且需要保证事务最终⼀致性的业务系统，Saga 模式⼀阶段就会提交本地事务，⽆锁，⻓流程情况下可以保证性能，多⽤于渠道层、集成层业务系统。事务参与者可能是其它公司的服务或者是遗留系统的服务，⽆法进⾏改造和提供 TCC 要求的接⼝，也可以使⽤Saga 模式。