SpringCloud Alibaba篇 - Seata【分布式事务】

逃离我推掉我的手 2023-10-03 11:03 28阅读 0赞

⑤ Seata【分布式事务】

5.1 分布式事务的由来

实例

单体应用被拆分成微服务应用，原来的三个模块被拆分成三个独立的应用,分别使用三个独立的数据源,
业务操作需要调用三个服务来完成。此时每个服务内部的数据一致性由本地事务来保证， 但是全局的数据一致性问题没法保证。

在这里插入图片描述

总结：一次业务操作需要跨多个数据源或需要跨多个系统进行远程调用,就会产生分布式事务问题。

5.2 Seata简介

Seata是一款开源的分布式事务解决方案，致力于在微服务架构下提供高性能和简单易用的分布式事务服务.

官网：http://seata.io/zh-cn/

分布式处理事务过程的 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下管辖的全部分支事务完成提交或回滚请求。

Seata的简单使用(怎么玩)

在spring的时候是通过注解@Transactional 管理本地事务

在这里插入图片描述

5.3 Seata-server的下载与安装

下载地址： https://github.com/seata/seata/releases

  • 先从官网中下载好Seata-server并解压到文件夹

  • 先备份file.conf,然后修改file.conf

    • 修改service模块，1.0以后没有这个选项，就默认是my_test_tx_group分组

      vgroup_mapping.my_test_tx_group = “fsp_tx_group”

    • 修改store模块

      store {

      store mode: file、db

      mode = “db”
      …

      database store

      db {

      1. ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
      2. datasource = "dbcp"
      3. ## mysql/oracle/h2/oceanbase etc.
      4. db-type = "mysql"
      5. driver-class-name = "com.mysql.jdbc.Driver"
      6. url = "jdbc:mysql://127.0.0.1:3306/seata"
      7. user = "root"
      8. password = "cvzhanshi"
      9. min-conn = 1
      10. max-conn = 3
      11. global.table = "global_table"
      12. branch.table = "branch_table"
      13. lock-table = "lock_table"
      14. query-limit = 100

      }
      }

  • 备份file.example.conf,然后修改file.example.conf,修改内容和file.conf一样
  • 备份registry.conf，修改文件

设置注册进nacos中

  1. type = "nacos"
  3. nacos {
  4. serverAddr = "localhost:8848"
  5. namespace = ""
  6. cluster = "default"
  7. }

  • 创建数据库和表

    创建seata数据库

    create database seata character set utf8;
    use seata;

    创建seata数据库需要的表（三张表）

    CREATE TABLE IF NOT EXISTS global_table
    (

    1. `xid`                       VARCHAR(128) NOT NULL,
    2. `transaction_id`            BIGINT,
    3. `status`                    TINYINT      NOT NULL,
    4. `application_id`            VARCHAR(32),
    5. `transaction_service_group` VARCHAR(32),
    6. `transaction_name`          VARCHAR(128),
    7. `timeout`                   INT,
    8. `begin_time`                BIGINT,
    9. `application_data`          VARCHAR(2000),
    10. `gmt_create`                DATETIME,
    11. `gmt_modified`              DATETIME,
    12. PRIMARY KEY (`xid`),
    13. KEY `idx_gmt_modified_status` (`gmt_modified`, `status`),
    14. KEY `idx_transaction_id` (`transaction_id`)

    ) ENGINE = InnoDB
    DEFAULT CHARSET = utf8;

    CREATE TABLE IF NOT EXISTS branch_table
    (

    1. `branch_id`         BIGINT       NOT NULL,
    2. `xid`               VARCHAR(128) NOT NULL,
    3. `transaction_id`    BIGINT,
    4. `resource_group_id` VARCHAR(32),
    5. `resource_id`       VARCHAR(256),
    6. `branch_type`       VARCHAR(8),
    7. `status`            TINYINT,
    8. `client_id`         VARCHAR(64),
    9. `application_data`  VARCHAR(2000),
    10. `gmt_create`        DATETIME(6),
    11. `gmt_modified`      DATETIME(6),
    12. PRIMARY KEY (`branch_id`),
    13. KEY `idx_xid` (`xid`)

    ) ENGINE = InnoDB
    DEFAULT CHARSET = utf8;

    CREATE TABLE IF NOT EXISTS lock_table
    (

    1. `row_key`        VARCHAR(128) NOT NULL,
    2. `xid`            VARCHAR(96),
    3. `transaction_id` BIGINT,
    4. `branch_id`      BIGINT       NOT NULL,
    5. `resource_id`    VARCHAR(256),
    6. `table_name`     VARCHAR(32),
    7. `pk`             VARCHAR(36),
    8. `gmt_create`     DATETIME,
    9. `gmt_modified`   DATETIME,
    10. PRIMARY KEY (`row_key`),
    11. KEY `idx_branch_id` (`branch_id`)

    ) ENGINE = InnoDB
    DEFAULT CHARSET = utf8;

  • 启动nacos和seata测试

