Seata TCC模式-TCC模式

转x33g5p2x  于2021-10-28 转载在 其他  
字(25.5k)|赞(0)|评价(0)|浏览(737)

项目源码： https://gitee.com/benwang6/seata-samples

一.TCC 基本原理

TCC 与 Seata AT 事务一样都是两阶段事务，它与 AT 事务的主要区别为：

TCC 对业务代码侵入严重
每个阶段的数据操作都要自己进行编码来实现，事务框架无法自动处理。
TCC 效率更高
不必对数据加全局锁，允许多个事务同时操作数据。

第一阶段 Try

以账户服务为例，当下订单时要扣减用户账户金额：

假如用户购买 100 元商品，要扣减 100 元。

TCC 事务首先对这100元的扣减金额进行预留，或者说是先冻结这100元：

第二阶段 Confirm

如果第一阶段能够顺利完成，那么说明“扣减金额”业务（分支事务）最终肯定是可以成功的。当全局事务提交时， TC会控制当前分支事务进行提交，如果提交失败，TC 会反复尝试，直到提交成功为止。

当全局事务提交时，就可以使用冻结的金额来最终实现业务数据操作：

第二阶段 Cancel

如果全局事务回滚，就把冻结的金额进行解冻，恢复到以前的状态，TC 会控制当前分支事务回滚，如果回滚失败，TC 会反复尝试，直到回滚完成为止。

多个事务并发的情况

多个TCC全局事务允许并发，它们执行扣减金额时，只需要冻结各自的金额即可：

Seata TCC事务模式

Seata 支持 TCC 事务模式，与 AT 模式相同的，也需要以下组件来支持全局事务的控制：

  • TC 事务协调器
  • TM 事务管理器
  • RM 资源管理器

下一节，我们还是以订单业务为例，来演示 Seata 如何实现 TCC 事务。

二.准备订单项目案例

项目源码：

1.新建工程，导入无事务版本

1. 新建empty工程 seata-tcc

工程命名为 seata-tcc，存放到 seata-samples 文件夹下，与 seata-at 工程存放在一起：

2. 导入订单项目，无事务版本

无事务版本的4个文件夹，解压缩到工程目录
1.下载项目代码
2.访问 git 仓库
访问项目标签

下载无事务版

1.解压到 seata-tcc 目录

压缩文件中的 7 个项目目录解压缩到 seata-tcc 目录：

2. 导入项目
  1. - project structure 中添加moudle
  2. 	- double shift搜索add maven project

然后选择 seata-tcc 工程目录下的 7 个项目的 pom.xml 导入：

拖拽pom.xml文件，右键add as maven project

3.order启动全局事务，添加“保存订单”分支事务

在订单项目中执行添加订单：

我们要添加以下 TCC 事务操作的代码：

  1. Try - 第一阶，冻结数据阶段，向订单表直接插入订单，订单状态设置为0（冻结状态）。

Confirm - 第二阶段，提交事务，将订单状态修改成1（正常状态）。

Cancel - 第二阶段，回滚事务，删除订单。

1.order-parent 添加 seata 依赖,

