手把手教你Spring Cloud集成Seata TCC模式(上)

简介: 手把手教你Spring Cloud集成Seata TCC模式

前言

在博客Spring Cloud集成分布式事务框架Seata 1.5.2中,我们已经集成了Seata AT模式,虽然AT模式可以覆盖大部分分布式事务需求,但是针对于一些追求高性能的业务场景,我们还是需要选择TCC模式;

因为TCC的资源预留概念降低了锁的粒度,在分布式事务未完成前并不会阻塞同业务下的其他分布式事务的执行;但是有一点不好的就是:TCC模式对于业务侵入性比较大,整个分布式事务的资源准备提交回滚逻辑全部需要开发人员自己完成,开发工作量比AT模式大出两三倍;

为了在将来的工作中能够顺利地使用TCC模式来作为高性能业务的分布式事务解决方案,我们下面就开始手把手教大家如何在Spring Cloud中集成Seata TCC模式。

业务场景

ce1f9b3582394c5990e9baff91a00970_tplv-k3u1fbpfcp-zoom-in-crop-mark_4536_0_0_0.png

同样还是购物车下单的业务场景:

1.用户通过业务入口提交下单数据;

2.先计算订单总金额并调用RPC进行预扣款;

  • 2.1:预扣款成功才能创建订单;
  • 2.2:预扣款失败,TM发起回滚,Account服务调用回滚逻辑,分布式事务回滚结束;

3.预扣款成功,那么再发起RPC创建订单;

  • 3.1:Order服务接收到创建订单的请求,先RPC执行预扣库存的操作;
  • 3.2:Storage服务预扣库存成功,Order服务执行预创建订单操作;
  • 3.3:Storage服务预扣库存失败,导致订单创建失败,下单业务抛出异常,TM发起回滚,Account服务与Storage服务调用回滚逻辑,分布式事务回滚结束;
  • 3.4:Order服务执行预创建订单成功,下单业务执行完毕,TM发起提交,所有RM调用提交逻辑,分布式事务成功;
  • 3.5:Order服务执行预创建订单失败,下单业务抛出异常,TM发起回滚,所有RM调用回滚逻辑,分布式事务回滚结束;

数据表创建

注意:TCC模式不需要创建undolog表

账户表:wallet_tcc_tbl