在这里插入图片描述

在这里插入图片描述

seate在nacos中注册成功，说明安装完成

5.4 订单/库存/账户业务数据库准备

业务说明：

这里我们会创建三个服务,一个订单服务, 一个库存服务, 一个账户服务。

当用户下单时，会在订单服务中创建一个订单 ，然后通过远程调用库存服务来扣减下单商品的库存,
再通过远程调用账户服务来扣减用户账户里面的余额,最后在订单服务中修改订单状态为已完成。

该操作跨越三三个数据库,有两次远程调用，很明显会有分布式事务问题。

  • 创建三个数据库

    create database seata_order; —订单
    create database seata_storage; —库存
    create database seata_account; —账户

  • 在每个库下建表

seata_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=7 DEFAULT CHARSET=utf8;
  10. select * from t_order;
  12. --回滚日志
  13. CREATE TABLE `undo_log` (
  14.   `id` bigint(20) NOT NULL AUTO_INCREMENT,
  15.   `branch_id` bigint(20) NOT NULL,
  16.   `xid` varchar(100) NOT NULL,
  17.   `context` varchar(128) NOT NULL,
  18.   `rollback_info` longblob NOT NULL,
  19.   `log_status` int(11) NOT NULL,
  20.   `log_created` datetime NOT NULL,
  21.   `log_modified` datetime NOT NULL,
  22.   `ext` varchar(100) DEFAULT NULL,
  23.   PRIMARY KEY (`id`),
  24.   UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
  25. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

seata_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=2 DEFAULT CHARSET=utf8;
  9. INSERT INTO t_storage(`id`,`product_id`,`total`,`used`,`residue`)VALUES('1','1','100','0','100');
  11. --回滚日志
  12. CREATE TABLE `undo_log` (
  13.   `id` bigint(20) NOT NULL AUTO_INCREMENT,
  14.   `branch_id` bigint(20) NOT NULL,
  15.   `xid` varchar(100) NOT NULL,
  16.   `context` varchar(128) NOT NULL,
  17.   `rollback_info` longblob NOT NULL,
  18.   `log_status` int(11) NOT NULL,
  19.   `log_created` datetime NOT NULL,
  20.   `log_modified` datetime NOT NULL,
  21.   `ext` varchar(100) DEFAULT NULL,
  22.   PRIMARY KEY (`id`),
  23.   UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
  24. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

seata_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=2 DEFAULT CHARSET=utf8;
  9. INSERT INTO t_account(`id`,`user_id`,`total`,`used`,`residue`)VALUES('1','1','1000','0','1000');
  11. --回滚日志
  12. CREATE TABLE `undo_log` (
  13.   `id` bigint(20) NOT NULL AUTO_INCREMENT,
  14.   `branch_id` bigint(20) NOT NULL,
  15.   `xid` varchar(100) NOT NULL,
  16.   `context` varchar(128) NOT NULL,
  17.   `rollback_info` longblob NOT NULL,
  18.   `log_status` int(11) NOT NULL,
  19.   `log_created` datetime NOT NULL,
  20.   `log_modified` datetime NOT NULL,
  21.   `ext` varchar(100) DEFAULT NULL,
  22.   PRIMARY KEY (`id`),
  23.   UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
  24. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

结果：

在这里插入图片描述

5.5 订单/库存/账户业务微服务准备

业务需求：下订单->减库存->扣余额->改(订单)状态

5.5.1 订单模块

  • 新建模块seata-order-service2001

  • 导入依赖




    com.alibaba.cloud
    spring-cloud-starter-alibaba-nacos-discovery



    com.alibaba.cloud
    spring-cloud-starter-alibaba-seata


    io.seata
    seata-all


    io.seata
    seata-spring-boot-starter




    io.seata
    seata-all
    1.2.0


    io.seata
    seata-spring-boot-starter
    1.2.0


    org.springframework.cloud
    spring-cloud-starter-openfeign


    org.mybatis.spring.boot
    mybatis-spring-boot-starter



    com.alibaba
    druid-spring-boot-starter


    mysql
    mysql-connector-java



    org.springframework.boot
    spring-boot-starter-jdbc


    org.springframework.boot
    spring-boot-starter-web


    org.springframework.boot
    spring-boot-starter-actuator


    org.projectlombok
    lombok
    true


    org.springframework.boot
    spring-boot-starter-test
    test

  • 编写配置文件

    server:
    port: 2001

    spring:
    application:

    1. name: seata-order-service

    cloud:

    1. alibaba:
    2.   seata:
    3.     # 自定义事务组名称需要与seata-server中的对应
    4.     tx-service-group: my_test_tx_group #因为seata的file.conf文件中没有service模块，事务组名默认为my_test_tx_group
    5.     #service要与tx-service-group对齐，vgroupMapping和grouplist在service的下一级，my_test_tx_group在再下一级
    6.     service:
    7.       vgroupMapping:
    8.         #要和tx-service-group的值一致
    9.         my_test_tx_group: default
    10.       grouplist:
    11.         # seata seaver的 地址配置，此处可以集群配置是个数组
    12.         default: 192.168.1.104:8091
    13. nacos:
    14.   discovery:
    15.     server-addr: localhost:8848  #nacos

    datasource:

    1. # 当前数据源操作类型
    2. type: com.alibaba.druid.pool.DruidDataSource
    3. # mysql驱动类
    4. driver-class-name: com.mysql.jdbc.Driver
    5. url: jdbc:mysql://localhost:3306/seata_order
    6. username: root
    7. password: lian0911

    feign:
    hystrix:

    1. enabled: false

    logging:
    level:

    1. io:
    2.   seata: info

    mybatis:
    mapperLocations: classpath:mapper/.xml

  • 编写 file.conf 文件

内容在解压的seata中从conf中的README.md中有链接

  1. transport {
  2.   # tcp, unix-domain-socket
  3.   type = "TCP"
  4.   #NIO, NATIVE
  5.   server = "NIO"
  6.   #enable heartbeat
  7.   heartbeat = true
  8.   # the client batch send request enable
  9.   enableClientBatchSendRequest = true
  10.   #thread factory for netty
  11.   threadFactory {
  12.     bossThreadPrefix = "NettyBoss"
  13.     workerThreadPrefix = "NettyServerNIOWorker"
  14.     serverExecutorThread-prefix = "NettyServerBizHandler"
  15.     shareBossWorker = false
  16.     clientSelectorThreadPrefix = "NettyClientSelector"
  17.     clientSelectorThreadSize = 1
  18.     clientWorkerThreadPrefix = "NettyClientWorkerThread"
  19.     # netty boss thread size
  20.     bossThreadSize = 1
  21.     #auto default pin or 8
  22.     workerThreadSize = "default"
  23.   }
  24.   shutdown {
  25.     # when destroy server, wait seconds
  26.     wait = 3
  27.   }
  28.   serialization = "seata"
  29.   compressor = "none"
  30. }
  31. service {
  32.   #transaction service group mapping
  33.   vgroupMapping.my_test_tx_group = "default"
  34.   #only support when registry.type=file, please don't set multiple addresses
  35.   default.grouplist = "127.0.0.1:8091"
  36.   #degrade, current not support
  37.   enableDegrade = false
  38.   #disable seata
  39.   disableGlobalTransaction = false
  40. }
  42. client {
  43.   rm {
  44.     asyncCommitBufferLimit = 10000
  45.     lock {
  46.       retryInterval = 10
  47.       retryTimes = 30
  48.       retryPolicyBranchRollbackOnConflict = true
  49.     }
  50.     reportRetryCount = 5
  51.     tableMetaCheckEnable = false
  52.     tableMetaCheckerInterval = 60000
  53.     reportSuccessEnable = false
  54.     sagaBranchRegisterEnable = false
  55.     sagaJsonParser = jackson
  56.     sagaRetryPersistModeUpdate = false
  57.     sagaCompensatePersistModeUpdate = false
  58.     tccActionInterceptorOrder = -2147482648 #Ordered.HIGHEST_PRECEDENCE + 1000
  59.   }
  60.   tm {
  61.     commitRetryCount = 5
  62.     rollbackRetryCount = 5
  63.     defaultGlobalTransactionTimeout = 60000
  64.     degradeCheck = false
  65.     degradeCheckPeriod = 2000
  66.     degradeCheckAllowTimes = 10
  67.     interceptorOrder = -2147482648 #Ordered.HIGHEST_PRECEDENCE + 1000
  68.   }
  69.   undo {
  70.     dataValidation = true
  71.     onlyCareUpdateColumns = true
  72.     logSerialization = "jackson"
  73.     logTable = "undo_log"
  74.     compress {
  75.       enable = true
  76.       # allow zip, gzip, deflater, 7z, lz4, bzip2, default is zip
  77.       type = zip
  78.       # if rollback info size > threshold, then will be compress
  79.       # allow k m g t
  80.       threshold = 64k
  81.     }
  82.   }
  83.   loadBalance {
  84.       type = "RandomLoadBalance"
  85.       virtualNodes = 10
  86.   }
  87. }
  88. log {
  89.   exceptionRate = 100
  90. }
  • 编写 registry.conf

内容和file.conf一样查找

  1. registry {
  2.   # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa、custom
  3.   type = "nacos"
  5.   nacos {
  6.     application = "seata-server"
  7.     serverAddr = "127.0.0.1:8848"
  8.     group = "SEATA_GROUP"
  9.     namespace = ""
  10.     username = ""
  11.     password = ""
  12.   }
  13.   eureka {
  14.     serviceUrl = "http://localhost:8761/eureka"
  15.     weight = "1"
  16.   }
  17.   redis {
  18.     serverAddr = "localhost:6379"
  19.     db = "0"
  20.     password = ""
  21.     timeout = "0"
  22.   }
  23.   zk {
  24.     serverAddr = "127.0.0.1:2181"
  25.     sessionTimeout = 6000
  26.     connectTimeout = 2000
  27.     username = ""
  28.     password = ""
  29.   }
  30.   consul {
  31.     serverAddr = "127.0.0.1:8500"
  32.     aclToken = ""
  33.   }
  34.   etcd3 {
  35.     serverAddr = "http://localhost:2379"
  36.   }
  37.   sofa {
  38.     serverAddr = "127.0.0.1:9603"
  39.     region = "DEFAULT_ZONE"
  40.     datacenter = "DefaultDataCenter"
  41.     group = "SEATA_GROUP"
  42.     addressWaitTime = "3000"
  43.   }
  44.   file {
  45.     name = "file.conf"
  46.   }
  47.   custom {
  48.     name = ""
  49.   }
  50. }
  52. config {
  53.   # file、nacos 、apollo、zk、consul、etcd3、springCloudConfig、custom
  54.   type = "file"
  56.   nacos {
  57.     serverAddr = "127.0.0.1:8848"
  58.     namespace = ""
  59.     group = "SEATA_GROUP"
  60.     username = ""
  61.     password = ""
  62.     dataId = "seata.properties"
  63.   }
  64.   consul {
  65.     serverAddr = "127.0.0.1:8500"
  66.     aclToken = ""
  67.   }
  68.   apollo {
  69.     appId = "seata-server"
  70.     apolloMeta = "http://192.168.1.204:8801"
  71.     namespace = "application"
  72.     apolloAccesskeySecret = ""
  73.   }
  74.   zk {
  75.     serverAddr = "127.0.0.1:2181"
  76.     sessionTimeout = 6000
  77.     connectTimeout = 2000
  78.     username = ""
  79.     password = ""
  80.     nodePath = "/seata/seata.properties"
  81.   }
  82.   etcd3 {
  83.     serverAddr = "http://localhost:2379"
  84.   }
  85.   file {
  86.     name = "file.conf"
  87.   }
  88.   custom {
  89.     name = ""
  90.   }
  91. }
  • domain包实体类编写

Order

  1. @Data
  2. @AllArgsConstructor
  3. @NoArgsConstructor
  4. public class Order {
  7.     private Long id;
  9.     private Long userId;
  11.     private Long productId;
  13.     private Integer count;
  15.     private BigDecimal money;
  17.     private Integer status; // 订单状态 0：创建中 1：已完结
  18. }

CommonResult

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 21:30
  4.  */
  5. @Data
  6. @AllArgsConstructor
  7. @NoArgsConstructor
  8. public class CommonResult<T> {
  10.     private Integer code;
  11.     private String message;
  12.     private T data;
  14.     public CommonResult(Integer code, String message) {
  16.         this(code, message, null);
  17.     }
  18. }
  • dao层的编写

OrderDao

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 21:33
  4.  */
  5. @Mapper
  6. @Repository
  7. public interface OrderDao {
  9.     //1 新建订单
  10.     void create(Order order);
  12.     //2 修改订单状态,从0改为1
  13.     void update(@Param("userId") Long userId, @Param("status") Integer status);
  14. }

对应的映射文件classpath:mapper/OrderMapper.xml

  1. <mapper namespace="cn.cvzhanshi.springcloud.alibaba.dao.OrderDao">
  3.     <resultMap id="BaseResultMap" type="cn.cvzhanshi.springcloud.alibaba.domain.Order">
  4.         <id column="id" property="id" jdbcType="BIGINT" />
  5.         <result column="user_id" property="userId" jdbcType="BIGINT" />
  6.         <result column="product_id" property="productId" jdbcType="BIGINT" />
  7.         <result column="count" property="count" jdbcType="INTEGER" />
  8.         <result column="money" property="money" jdbcType="DECIMAL" />
  9.         <result column="status" property="status" jdbcType="INTEGER" />
  10.     </resultMap>
  12.     <insert id="create" parameterType="cn.cvzhanshi.springcloud.alibaba.domain.Order"
  13.             useGeneratedKeys="true" keyProperty="id">
  14.         insert into t_order(`user_id`, `product_id`, `count`, `money`, `status`)
  15.         values(#{userId}, #{productId}, #{count}, #{money}, 0);
  16.     </insert>
  17.     <update id="update" parameterType="cn.cvzhanshi.springcloud.alibaba.domain.Order">
  18.         update t_order set `status` = 1
  19.         where `user_id` = #{userId} and `status` = #{status};
  20.     </update>
  21. </mapper>
  • service层的编写

AccountService

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 21:57
  4.  */
  5. @Service
  6. @FeignClient(value = "seata-account-service")
  7. public interface AccountService {
  10.     @PostMapping(value = "/account/decrease")
  11.     CommonResult decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money);
  12. }

StorageService

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 21:56
  4.  */
  5. @Service
  6. @FeignClient(value = "seata-storage-service")
  7. public interface StorageService {
  9.     //减库存
  10.     @PostMapping(value = "/storage/decrease")
  11.     CommonResult decrease(@RequestParam("productId") Long productId, @RequestParam("count") Integer count);
  12. }

OrderService

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 21:57
  4.  */
  5. public interface OrderService {
  7.     void create(Order order);
  8. }

OrderServiceImpl

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 22:04
  4.  */
  5. @Slf4j
  6. @Service
  7. public class OrderServiceImpl implements OrderService {
  10.     @Autowired
  11.     private OrderDao orderDao;
  13.     @Autowired
  14.     private AccountService accountService;
  16.     @Autowired
  17.     private StorageService storageService;
  19.     @Override
  20.     public void create(Order order) {
  23.         //1. 新建订单
  24.         log.info("-------> 开始新建订单");
  25.         orderDao.create(order);
  27.         //2. 扣减库存
  28.         log.info("-------> 订单微服务开始调用库存，做扣减count");
  29.         storageService.decrease(order.getProductId(), order.getCount());
  30.         log.info("-------> 订单微服务开始调用库存，做扣减完成");
  32.         //3. 扣减账号余额
  33.         log.info("-------> 订单微服务开始调用账号，做扣减money");
  34.         accountService.decrease(order.getUserId(), order.getMoney());
  35.         log.info("-------> 订单微服务开始调用账号，做扣减完成");
  37.         //4、修改状态
  38.         log.info("-------> 修改订单状态");
  39.         orderDao.update(order.getUserId(), 0);
  40.         log.info("-------> 修改订单状态完成");
  42.         log.info("-------> 新建订单完成");
  43.     }
  44. }

  • controller层的编写

    /**

    • @author cVzhanshi

    • @create 2021-07-12 22:56
      */
      @RestController
      public class OrderController {

      @Autowired
      private OrderService orderService;

      @GetMapping(“/order/create”)
      public CommonResult create(Order order){

      1.  orderService.create(order);
      2.  return new CommonResult(200, "订单创建成功!");

      }
      }

  • config配置类的编写

MybatisConfig

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 22:58
  4.  */
  5. @MapperScan("cn.cvzhanshi.springcloud.alibaba.dao")
  6. @Configuration
  7. public class MybatisConfig {
  9. }

DataSourceProxyConfig

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-12 22:58
  4.  */
  5. //使用Seata对数据源进行代理
  6. @Configuration
  7. public class DataSourceProxyConfig {
  10.     @Value("${mybatis.mapperLocations}")
  11.     private String mapperLocations;
  13.     @Bean
  14.     @ConfigurationProperties(prefix = "spring.datasource")
  15.     public DataSource druidDataSource() {
  17.         return new DruidDataSource();
  18.     }
  20.     @Bean
  21.     public DataSourceProxy dataSourceProxy(DataSource druidDataSource) {
  23.         return new DataSourceProxy(druidDataSource);
  24.     }
  26.     @Bean
  27.     public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception {
  29.         SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
  30.         bean.setDataSource(dataSourceProxy);
  31.         ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
  32.         bean.setMapperLocations(resolver.getResources(mapperLocations));
  33.         return bean.getObject();
  34.     }
  35. }

  • 主启动类

    /**

    • @author cVzhanshi

    • @create 2021-07-12 23:02
      */
      @SpringBootApplication(exclude = DataSourceAutoConfiguration.class) //取消数据源的自动创建
      @EnableDiscoveryClient
      @EnableFeignClients
      public class SeataOrderMain2001 {

      public static void main(String[] args) {

      1.  SpringApplication.run(SeataOrderMain2001.class,args);

      }
      }

启动测试

  • 出现错误

在这里插入图片描述

  • 出错原因

seata 版本问题， spring-cloud-alibaba 依赖下的seata-spring-boot-starter 版本是1.1.0，而我用的seata是 1.2.0 版本 。
1.2.0 版本SeataAutoConfiguration.class 没有seataDataSourceBeanPostProcessor bean ,而是用SeataAutoDataSourceProxyCreator代替

  • 解决方法

去掉spring-cloud-starter-alibaba-seata 下seata-spring-boot-starter 依赖

  1. <dependency>
  2.     <groupId>com.alibaba.cloud</groupId>
  3.     <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
  4.     <exclusions>
  5.         <exclusion>
  6.             <groupId>io.seata</groupId>
  7.             <artifactId>seata-all</artifactId>
  8.         </exclusion>
  9.         <exclusion>
  10.             <groupId>io.seata</groupId>
  11.             <artifactId>seata-spring-boot-starter</artifactId>
  12.         </exclusion>
  13.         <exclusion>
  14.             <artifactId>spring-boot-starter</artifactId>
  15.             <groupId>org.springframework.boot</groupId>
  16.         </exclusion>
  17.     </exclusions>
  18. </dependency>
  19. <dependency>
  20.     <groupId>io.seata</groupId>
  21.     <artifactId>seata-all</artifactId>
  22.     <version>1.2.0</version>
  23. </dependency>
  • 问题解决，继续测试

在这里插入图片描述

5.5.2 库存模块

  • 新建模块seata-storage-service2002
  • 导入依赖和2001一样直接拷贝

  • 编写配置文件

    server:
    port: 2002

    spring:
    application:

    1. name: seata-storage-service

    cloud:

    1. alibaba:
    2.   seata:
    3.     # 自定义事务组名称需要与seata-server中的对应
    4.     tx-service-group: my_test_tx_group #因为seata的file.conf文件中没有service模块，事务组名默认为my_test_tx_group
    5.     #service要与tx-service-group对齐，vgroupMapping和grouplist在service的下一级，my_test_tx_group在再下一级
    6.     service:
    7.       vgroupMapping:
    8.         #要和tx-service-group的值一致
    9.         my_test_tx_group: default
    10.       grouplist:
    11.         # seata seaver的 地址配置，此处可以集群配置是个数组
    12.         default: 192.168.1.104:8091
    13. nacos:
    14.   discovery:
    15.     server-addr: localhost:8848  #nacos

    datasource:

    1. # 当前数据源操作类型
    2. type: com.alibaba.druid.pool.DruidDataSource
    3. # mysql驱动类
    4. driver-class-name: com.mysql.jdbc.Driver
    5. url: jdbc:mysql://localhost:3306/seata_storage
    6. username: root
    7. password: lian0911

    feign:
    hystrix:

    1. enabled: false

    logging:
    level:

    1. io:
    2.   seata: info

    mybatis:
    mapperLocations: classpath:mapper/*.xml

  • file.conf 和registry.conf 与2001一样，直接拷贝

  • domain层实体类

CommonResult

  1. @Data
  2. @AllArgsConstructor
  3. @NoArgsConstructor
  4. public class CommonResult<T> {
  6.     private Integer code;
  7.     private String message;
  8.     private T data;
  10.     public CommonResult(Integer code, String message) {
  12.         this(code, message, null);
  13.     }
  14. }

Storage

  1. @Data
  2. @AllArgsConstructor
  3. @NoArgsConstructor
  4. public class Storage {
  6.     private Long id;
  7.     //产品id
  8.     private Long productId;
  9.     //总库存
  10.     private Integer total;
  11.     //已用库存
  12.     private Integer used;
  13.     //剩余库存
  14.     private Integer residue;
  15. }
  • dao层的编写

StorageDao

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-13 0:17
  4.  */
  5. @Mapper
  6. @Repository
  7. public interface StorageDao {
  9.     void decrease(@Param("productId") Long productId, @Param("count") Integer count);
  10. }

对应的映射文件classpath:mapper/StorageMapper.xml

  1. <mapper namespace="com.angenin.springcloud.dao.StorageDao">
  3.     <resultMap id="BaseResultMap" type="com.angenin.springcloud.domain.Storage">
  4.         <id column="id" property="id" jdbcType="BIGINT"/>
  5.         <result column="product_id" property="productId" jdbcType="BIGINT"/>
  6.         <result column="total" property="total" jdbcType="INTEGER"/>
  7.         <result column="used" property="used" jdbcType="INTEGER"/>
  8.         <result column="residue" property="residue" jdbcType="INTEGER"/>
  9.     </resultMap>
  11.     <update id="decrease">
  12.         update t_storage
  13.         set used = used + #{count}, residue = residue - #{count}
  14.         where product_id= #{productId};
  15.     </update>
  16. </mapper>
  • service层的编写

StorageService

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-13 0:22
  4.  */
  5. public interface StorageService {
  7.     /**
  8.      * 扣减库存
  9.      * @param productId
  10.      * @param count
  11.      */
  12.     void decrease(Long productId, Integer count);
  13. }

StorageServiceImpl

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-13 0:23
  4.  */
  5. @Service
  6. public class StorageServiceImpl implements StorageService {
  9.     private static final Logger LOGGER = LoggerFactory.getLogger(StorageServiceImpl.class);
  11.     @Resource
  12.     private StorageDao storageDao;
  14.     @Override
  15.     public void decrease(Long productId, Integer count) {
  17.         LOGGER.info("----> StorageService中扣减库存");
  18.         storageDao.decrease(productId, count);
  19.         LOGGER.info("----> StorageService中扣减库存完成");
  20.     }
  21. }
  • controller层的编写

StorageController

  1. @RestController
  2. public class StorageController {
  5.     @Resource
  6.     private StorageService storageService;
  8.     @RequestMapping("/storage/decrease")
  9.     public CommonResult decrease(@RequestParam("productId") Long productId, @RequestParam("count") Integer count){
  11.         storageService.decrease(productId, count);
  12.         return new CommonResult(200, "扣减库存成功!");
  13.     }
  15. }
  • 配置类 MyBatisConfig 和 DataSourceProxyConfig 与2001一样，直接copy

  • 主启动类

    /**

    • @author cVzhanshi
    • @create 2021-07-13 0:29
      */
      @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
      @EnableFeignClients
      @EnableDiscoveryClient
      public class SeataStorageMain2002 {
  1.     public static void main(String[] args) {
  3.         SpringApplication.run(SeataStorageMain2002.class,args);
  4.     }
  5. }
  • 启动测试

在这里插入图片描述

5.5.3 账户模块

  • 新建模块seata-account-service2003
  • 导入依赖和2001一样直接拷贝

  • 编写配置文件

    server:
    port: 2003

    spring:
    application:

    1. name: seata-account-service

    cloud:

    1. alibaba:
    2.   seata:
    3.     # 自定义事务组名称需要与seata-server中的对应
    4.     tx-service-group: my_test_tx_group #因为seata的file.conf文件中没有service模块，事务组名默认为my_test_tx_group
    5.     #service要与tx-service-group对齐，vgroupMapping和grouplist在service的下一级，my_test_tx_group在再下一级
    6.     service:
    7.       vgroupMapping:
    8.         #要和tx-service-group的值一致
    9.         my_test_tx_group: default
    10.       grouplist:
    11.         # seata seaver的 地址配置，此处可以集群配置是个数组
    12.         default: 192.168.1.104:8091
    13. nacos:
    14.   discovery:
    15.     server-addr: localhost:8848  #nacos

    datasource:

    1. # 当前数据源操作类型
    2. type: com.alibaba.druid.pool.DruidDataSource
    3. # mysql驱动类
    4. driver-class-name: com.mysql.jdbc.Driver
    5. url: jdbc:mysql://localhost:3306/seata_account
    6. username: root
    7. password: lian0911

    feign:
    hystrix:

    1. enabled: false

    logging:
    level:

    1. io:
    2.   seata: info

    mybatis:
    mapperLocations: classpath:mapper/*.xml

  • file.conf 和registry.conf 与2001一样，直接拷贝

  • domain层实体类

CommonResult

  1. @Data
  2. @AllArgsConstructor
  3. @NoArgsConstructor
  4. public class CommonResult<T> {
  6.     private Integer code;
  7.     private String message;
  8.     private T data;
  10.     public CommonResult(Integer code, String message) {
  12.         this(code, message, null);
  13.     }
  14. }

Account

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-13 0:37
  4.  */
  5. @Data
  6. @AllArgsConstructor
  7. @NoArgsConstructor
  8. public class Account {
  10.     private Long id;
  12.     /**
  13.      * 用户id
  14.      */
  15.     private Long userId;
  17.     /**
  18.      * 总额度
  19.      */
  20.     private Integer total;
  22.     /**
  23.      * 已用额度
  24.      */
  25.     private Integer used;
  27.     /**
  28.      * 剩余额度
  29.      */
  30.     private Integer residue;
  31. }
  • dao 层的编写

AccountDao

  1. @Mapper
  2. @Repository
  3. public interface AccountDao {
  5.     void decrease(@Param("userId") Long userId, @Param("money") BigDecimal money);
  6. }

对应的映射文件classpath:mapper/ AccountMapper.xml

  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
  3.         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
  4. <mapper namespace="cn.cvzhanshi.springcloud.alibaba.dao.AccountDao">
  6.     <resultMap id="BaseResultMap" type="cn.cvzhanshi.springcloud.alibaba.domain.Account">
  7.         <id column="id" property="id" jdbcType="BIGINT"/>
  8.         <result column="user_id" property="userId" jdbcType="BIGINT"/>
  9.         <result column="total" property="total" jdbcType="DECIMAL"/>
  10.         <result column="used" property="used" jdbcType="DECIMAL"/>
  11.         <result column="residue" property="residue" jdbcType="DECIMAL"/>
  12.     </resultMap>
  14.     <update id="decrease">
  15.         update t_account
  16.         set used = used + #{money}, residue = residue - #{money}
  17.         where user_id = #{userId};
  18.     </update>
  19. </mapper>
  • service层的编写

AccountService

  1. public interface AccountService {
  3.     void decrease(Long userId, BigDecimal money);
  4. }

AccountServiceImpl

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2021-07-13 0:43
  4.  */
  5. @Service
  6. public class AccountServiceImpl implements AccountService {
  9.     private static final Logger LOGGER = LoggerFactory.getLogger(AccountServiceImpl.class);
  11.     @Resource
  12.     private AccountDao accountDao;
  14.     @Override
  15.     public void decrease(Long userId, BigDecimal money) {
  17.         LOGGER.info("---> AccountService中扣减账户余额");
  18.         accountDao.decrease(userId, money);
  19.         LOGGER.info("---> AccountService中扣减账户余额完成");
  20.     }
  21. }

  • controller层的编写

    /**

    • @author cVzhanshi
    • @create 2021-07-13 0:44
      */
      @RestController
      public class AccountController {
  1.     @Autowired
  2.     private AccountService accountService;
  4.     @RequestMapping("/account/decrease")
  5.     public CommonResult decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money){
  7.         accountService.decrease(userId, money);
  8.         return new CommonResult(200, "扣减库存成功!");
  9.     }
  11. }

  • 主启动类

    /**

    • @author cVzhanshi
    • @create 2021-07-13 0:46
      */
      @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
      @EnableFeignClients
      @EnableDiscoveryClient
      public class SeataAccountMain2003 {
  1.     public static void main(String[] args) {
  3.         SpringApplication.run(SeataAccountMain2003.class,args);
  4.     }
  5. }
  • 启动测试

在这里插入图片描述

环境准备完毕

5.5.4 Test

  • 正常下单
  • 启动nacos、seata、2001、2002、2003
  • 访问 http://localhost:2001/order/create?userId=1&productId=1&count=10&money=10

在这里插入图片描述

  • 查看数据库中的数据变化

order表

在这里插入图片描述

account表

在这里插入图片描述

storage表

在这里插入图片描述

输入日志

在这里插入图片描述

  • 超时异常
  • 停止2003

  • 在2003的AccountServiceImpl里的decrease中添加异常代码

    @Override
    public void decrease(Long userId, BigDecimal money) {

    LOGGER.info(“—-> AccountService中扣减账户余额”);
    //模拟超时异常，暂停20秒
    try {

    1.  TimeUnit.SECONDS.sleep(20);

    } catch (InterruptedException e) {

    1.  e.printStackTrace();

    }
    accountDao.decrease(userId, money);
    LOGGER.info(“—-> AccountService中扣减账户余额完成”);
    }

  • 重新启动2003

  • 继续访问http://localhost:2001/order/create?userId=1&productId=1&count=10&money=10

在这里插入图片描述

  • 查看数据库

storage表
在这里插入图片描述

account表

在这里插入图片描述

order表

在这里插入图片描述

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

  • 超市异常，添加 @GlobalTransactional 注解
  • 停止2001

  • 在2001的OderServiceImpl里的create方法上加上 @GlobalTransactional

    @Override
    //name随便命名，只要不重复即可
    //rollbackFor = Exception.class表示出现所有异常都回滚
    //rollbackFor表示哪些需要回滚
    //noRollbackFor表示哪些不需要回滚
    @GlobalTransactional(name = “test-create-order”, rollbackFor = Exception.class)
    public void create(Order order) {

    1.  }

  • 重启2001。

  • 刷新页面http://localhost:2001/order/create?userId=1&productId=1&count=10&money=10

在这里插入图片描述

  • 查看数据库

order表

在这里插入图片描述

account表

在这里插入图片描述

storage表

在这里插入图片描述

order表没有插入，相应的账户表和库存表都没有改变，回滚成功

5.6 Seata的再理解

5.6.1 图解TC/TM/RM三组件

  • 官方图解

在这里插入图片描述

  • 自身理解

在这里插入图片描述

5.6.2 分布式事务的执行流程

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

模式分类

  • AT 模式

    提供无侵入自动补偿的事务模式，目前已支持 MySQL、 Oracle 、PostgreSQL和 TiDB的AT模式，H2 开发中

  • TCC 模式

    支持 TCC 模式并可与 AT 混用，灵活度更高

  • SAGA 模式

    为长事务提供有效的解决方案

AT模式详解

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

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

在一阶段, Seata 会拦截“业务SQL”,
1、解析SQL语义,找到“业务SQL”要更新的业务数据,在业务数据被更新前,将其保存成”before image”,
2、执行“业务SQL”更新业务数据，在业务数据更新之后,
3、其保存成”after image” ,最后生成行锁。
以上操作全部在一个数据库事务内完成，这样保证了一阶段操作的原子性。

在这里插入图片描述

  • 二阶段：

    • 提交异步化，非常快速地完成。

    二阶段如是顺利提交的话,
    因为“业务SQL”在一阶段已经提交至数据库,所以Seata框架只需将一阶段保存的快照数据和行锁删掉， 完成数据清理即可。

    在这里插入图片描述

    • 回滚通过一阶段的回滚日志进行反向补偿

    二阶段回滚: .
    二阶段如果是回滚的话, Seata就需要回滚一阶段已经执行的“业务SQL” ,还原业务数据。
    回滚方式便是用”before image”还原业务数据;但在还原前要首先要校验脏写,对比”数据库当前业务数据”和”after image”,如果两份数据完全一致就说明没有脏写， 可以还原业务数据，如果不一致就说明有脏写，出现脏写就需要转人工处理。

    在这里插入图片描述

简略流程图
在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，28人围观）

还没有评论，来说两句吧...

相关阅读