打开 order-parent 中注释掉的 seata 依赖：

  1. <?xml version="1.0" encoding="UTF-8"?>
  3. <project xmlns="http://maven.apache.org/POM/4.0.0" 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.     <modelVersion>4.0.0</modelVersion>
  6.     <parent>
  7.         <groupId>org.springframework.boot</groupId>
  8.         <artifactId>spring-boot-starter-parent</artifactId>
  9.         <version>2.3.2.RELEASE</version>
  10.         <relativePath/> <!-- lookup parent from repository -->
  11.     </parent>
  12.     <groupId>cn.tedu</groupId>
  13.     <artifactId>order-parent</artifactId>
  14.     <version>1.0-SNAPSHOT</version>
  15.     <packaging>pom</packaging>
  16.     <name>order-parent</name>
  17.     <modules>
  18.         <module>account</module>
  19.         <module>storage</module>
  20.         <module>order</module>
  21.     </modules>
  23.     <properties>
  24.         <mybatis-plus.version>3.3.2</mybatis-plus.version>
  25.         <druid-spring-boot-starter.version>1.1.23</druid-spring-boot-starter.version>
  26.         <seata.version>1.3.0</seata.version>
  27.         <spring-cloud-alibaba-seata.version>2.0.0.RELEASE</spring-cloud-alibaba-seata.version>
  28.         <spring-cloud.version>Hoxton.SR12</spring-cloud.version>
  29.     </properties>
  31.     <dependencies>
  32.         <dependency>
  33.             <groupId>org.springframework.boot</groupId>
  34.             <artifactId>spring-boot-starter</artifactId>
  35.         </dependency>
  36.         <dependency>
  37.             <groupId>org.springframework.boot</groupId>
  38.             <artifactId>spring-boot-starter-web</artifactId>
  39.         </dependency>
  40.         <dependency>
  41.             <groupId>org.springframework.cloud</groupId>
  42.             <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
  43.         </dependency>
  44.         <dependency>
  45.             <groupId>org.springframework.cloud</groupId>
  46.             <artifactId>spring-cloud-starter-openfeign</artifactId>
  47.         </dependency>
  49.         <dependency>
  50.             <groupId>com.baomidou</groupId>
  51.             <artifactId>mybatis-plus-boot-starter</artifactId>
  52.             <version>${mybatis-plus.version}</version>
  53.         </dependency>
  55.         <dependency>
  56.             <groupId>mysql</groupId>
  57.             <artifactId>mysql-connector-java</artifactId>
  58.         </dependency>
  59.         <dependency>
  60.             <groupId>com.alibaba</groupId>
  61.             <artifactId>druid-spring-boot-starter</artifactId>
  62.             <version>${druid-spring-boot-starter.version}</version>
  63.         </dependency>
  65.         <dependency>
  66.           <groupId>com.alibaba.cloud</groupId>
  67.           <artifactId>spring-cloud-alibaba-seata</artifactId>
  68.           <version>${spring-cloud-alibaba-seata.version}</version>
  69.           <exclusions>
  70.             <exclusion>
  71.               <artifactId>seata-all</artifactId>
  72.               <groupId>io.seata</groupId>
  73.             </exclusion>
  74.           </exclusions>
  75.         </dependency>
  76.         <dependency>
  77.           <groupId>io.seata</groupId>
  78.           <artifactId>seata-all</artifactId>
  79.           <version>${seata.version}</version>
  80.         </dependency>
  82.         <dependency>
  83.             <groupId>org.projectlombok</groupId>
  84.             <artifactId>lombok</artifactId>
  85.         </dependency>
  86.         <dependency>
  87.             <groupId>org.springframework.boot</groupId>
  88.             <artifactId>spring-boot-starter-test</artifactId>
  89.             <scope>test</scope>
  90.             <exclusions>
  91.                 <exclusion>
  92.                     <groupId>org.junit.vintage</groupId>
  93.                     <artifactId>junit-vintage-engine</artifactId>
  94.                 </exclusion>
  95.             </exclusions>
  96.         </dependency>
  97.     </dependencies>
  99.     <dependencyManagement>
  100.         <dependencies>
  101.             <dependency>
  102.                 <groupId>org.springframework.cloud</groupId>
  103.                 <artifactId>spring-cloud-dependencies</artifactId>
  104.                 <version>${spring-cloud.version}</version>
  105.                 <type>pom</type>
  106.                 <scope>import</scope>
  107.             </dependency>
  108.         </dependencies>
  109.     </dependencyManagement>
  111.     <build>
  112.         <plugins>
  113.             <plugin>
  114.                 <groupId>org.springframework.boot</groupId>
  115.                 <artifactId>spring-boot-maven-plugin</artifactId>
  116.             </plugin>
  117.         </plugins>
  118.     </build>
  120. </project>