-- ----------------------------
-- Table structure for wallet_tcc_tbl
-- ----------------------------
DROP TABLE IF EXISTS `wallet_tcc_tbl`;
CREATE TABLE `wallet_tcc_tbl` (
  `id` int NOT NULL COMMENT '主键ID',
  `user_id` varchar(255) COLLATE utf8mb4_bin NOT NULL COMMENT '用户ID',
  `money` int NOT NULL COMMENT '账户金额',
  `pre_money` int NOT NULL COMMENT '预扣金额',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
INSERT INTO `wallet_tcc_tbl` (`id`, `user_id`, `money`, `pre_money`) VALUES (1, '123456', 1000000, 0);
复制代码

订单表:order_tcc_tbl

-- ----------------------------
-- Table structure for order_tcc_tbl
-- ----------------------------
DROP TABLE IF EXISTS `order_tcc_tbl`;
CREATE TABLE `order_tcc_tbl` (
  `id` int NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `user_id` varchar(255) COLLATE utf8mb4_bin NOT NULL COMMENT '用户ID',
  `commodity_code` varchar(255) COLLATE utf8mb4_bin NOT NULL COMMENT '商品编码',
  `count` int NOT NULL COMMENT '商品数量',
  `unit_price` int NOT NULL COMMENT '商品单价',
  `status` varchar(8) COLLATE utf8mb4_bin NOT NULL COMMENT '订单状态',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=14 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
复制代码

库存表:stock_tcc_tbl

-- ----------------------------
-- Table structure for stock_tcc_tbl
-- ----------------------------
DROP TABLE IF EXISTS `stock_tcc_tbl`;
CREATE TABLE `stock_tcc_tbl` (
  `id` int NOT NULL COMMENT '主键ID',
  `commodity_code` varchar(255) COLLATE utf8mb4_bin NOT NULL COMMENT '商品编码',
  `count` int NOT NULL COMMENT '商品总数',
  `pre_deduct_count` int NOT NULL COMMENT '预扣数量',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
INSERT INTO `stock_tcc_tbl` (`id`, `commodity_code`, `count`, `pre_deduct_count`) VALUES (1, 'CC-54321', 10000, 0);
复制代码

创建TCC Action

在各服务表创建完毕后,我们开始编写核心的TCC Action,用以完成我们的TryCommitCancal逻辑。

Account服务

  • 先构建接口
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;
/**
 * @author zouwei
 * @className IWalletTccAction
 * @date: 2022/10/10 20:50
 * @description:
 */
@LocalTCC
public interface IWalletTccAction {
  /**
   * 预扣款
   *
   * @param businessActionContext
   * @param userId
   * @param amount
   * @return
   */
  @TwoPhaseBusinessAction(name = "prepareDeductMoney", commitMethod = "commitDeductMoney", rollbackMethod = "rollbackDeductMoney")
  boolean prepareDeductMoney(BusinessActionContext businessActionContext,
                             @BusinessActionContextParameter(paramName = "userId") String userId,
                             @BusinessActionContextParameter(paramName = "amount") Long amount);
  /**
   * 提交扣款
   *
   * @param businessActionContext
   * @return
   */
  boolean commitDeductMoney(BusinessActionContext businessActionContext);
  /**
   * 回滚扣款
   *
   * @param businessActionContext
   * @return
   */
  boolean rollbackDeductMoney(BusinessActionContext businessActionContext);
}
复制代码

1.@LocalTCC注解表示该接口是一个TCC Action接口,需要Seata处理;

2.@TwoPhaseBusinessAction注解标注了提交和回滚方法,以便Seata根据预处理结果来决定时调用提交方法还是回滚方法

  • 实现扣款服务逻辑
import com.example.awesomeaccount.dao.mapper.WalletTccEnhanceMapper;
import com.example.awesomeaccount.tcc.IWalletTccAction;
import com.example.awesomeaccount.tcc.TccActionResultWrap;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.util.Map;
/**
 * @author zouwei
 * @className WalletTccActionImpl
 * @date: 2022/10/10 21:05
 * @description:
 */
@Component
public class WalletTccActionImpl implements IWalletTccAction {
  // 用来调用数据库
  @Resource
  private WalletTccEnhanceMapper walletTccEnhanceMapper;
  @Override
  public boolean prepareDeductMoney(BusinessActionContext businessActionContext, String userId, Long amount) {
    String xid = businessActionContext.getXid();
    // 幂等性判断
    if (TccActionResultWrap.hasPrepareResult(xid)) {
      return true;
    }
    // 避免空悬挂,已经执行过回滚了就不能再预留资源
    if (TccActionResultWrap.hasRollbackResult(xid) || TccActionResultWrap.hasCommitResult(xid)) {
      return false;
    }
    // 预留资源
    // 相关sql: update wallet_tcc_tbl set money = money - #{amount,jdbcType=INTEGER}, pre_money = pre_money + #{amount,jdbcType=INTEGER} where user_id = #{userId,jdbcType=VARCHAR} and money <![CDATA[ >= ]]>#{amount,jdbcType=INTEGER}
    boolean result = walletTccEnhanceMapper.prepareDeductMoney(userId, amount) > 0;
    // 记录执行结果:xid:result
    // 以便回滚时判断是否是空回滚
    TccActionResultWrap.prepareSuccess(xid);
    return result;
  }
  // 保证提交逻辑的原子性
  @Transactional
  @Override
  public boolean commitDeductMoney(BusinessActionContext businessActionContext) {
    String xid = businessActionContext.getXid();
    // 幂等性判断
    if (TccActionResultWrap.hasCommitResult(xid)) {
      return true;
    }
    Map<String, Object> actionContext = businessActionContext.getActionContext();
    String userId = (String) actionContext.get("userId");
    long amount = (Integer) actionContext.get("amount");
    // 执行提交操作,扣除预留款
    // 相关sql: update wallet_tcc_tbl set pre_money = pre_money - #{amount,jdbcType=INTEGER} where user_id = #{userId,jdbcType=VARCHAR} and pre_money <![CDATA[ >= ]]>#{amount,jdbcType=INTEGER}
    boolean result = walletTccEnhanceMapper.commitDeductMoney(userId, amount) > 0;
    // 清除预留结果
    TccActionResultWrap.removePrepareResult(xid);
    // 设置提交结果
    TccActionResultWrap.commitSuccess(xid);
    return result;
  }
  @Transactional
  @Override
  public boolean rollbackDeductMoney(BusinessActionContext businessActionContext) {
    String xid = businessActionContext.getXid();
    // 幂等性判断
    if (TccActionResultWrap.hasRollbackResult(xid)) {
      return true;
    }
    // 没有预留资源结果,回滚不做任何处理;
    if (!TccActionResultWrap.hasPrepareResult(xid)) {
      // 设置回滚结果,防止空悬挂
      TccActionResultWrap.rollbackResult(xid);
      return true;
    }
    // 执行回滚
    Map<String, Object> actionContext = businessActionContext.getActionContext();
    String userId = (String) actionContext.get("userId");
    long amount = (Integer) actionContext.get("amount");
    // 相关sql: update wallet_tcc_tbl set money = money + #{amount,jdbcType=INTEGER}, pre_money = pre_money - #{amount,jdbcType=INTEGER} where user_id = #{userId,jdbcType=VARCHAR} and pre_money <![CDATA[ >= ]]>#{amount,jdbcType=INTEGER}
    boolean result = walletTccEnhanceMapper.rollbackDeductMoney(userId, amount) > 0;
    // 清除预留结果
    TccActionResultWrap.removePrepareResult(xid);
    // 设置回滚结果
    TccActionResultWrap.rollbackResult(xid);
    return result;
  }
}
复制代码

1.预扣款逻辑就是先把订单总金额从money字段下挪到pre_money下,前提是当前账户中有足够的金额可以扣除;

2.当账户有足够的金额时,那么意味着预扣款成功,这笔钱是其他事务不能操作的,只能静静地等待TM发起分布式事务提交或回滚指令;

3.一旦接收到提交指令,那么Seata就会调用commitDeductMoney方法把已经扣除的金额从pre_money扣掉,这就意味着这笔钱真正地被花掉了;

4.如果接收到回滚指令,说明下单失败,我们就需要把预扣款的钱还给客户,此时Seata就会调用rollbackDeductMoney方法,把之前挪到pre_money下的金额加回到总金额money中,用户账户里的钱就回来了。

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;
/**
 * @author zouwei
 * @className IOrderTccAction
 * @date: 2022/10/11 12:58
 * @description:
 */
@LocalTCC
public interface IOrderTccAction {
  /**
   * 预创建订单
   *
   * @param businessActionContext
   * @param userId
   * @param commodityCode
   * @param count
   * @param unitPrice
   * @return
   */
  @TwoPhaseBusinessAction(name = "prepareOrder", commitMethod = "commitOrder", rollbackMethod = "rollbackOrder")
  boolean prepareOrder(BusinessActionContext businessActionContext,
                       @BusinessActionContextParameter(paramName = "userId") String userId,
                       @BusinessActionContextParameter(paramName = "userId") String commodityCode,
                       @BusinessActionContextParameter(paramName = "userId") int count,
                       @BusinessActionContextParameter(paramName = "userId") long unitPrice);
  /**
   * 订单生效
   *
   * @param businessActionContext
   * @return
   */
  boolean commitOrder(BusinessActionContext businessActionContext);
  /**
   * 回滚预创建订单
   *
   * @param businessActionContext
   * @return
   */
  boolean rollbackOrder(BusinessActionContext businessActionContext);
}
复制代码
import com.example.awesomeorder.dao.entity.OrderTcc;
import com.example.awesomeorder.dao.mapper.OrderTccMapper;
import com.example.awesomeorder.tcc.IOrderTccAction;
import com.example.awesomeorder.tcc.TccActionResultWrap;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.time.LocalDateTime;
/**
 * @author zouwei
 * @className OrderTccActionImpl
 * @date: 2022/10/11 13:05
 * @description:
 */
@Component
public class OrderTccActionImpl implements IOrderTccAction {
  @Resource
  private OrderTccMapper orderTccMapper;
  @Override
  public boolean prepareOrder(BusinessActionContext businessActionContext, String userId, String commodityCode, int count, long unitPrice) {
    String xid = businessActionContext.getXid();
    // 幂等性判断
    if (TccActionResultWrap.hasPrepareResult(xid)) {
      return true;
    }
    // 避免空悬挂,已经执行过回滚了就不能再预留资源
    if (TccActionResultWrap.hasRollbackResult(xid) || TccActionResultWrap.hasCommitResult(xid)) {
      return false;
    }
    // 准备预创建订单
    OrderTcc order = new OrderTcc();
    // 创建时间
    order.setCreateTime(LocalDateTime.now());
    // 用户ID
    order.setUserId(userId);
    // 数量
    order.setCount(count);
    // 商品编码
    order.setCommodityCode(commodityCode);
    // 单价
    order.setUnitPrice(unitPrice);
    // 设置状态
    order.setStatus("预创建");
    // 创建订单
    boolean result = orderTccMapper.insert(order) > 0;
    // 记录主键ID,为了传递给提交或回滚
    businessActionContext.addActionContext("id", order.getId());
    // 记录执行结果:xid:result
    // 以便回滚时判断是否是空回滚
    TccActionResultWrap.prepareSuccess(xid);
    return result;
  }
  @Transactional
  @Override
  public boolean commitOrder(BusinessActionContext businessActionContext) {
    String xid = businessActionContext.getXid();
    // 幂等性判断
    if (TccActionResultWrap.hasCommitResult(xid)) {
      return true;
    }
    // 修改预留资源状态
    Integer id = (Integer) businessActionContext.getActionContext("id");
    OrderTcc row = new OrderTcc();
    row.setId(id);
    row.setStatus("成功");
    boolean result = orderTccMapper.updateByPrimaryKeySelective(row) > 0;
    // 清除预留结果
    TccActionResultWrap.removePrepareResult(xid);
    // 设置提交结果
    TccActionResultWrap.commitSuccess(xid);
    return result;
  }
  @Transactional
  @Override
  public boolean rollbackOrder(BusinessActionContext businessActionContext) {
    String xid = businessActionContext.getXid();
    // 幂等性判断
    if (TccActionResultWrap.hasRollbackResult(xid)) {
      return true;
    }
    // 没有预留资源结果,回滚不做任何处理;
    if (!TccActionResultWrap.hasPrepareResult(xid)) {
      // 设置回滚结果,防止空悬挂
      TccActionResultWrap.rollbackResult(xid);
      return true;
    }
    // 执行回滚,删除预创建订单
    Integer id = (Integer) businessActionContext.getActionContext("id");
    boolean result = orderTccMapper.deleteByPrimaryKey(id) > 0;
    // 清除预留结果
    TccActionResultWrap.removePrepareResult(xid);
    // 设置回滚结果
    TccActionResultWrap.rollbackResult(xid);
    return result;
  }
}



相关文章
|
2月前
|
存储 数据可视化 Java
基于MicrometerTracing门面和Zipkin实现集成springcloud2023的服务追踪
Sleuth将会停止维护,Sleuth最新版本也只支持springboot2。作为替代可以使用MicrometerTracing在微服务中作为服务追踪的工具。
158 1
|
2月前
|
Java Maven Docker
gitlab-ci 集成 k3s 部署spring boot 应用
gitlab-ci 集成 k3s 部署spring boot 应用
|
25天前
|
XML Java API
Spring Boot集成MinIO
本文介绍了如何在Spring Boot项目中集成MinIO,一个高性能的分布式对象存储服务。主要步骤包括:引入MinIO依赖、配置MinIO属性、创建MinIO配置类和服务类、使用服务类实现文件上传和下载功能,以及运行应用进行测试。通过这些步骤,可以轻松地在项目中使用MinIO的对象存储功能。
|
26天前
|
消息中间件 监控 Java
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
31 6
|
26天前
|
Java 关系型数据库 MySQL
如何将Spring Boot + MySQL应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot + MySQL应用程序部署到Pivotal Cloud Foundry (PCF)
45 5
|
26天前
|
缓存 监控 Java
如何将Spring Boot应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot应用程序部署到Pivotal Cloud Foundry (PCF)
35 5
|
27天前
|
消息中间件 Java Kafka
什么是Apache Kafka?如何将其与Spring Boot集成?
什么是Apache Kafka?如何将其与Spring Boot集成?
65 5
|
29天前
|
消息中间件 Java Kafka
Spring Boot 与 Apache Kafka 集成详解:构建高效消息驱动应用
Spring Boot 与 Apache Kafka 集成详解:构建高效消息驱动应用
43 1
|
1月前
|
消息中间件 监控 Java
您是否已集成 Spring Boot 与 ActiveMQ?
您是否已集成 Spring Boot 与 ActiveMQ?
53 0
|
2月前
|
前端开发 Java 程序员
springboot 学习十五:Spring Boot 优雅的集成Swagger2、Knife4j
这篇文章是关于如何在Spring Boot项目中集成Swagger2和Knife4j来生成和美化API接口文档的详细教程。
246 1

热门文章

最新文章