2. 三个配置文件
1.application.yml --事务组
  1. spring:
  2.   application:
  3.     name: order
  5.   datasource:
  6.     driver-class-name: com.mysql.cj.jdbc.Driver
  7.     url: jdbc:mysql://localhost/seata_order?useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8
  8.     username: root
  9.     password: root
  10.   cloud:
  11.     alibaba:
  12.       seata:
  13.         tx-service-group: order_tx_group
  15. server:
  16.   port: 8083
  18. eureka:
  19.   client:
  20.     service-url:
  21.       defaultZone: http://localhost:8761/eureka
  22.   instance:
  23.     prefer-ip-address: true
  24.     instance-id: ${spring.cloud.client.ip-address}:${spring.application.name}:${server.port}
  26. mybatis-plus:
  27.   type-aliases-package: cn.tedu.order.entity
  28.   mapper-locations:
  29.     - classpath:/mapper/*Mapper.xml configuration: map-underscore-to-camel-case: true logging: level: cn.tedu.order.mapper: DEBUG ribbon: MaxAutoRetriesNextServer: 0 #默认1
2. registry.conf – 注册中心的地址
  1. registry {
  2.   # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  3.   type = "eureka"
  5.   nacos {
  6.     serverAddr = "localhost"
  7.     namespace = ""
  8.     cluster = "default"
  9.   }
  10.   eureka {
  11.     # 连接eureka，要从注册表发现 seata-server
  12.     serviceUrl = "http://localhost:8761/eureka"
  13.     # application = "default"
  14.     # weight = "1"
  15.   }
  16.   redis {
  17.     serverAddr = "localhost:6379"
  18.     db = "0"
  19.     password = ""
  20.     cluster = "default"
  21.     timeout = "0"
  22.   }
  23.   zk {
  24.     cluster = "default"
  25.     serverAddr = "127.0.0.1:2181"
  26.     session.timeout = 6000
  27.     connect.timeout = 2000
  28.     username = ""
  29.     password = ""
  30.   }
  31.   consul {
  32.     cluster = "default"
  33.     serverAddr = "127.0.0.1:8500"
  34.   }
  35.   etcd3 {
  36.     cluster = "default"
  37.     serverAddr = "http://localhost:2379"
  38.   }
  39.   sofa {
  40.     serverAddr = "127.0.0.1:9603"
  41.     application = "default"
  42.     region = "DEFAULT_ZONE"
  43.     datacenter = "DefaultDataCenter"
  44.     cluster = "default"
  45.     group = "SEATA_GROUP"
  46.     addressWaitTime = "3000"
  47.   }
  48.   file {
  49.     name = "file.conf"
  50.   }
  51. }
  53. config {
  54.   # file、nacos 、apollo、zk、consul、etcd3、springCloudConfig
  55.   type = "file"
  57.   nacos {
  58.     serverAddr = "localhost"
  59.     namespace = ""
  60.     group = "SEATA_GROUP"
  61.   }
  62.   consul {
  63.     serverAddr = "127.0.0.1:8500"
  64.   }
  65.   apollo {
  66.     app.id = "seata-server"
  67.     apollo.meta = "http://192.168.1.204:8801"
  68.     namespace = "application"
  69.   }
  70.   zk {
  71.     serverAddr = "127.0.0.1:2181"
  72.     session.timeout = 6000
  73.     connect.timeout = 2000
  74.     username = ""
  75.     password = ""
  76.   }
  77.   etcd3 {
  78.     serverAddr = "http://localhost:2379"
  79.   }
  80.   file {
  81.     name = "file.conf"
  82.   }
  83. }
3.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.   # 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,will not be used for UDT
  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.   # order_tx_group 与 yml 中的 “tx-service-group: order_tx_group” 配置一致
  34.   # “seata-server” 与 TC 服务器的注册名一致
  35.   # 从eureka获取seata-server的地址，再向seata-server注册自己，设置group
  38.   # order_tx_group 事务组，对应使用哪个协调器
  39.   # seata-server 是注册表中的服务id
  40.   vgroupMapping.order_tx_group = "seata-server"
  44.   #only support when registry.type=file, please don't set multiple addresses
  45.   order_tx_group.grouplist = "127.0.0.1:8091"
  46.   #degrade, current not support
  47.   enableDegrade = false
  48.   #disable seata
  49.   disableGlobalTransaction = false
  50. }
  52. client {
  53.   rm {
  54.     asyncCommitBufferLimit = 10000
  55.     lock {
  56.       retryInterval = 10
  57.       retryTimes = 30
  58.       retryPolicyBranchRollbackOnConflict = true
  59.     }
  60.     reportRetryCount = 5
  61.     tableMetaCheckEnable = false
  62.     reportSuccessEnable = false
  63.   }
  64.   tm {
  65.     commitRetryCount = 5
  66.     rollbackRetryCount = 5
  67.   }
  68.   undo {
  69.     dataValidation = true
  70.     logSerialization = "jackson"
  71.     logTable = "undo_log"
  72.   }
  73.   log {
  74.     exceptionRate = 100
  75.   }
  76. }
3. OrderMapper 添加更新订单状态、删除订单

根据前面的分析，订单数据操作有以下三项：

  • 插入订单
  • 修改订单状态
  • 删除订单

在 OrderMapper 中已经有插入订单的方法了，现在需要添加修改订单和删除订单的方法（删除方法从BaseMapper继承）：

  1. package cn.tedu.order.mapper;
  3. import cn.tedu.order.entity.Order;
  4. import com.baomidou.mybatisplus.core.mapper.BaseMapper;
  6. public interface  OrderMapper extends BaseMapper<Order> {
  7.     //创建正常订单
  8.     void create(Order order);
  10.     //创建冻结订单
  11.     void createFrozen(Order order);
  12.     //修订订单状态
  13.     void updateStatus(Long orderId,Integer status);
  14.     //取消回滚 删除订单，使用继承的方法deleteById()
  16. }

那么对应的 OrderMapper.xml 中也要添加 sql：

  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="cn.tedu.order.mapper.OrderMapper" >
  4.     <resultMap id="BaseResultMap" type="cn.tedu.order.entity.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.     <!--关键字必须加反引号-->
  13.     <insert id="create">
  14.         INSERT INTO `order` (`id`,`user_id`,`product_id`,`count`,`money`,`status`)
  15.         VALUES(#{id}, #{userId}, #{productId}, #{count}, #{money},1);
  16.     </insert>
  17.     <!--创建冻结订单方法-->
  18.     <insert id="createFrozen">
  19.         INSERT INTO `order` (`id`,`user_id`,`product_id`,`count`,`money`,`status`)
  20.         VALUES(#{id}, #{userId}, #{productId}, #{count}, #{money},0);
  21.     </insert>
  22.     <!--修改订单状态-->
  23.     <update id="updateStatus">
  24.         update `order` set  status = #{status}
  25.         where id = #{orderId};
  26.     </update>
  27.     <!--删除订单-->
  28.     <delete id="deleteById">
  29.         delete from `order` where id=#{orderId}
  30.     </delete>
  31. </mapper>
4. 按照seata tcc 的实现规则，定义TccAction接口和实现
1.添加三个方法，实现TCC三个操作

第一阶段
Try - prepareCreateOrder()
第二阶段
Confirm - commit()
Cancel - rollback()

  1. package cn.tedu.order.tcc;
  3. import io.seata.rm.tcc.api.BusinessActionContext;
  4. import io.seata.rm.tcc.api.BusinessActionContextParameter;
  5. import io.seata.rm.tcc.api.LocalTCC;
  6. import io.seata.rm.tcc.api.TwoPhaseBusinessAction;
  8. import java.math.BigDecimal;
  10. @LocalTCC
  11. public interface OrderTccAction {
  12.     /* 为了避开seata的一个bug，不使用封装对象 而是一个个的单独传递参数 */
  13.     @TwoPhaseBusinessAction(name = "OrderTccAction")
  14.     boolean prepare(BusinessActionContext ctx,
  15.                     @BusinessActionContextParameter(paramName = "orderId") Long id,
  16.                     Long userId,
  17.                     Long productId,
  18.                     Integer count,
  19.                     BigDecimal money);
  21.     boolean commit(BusinessActionContext ctx);
  22.     boolean rollback(BusinessActionContext ctx);
  24. }

实现类：

  1. package cn.tedu.order.tcc;
  3. import cn.tedu.order.entity.Order;
  4. import cn.tedu.order.mapper.OrderMapper;
  5. import io.seata.rm.tcc.api.BusinessActionContext;
  6. import org.checkerframework.checker.units.qual.A;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  9. import org.springframework.transaction.annotation.Transactional;
  11. import java.math.BigDecimal;
  13. @Component
  14. public class OrderTccActionImpl implements OrderTccAction{
  15.     @Autowired
  16.     private OrderMapper orderMapper;
  18.     @Transactional
  19.     @Override
  20.     public boolean prepare(BusinessActionContext ctx, Long id, Long userId, Long productId,Integer count, BigDecimal money) {
  21.         orderMapper.createFrozen(new Order(id,userId,productId,count,money,0));
  22.         //true表示成功
  23.         //false表示失败
  24.         return true;
  25.     }
  26.     @Transactional
  27.     @Override
  28.     public boolean commit(BusinessActionContext ctx) {
  29.         //
  30.         Long orderId = Long.valueOf(ctx.getActionContext("orderId").toString());
  31.         orderMapper.updateStatus(orderId, 1);
  32.         return true;
  33.     }
  34.     @Transactional
  35.     @Override
  36.     public boolean rollback(BusinessActionContext ctx) {
  37.         Long orderId = Long.valueOf(ctx.getActionContext("orderId").toString());
  38.         orderMapper.deleteById(orderId);
  39.         return true;
  40.     }
  41. }
2. 三个方法都添加@Transactional注解，控制本地事务
  1. package cn.tedu.order.tcc;
  3. import cn.tedu.order.entity.Order;
  4. import cn.tedu.order.mapper.OrderMapper;
  5. import io.seata.rm.tcc.api.BusinessActionContext;
  6. import org.checkerframework.checker.units.qual.A;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  9. import org.springframework.transaction.annotation.Transactional;
  11. import java.math.BigDecimal;
  13. @Component
  14. public class OrderTccActionImpl implements OrderTccAction{
  15.     @Autowired
  16.     private OrderMapper orderMapper;
  18.     @Transactional
  19.     @Override
  20.     public boolean prepare(BusinessActionContext ctx, Long id, Long userId, Long productId,Integer count, BigDecimal money) {
  21.         orderMapper.createFrozen(new Order(id,userId,productId,count,money,0));
  22.         //true表示成功
  23.         //false表示失败
  24.         return true;
  25.     }
  26.     @Transactional
  27.     @Override
  28.     public boolean commit(BusinessActionContext ctx) {
  29.         //
  30.         Long orderId = Long.valueOf(ctx.getActionContext("orderId").toString());
  31.         orderMapper.updateStatus(orderId, 1);
  32.         return true;
  33.     }
  34.     @Transactional
  35.     @Override
  36.     public boolean rollback(BusinessActionContext ctx) {
  37.         Long orderId = Long.valueOf(ctx.getActionContext("orderId").toString());
  38.         orderMapper.deleteById(orderId);
  39.         return true;
  40.     }
  41. }
5.在业务代码中调用 Try 阶段方法

业务代码中不再直接保存订单数据，而是调用 TCC 第一阶段方法prepareCreateOrder()，并添加全局事务注解 @GlobalTransactional：

  1. package cn.tedu.order.service;
  3. import cn.tedu.order.entity.Order;
  4. import cn.tedu.order.fegin.AccountClient;
  5. import cn.tedu.order.fegin.EasyIdClient;
  6. import cn.tedu.order.fegin.StorageClient;
  7. import cn.tedu.order.mapper.OrderMapper;
  8. import cn.tedu.order.tcc.OrderTccActionImpl;
  9. import io.seata.spring.annotation.GlobalTransactional;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.stereotype.Service;
  13. import java.util.Random;
  15. @Service
  16. public class OrderServiceImpl implements OrderService{
  17.     @Autowired
  18.     private OrderMapper orderMapper;
  19.     @Autowired
  20.     private EasyIdClient easyIdClient;
  21.     @Autowired
  22.     private AccountClient accountClient;
  23.     @Autowired
  24.     private StorageClient storageClient;
  25.     @Autowired
  26.     private OrderTccActionImpl tcc;
  28.     @GlobalTransactional //启动全局事务
  29.     @Override
  30.     public void create(Order order) {
  31.         //远程调用发号器，生成订单id
  32.         String s = easyIdClient.nextId("order_business");
  33.         Long id = Long.valueOf(s);
  35.         order.setId(id);
  36.         /*//先临时随机产生id，加了发号器后，这行代码删除 Long id = Math.abs(new Random().nextLong());*/
  38.         /* tcc是一个动态代理对象，不是原始的对象 用AOP切入了代码，会拦截调用，新建上下文对象并传入目标方法 */
  39.         //冻结订单
  40.         tcc.prepare(
  41.                 null,
  42.                 order.getId(),
  43.                 order.getUserId(),
  44.                 order.getProductId(),
  45.                 order.getCount(),
  46.                 order.getMoney()
  47.                 );
  49.         //orderMapper.create(order);//创建正常订单
  50.         // TODO:远程调用库存，减少库存
  51.         storageClient.decrease(order.getProductId(),order.getCount());
  53.         // TODO：远程调用账号，扣减账户
  54.         //accountClient.decrease(order.getUserId(),order.getMoney());
  55.     }
  56. }
6.启动 order 进行测试

按顺序启动服务：

  1. Eureka
  2. Seata Server
  3. Easy Id Generator
  4. Order

调用保存订单，地址：

观察控制台日志：

查看数据库表中的订单数据：

4.storage添加“减少库存”分支事务

在库存项目中执行减少库存：

我们要添加以下 TCC 事务操作的代码：

  • Try - 第一阶段，冻结数据阶段，将要减少的库存量先冻结：

Confirm - 第二阶段，提交事务，使用冻结的库存完成业务数据处理：

Cancel - 第二阶段，回滚事务，冻结的库存解冻，恢复以前的库存量：

1.配置

有三个文件需要配置：

1.application.yml
  1. spring:
  2.   application:
  3.     name: storage
  4.   datasource:
  5.     driver-class-name: com.mysql.cj.jdbc.Driver
  6.     url: jdbc:mysql://localhost/seata_storage?useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8
  7.     username: root
  8.     password: root
  9.   cloud:
  10.     alibaba:
  11.       seata:
  12.         tx-service-group: order_tx_group
  13. server:
  14.   port: 8082
  16. eureka:
  17.   client:
  18.     service-url:
  19.       defaultZone: http://localhost:8761/eureka
  20.   instance:
  21.     prefer-ip-address: true
  22.     instance-id: ${spring.cloud.client.ip-address}:${spring.application.name}:${server.port}
  24. mybatis-plus:
  25.   type-aliases-package: cn.tedu.storage.entity
  26.   mapper-locations:
  27.     - classpath:/mapper/*Mapper.xml
  28.   configuration:
  29.     map-underscore-to-camel-case: true
  31. logging:
  32.   level:
  33.     cn.tedu.storage.mapper: DEBUG
2.registry.conf
  1. registry {
  2.   # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  3.   type = "eureka"
  5.   nacos {
  6.     serverAddr = "localhost"
  7.     namespace = ""
  8.     cluster = "default"
  9.   }
  10.   eureka {
  11.     # 连接eureka，要从注册表发现 seata-server
  12.     serviceUrl = "http://localhost:8761/eureka"
  13.     # application = "default"
  14.     # weight = "1"
  15.   }
  16.   redis {
  17.     serverAddr = "localhost:6379"
  18.     db = "0"
  19.     password = ""
  20.     cluster = "default"
  21.     timeout = "0"
  22.   }
  23.   zk {
  24.     cluster = "default"
  25.     serverAddr = "127.0.0.1:2181"
  26.     session.timeout = 6000
  27.     connect.timeout = 2000
  28.     username = ""
  29.     password = ""
  30.   }
  31.   consul {
  32.     cluster = "default"
  33.     serverAddr = "127.0.0.1:8500"
  34.   }
  35.   etcd3 {
  36.     cluster = "default"
  37.     serverAddr = "http://localhost:2379"
  38.   }
  39.   sofa {
  40.     serverAddr = "127.0.0.1:9603"
  41.     application = "default"
  42.     region = "DEFAULT_ZONE"
  43.     datacenter = "DefaultDataCenter"
  44.     cluster = "default"
  45.     group = "SEATA_GROUP"
  46.     addressWaitTime = "3000"
  47.   }
  48.   file {
  49.     name = "file.conf"
  50.   }
  51. }
  53. config {
  54.   # file、nacos 、apollo、zk、consul、etcd3、springCloudConfig
  55.   type = "file"
  57.   nacos {
  58.     serverAddr = "localhost"
  59.     namespace = ""
  60.     group = "SEATA_GROUP"
  61.   }
  62.   consul {
  63.     serverAddr = "127.0.0.1:8500"
  64.   }
  65.   apollo {
  66.     app.id = "seata-server"
  67.     apollo.meta = "http://192.168.1.204:8801"
  68.     namespace = "application"
  69.   }
  70.   zk {
  71.     serverAddr = "127.0.0.1:2181"
  72.     session.timeout = 6000
  73.     connect.timeout = 2000
  74.     username = ""
  75.     password = ""
  76.   }
  77.   etcd3 {
  78.     serverAddr = "http://localhost:2379"
  79.   }
  80.   file {
  81.     name = "file.conf"
  82.   }
  83. }
3.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.   # 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,will not be used for UDT
  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.   # order_tx_group 与 yml 中的 “tx-service-group: order_tx_group” 配置一致
  34.   # “seata-server” 与 TC 服务器的注册名一致
  35.   # 从eureka获取seata-server的地址，再向seata-server注册自己，设置group
  38.   # order_tx_group 事务组，对应使用哪个协调器
  39.   # seata-server 是注册表中的服务id
  40.   vgroupMapping.order_tx_group = "seata-server"
  44.   #only support when registry.type=file, please don't set multiple addresses
  45.   order_tx_group.grouplist = "127.0.0.1:8091"
  46.   #degrade, current not support
  47.   enableDegrade = false
  48.   #disable seata
  49.   disableGlobalTransaction = false
  50. }
  52. client {
  53.   rm {
  54.     asyncCommitBufferLimit = 10000
  55.     lock {
  56.       retryInterval = 10
  57.       retryTimes = 30
  58.       retryPolicyBranchRollbackOnConflict = true
  59.     }
  60.     reportRetryCount = 5
  61.     tableMetaCheckEnable = false
  62.     reportSuccessEnable = false
  63.   }
  64.   tm {
  65.     commitRetryCount = 5
  66.     rollbackRetryCount = 5
  67.   }
  68.   undo {
  69.     dataValidation = true
  70.     logSerialization = "jackson"
  71.     logTable = "undo_log"
  72.   }
  73.   log {
  74.     exceptionRate = 100
  75.   }
  76. }

这三个文件的设置与上面 order 项目的配置完全相同，请参考上面订单配置一章进行配置。

2.StorageMapper 添加冻结库存相关方法

根据前面的分析，库存数据操作有以下三项：

  • 冻结库存
  • 冻结库存量修改为已售出量
  • 解冻库存

在 StorageMapper 中添加三个方法：

  1. package cn.tedu.storage.mapper;
  3. import cn.tedu.storage.entity.Storage;
  4. import com.baomidou.mybatisplus.core.mapper.BaseMapper;
  6. public interface StorageMapper extends BaseMapper<Storage> {
  7.     //减少库存
  8.     void decrease(Long productId,Integer count);
  10.     //查询库存，用来判断有没有足够的库存
  11.     Storage findByProductId(Long productId);
  12.     //冻结库存
  13.     void updateResidueToFrozen(Long productId,Integer count);
  14.     //冻结 --> 已售出
  15.     void updateFrozenToUsed(Long productId,Integer count);
  16.     //冻结 --> 可用
  17.     void updateFrozenToResidue(Long productId,Integer count);
  19. }

对应的 StorageMapper.xml 中也要添加 sql：

  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="cn.tedu.storage.mapper.StorageMapper" >
  4.     <resultMap id="BaseResultMap" type="cn.tedu.storage.entity.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 storage
  13.       SET used = used + #{count},
  14.           residue = residue - #{count}
  15.       WHERE product_id = #{productId}
  16.     </update>
  17.     <!--查询库存-->
  18.     <select id="findByProductId" resultMap="BaseResultMap">
  19.         select * from storage where product_id = #{productId}
  20.     </select>
  21.     <!--冻结库存-->
  22.     <update id="updateResidueToFrozen">
  23.         update storage
  24.         set Residue=Residue-#{count},Frozen=Frozen+#{count}
  25.         where product_id = #{productId}
  26.     </update>
  27.     <!--冻结 &ndash;&gt; 已售出-->
  28.     <update id="updateFrozenToUsed">
  29.         update storage
  30.         set Frozen=Frozen-#{count},Used=Used+#{count}
  31.         where product_id = #{productId}
  32.     </update>
  33.     <!--冻结 &ndash;&gt; 可用-->
  34.     <update id="updateFrozenToResidue">
  35.         update storage
  36.         set Frozen=Frozen-#{count},Residue=Residue-#{count}
  37.         where product_id = #{productId}
  38.     </update>
  39. </mapper>
3.添加 TCC 接口，在接口中添加以下方法：

Try - prepareDecreaseStorage()
Confirm - commit()
Cancel - rollback()

  1. package cn.tedu.storage.tcc;
  3. import io.seata.rm.tcc.api.BusinessActionContext;
  4. import io.seata.rm.tcc.api.BusinessActionContextParameter;
  5. import io.seata.rm.tcc.api.LocalTCC;
  6. import io.seata.rm.tcc.api.TwoPhaseBusinessAction;
  8. /* 1.@LocalTCC 2.添加三个方法，都添加BusssinessActionContext 参数 3.T方法添加@TwoPhaseBusinessAction 4.T方法添加业务数据参数 5.需要添加到上下文对象的参数，添加@BussinessActionContextParameter */
  9. @LocalTCC
  10. public interface StorageTccAction {
  11.     @TwoPhaseBusinessAction(name = "StorageTccAction")
  12.     boolean prepare(BusinessActionContext ctx,
  13.                     @BusinessActionContextParameter(paramName = "productId") Long productId,
  14.                     @BusinessActionContextParameter(paramName = "count") Integer count);
  15.     boolean commit(BusinessActionContext ctx);
  16.     boolean rollback(BusinessActionContext ctx);
  17. }

实现类：

  1. package cn.tedu.storage.tcc;
  3. import cn.tedu.storage.entity.Storage;
  4. import cn.tedu.storage.mapper.StorageMapper;
  5. import io.seata.rm.tcc.api.BusinessActionContext;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Component;
  8. import org.springframework.transaction.annotation.Transactional;
  9. import rx.Producer;
  11. @Component
  12. public class StorageTccActionImpl implements StorageTccAction {
  13.     @Autowired
  14.     private StorageMapper storageMapper;
  16.     @Transactional
  17.     @Override
  18.     public boolean prepare(BusinessActionContext ctx, Long productId, Integer count) {
  19.         Storage storage = storageMapper.findByProductId(productId);
  20.         if (storage.getResidue() < count){
  21.             throw new RuntimeException("库存不足");
  22.         }
  23.         storageMapper.updateResidueToFrozen(productId, count);
  24.         return  true;
  25.     }
  26.     @Transactional
  27.     @Override
  28.     public boolean commit(BusinessActionContext ctx) {
  29.         Long productId = Long.valueOf(ctx.getActionContext("productId").toString());
  30.         Integer count = Integer.valueOf(ctx.getActionContext("count").toString());
  31.         storageMapper.updateFrozenToUsed(productId, count);
  32.         if (Math.random()<0.8){
  33.             return false;
  34.         }
  35.         return true;
  36.     }
  37.     @Transactional
  38.     @Override
  39.     public boolean rollback(BusinessActionContext ctx) {
  40.         Long productId = Long.valueOf(ctx.getActionContext("productId").toString());
  41.         Integer count = Integer.valueOf(ctx.getActionContext("count").toString());
  42.         storageMapper.updateFrozenToResidue(productId, count);
  43.         return true;
  44.     }
  45. }
4.在业务代码中调用 Try 阶段方法

业务代码中调用 TCC 第一阶段方法prepareDecreaseStorage()，并添加全局事务注解 @GlobalTransactional：

  1. package cn.tedu.storage.service;
  3. import cn.tedu.storage.mapper.StorageMapper;
  4. import cn.tedu.storage.tcc.StorageTccAction;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.stereotype.Service;
  8. @Service
  9. public class StorageServiceImpl implements StorageService{
  10.     @Autowired
  11.     private StorageTccAction tcc;
  12.     @Override
  13.     public void decrease(Long productId, Integer count) {
  14.         tcc.prepare(null, productId, count);
  15.     }
  16. }
5.启动 storage 进行测试

按顺序启动服务：

  1. Eureka
  2. Seata Server
  3. Easy Id Generator
  4. Storage
  5. Order
    调用保存订单，地址：

观察 storage 的控制台日志：

查看数据库表中的库存数据：

5.account添加“扣减金额”分支事务

扣减金额 TCC 事务分析请见Seata TCC模式-TCC模式介绍
配置
有三个文件需要配置：

application.yml
registry.conf
file.conf
这三个文件的设置与上面 order 项目的配置完全相同，请参考上面订单配置一章进行配置。

1.AccountMapper 添加冻结库存相关方法

根据前面的分析，库存数据操作有以下三项：

  • 冻结库存
  • 冻结库存量修改为已售出量
  • 解冻库存

在 AccountMapper 中添加三个方法：

添加 TCC 接口，在接口中添加以下方法：

Try - prepareDecreaseAccount()
Confirm - commit()
Cancel - rollback()

实现类：

在业务代码中调用 Try 阶段方法
业务代码中调用 TCC 第一阶段方法prepareDecreaseAccount()，并添加全局事务注解 @GlobalTransactional：

启动 account 进行测试
按顺序启动服务：

Eureka
Seata Server
Easy Id Generator
Storage
Account
Order
调用保存订单，地址：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

观察 account 的控制台日志：

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_53244569/article/details/121012269

内容来源于网络，如有侵权，请联系作者删除！

Javaspring cloudseata

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    目录

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com