SpringCloud Alibaba入门到精通（十六）- Seata处理分布式事务

阅读数：次 2021-01-23

原创字数统计: 5.9k字 | 预计阅读时长≈ 28分钟

此文章主要讲解 springcloud 中使用 Seata 处理分布式事务相关知识。

Seata

微服务模块，连接多个数据库，多个数据源，而数据库之间的数据一致性需要被保证。

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

Seata 概述

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

Seata 是 Simple Extensible Autonomous Transaction Architecture 的简写，由 feascar 改名而来。

Seata 是阿里开源的分布式事务框架，属于二阶段提交模式。

目前 github 上已经有 12267 颗星了，也很活跃，最新的提交时间很多都是几天前。

首先我们回顾一下在单体应用中，例如一个业务调用了 3 个模块，他们都使用同一个数据源，是靠本地事务来保证事务一致性。

但在微服务架构中，这 3 个模块会变为 3 个独立的微服务，各自有自己的数据源，调用逻辑就变为：

Seata 如何处理呢？

Business 是业务入口，在程序中会通过注解来说明他是一个全局事务，这时他的角色为 TM（事务管理者）。

Business 会请求 TC（事务协调器，一个独立运行的服务），说明自己要开启一个全局事务，TC 会生成一个全局事务 ID（XID），并返回给 Business。

Business 得到 XID 后，开始调用微服务，例如调用 Storage。

Storage 会收到 XID，知道自己的事务属于这个全局事务。Storage 执行自己的业务逻辑，操作本地数据库。

Storage 会把自己的事务注册到 TC，作为这个 XID 下面的一个分支事务，并且把自己的事务执行结果也告诉 TC。

此时 Storage 的角色是 RM（资源管理者），资源是指本地数据库。

Order、Account 的执行逻辑与 Storage 一致。

在各个微服务都执行完成后，TC 可以知道 XID 下各个分支事务的执行结果，TM（Business）也就知道了。

Business 如果发现各个微服务的本地事务都执行成功了，就请求 TC 对这个 XID 提交，否则回滚。

TC 收到请求后，向 XID 下的所有分支事务发起相应请求。

各个微服务收到 TC 的请求后，执行相应指令，并把执行结果上报 TC。

重要机制

（1）全局事务的回滚是如何实现的呢？

Seata 有一个重要的机制：回滚日志。

每个分支事务对应的数据库中都需要有一个回滚日志表 UNDO_LOG，在真正修改数据库记录之前，都会先记录修改前的记录值，以便之后回滚。

在收到回滚请求后，就会根据 UNDO_LOG 生成回滚操作的 SQL 语句来执行。

如果收到的是提交请求，就把 UNDO_LOG 中的相应记录删除掉。

（2）RM 是怎么自动和 TC 交互的？

是通过监控拦截JDBC实现的，例如监控到开启本地事务了，就会自动向 TC 注册、生成回滚日志、向 TC 汇报执行结果。

（3）二阶段回滚失败怎么办？

例如 TC 命令各个 RM 回滚的时候，有一个微服务挂掉了，那么所有正常的微服务也都不会执行回滚，当这个微服务重新正常运行后，TC 会重新执行全局回滚。

分布式事务产生背景

分布式前

单机单库没有问题
从单机单库-> 一对多个库-> 多服务对多个库（分布式微服务）

分布式之后

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

数据库拆分

单库单表支撑不了业务时需要对数据库进行水平拆分。分库分表后，原来在一个数据库上就能完成的写操作，可能会跨多个数据库，就产生了跨数据库事务问题

业务服务化拆分

业务拆分后，一个完整的业务逻辑可能会涉及多个服务，多个服务之间存在跨服务事务问题

用一句话来说：一次业务操作需要跨多个数据源或需要跨多个系统进行远程调用，就会产生分布式事务问题

核心组件

一个典型的分布式事务过程：

分布式事务处理过程的-ID + 三组件模型

Transaction ID XID：# 全局唯一的事务ID

# 3组件概念：
Transaction Coordinator(TC) ：# 事务协调器，维护全局事务的运行状态，负责协调并驱动全局事务的提交或回滚;

Transaction Manager(TM) ：# 控制全局事务的边界，负责开启一个全局事务，并最终发起全局提交或全局回滚的决议;

Resource Manager(RM) ：# 控制分支事务，负责分支注册，状态汇报，并接收事务协调器的指令，驱动分支（本地）事务的提交和回滚；

1597982858545

具体工作过程

再从宏观上梳理一下 Seata 的工作过程：

TM 请求 TC，开始一个新的全局事务，TC 会为这个全局事务生成一个 XID。
XID 通过微服务的调用链传递到其他微服务。
RM 把本地事务作为这个 XID 的分支事务注册到 TC。
TM 请求 TC 对这个 XID 进行提交或回滚。
TC 指挥这个 XID 下面的所有分支事务进行提交、回滚。

下载安装

下载

下载地址：点我去下载

发布说明：点我跳转

本地事务： @Transactional

全局事务： @GlobalTransactional ，我们只需要使用这个注解在业务方法上即可。

安装

我这里用的 0.9.0 版本。

修改 file.conf

修改 conf/file.conf 文件，修改之前先备份。

主要修改：自定义事务组名称 + 事务日志存储模式为 db + 数据库连接信息

新建数据库

名字和 file.conf 指定一致，比如我的是 seata

新建数据表

在新建的数据库 seata 里面创建数据表，db_store.sql 脚本文件在 conf 目录下

修改 registry.conf

养成好习惯，修改前备份。

修改 conf/registry.conf 文件内容：

启动服务

先启动 nacos Server 服务，再启动 seata Server。启动文件都在对应的bin目录下。

案例

数据库准备

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

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

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

下订单—–> 扣库存—–>减账户（余额）

创建 3 个数据库

1
2
3

create database seata_order
create database seata_storage
create database seata_account

seata_order 下建 t_order 表

DROP TABLE IF EXISTS `t_order`;
CREATE TABLE `t_order` (
  `id` bigint(11) NOT NULL AUTO_INCREMENT,
  `user_id` bigint(11) NOT NULL COMMENT '用户id',
  `product_id` bigint(11) NOT NULL COMMENT '产品id',
  `count` int(11) NOT NULL COMMENT '数量',
  `money` decimal(11,0) NOT NULL COMMENT '金额',
  `status` int(1) NOT NULL COMMENT '订单状态：0：创建中；1：已完成',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

seata_storage 下建 t_storage 表

DROP TABLE IF EXISTS `t_storage`;
CREATE TABLE `t_storage` (
  `id` bigint(11) NOT NULL AUTO_INCREMENT,
  `product_id` bigint(11) DEFAULT NULL COMMENT '产品id',
  `total` int(11) DEFAULT NULL COMMENT '总库存',
  `used` int(11) DEFAULT NULL COMMENT '已用库存',
  `residue` int(11) DEFAULT NULL COMMENT '剩余库存',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;


INSERT INTO `t_storage` VALUES ('1', '1', '100', '0', '100');

seata_account 下建 t_account 表

DROP TABLE IF EXISTS `t_account`;
CREATE TABLE `t_account` (
  `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'id',
  `user_id` bigint(11) DEFAULT NULL COMMENT '用户id',
  `total` decimal(10,0) DEFAULT NULL COMMENT '总额度',
  `used` decimal(10,0) DEFAULT NULL COMMENT '已用余额',
  `residue` decimal(10,0) DEFAULT '0' COMMENT '剩余可用额度',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

INSERT INTO `t_account` VALUES ('1', '1', '1000', '0', '1000');

3 库分别建对应的回滚日志表

三个数据库都创建一个回滚日志表，seata/conf/ 有相应的脚本文件 db_undo_log.sql

编写微服务

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

订单模块

新建模块

seata-order-service2001

POM 文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>spring_cloud_atguigu_2020</artifactId>
        <groupId>com.itjing.springcloud</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>seata-order-service2001</artifactId>

    <dependencies>
        <!-- seata -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
            <exclusions>
                <exclusion>
                    <artifactId>seata-all</artifactId>
                    <groupId>io.seata</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>io.seata</groupId>
            <artifactId>seata-all</artifactId>
            <version>0.9.0</version>
        </dependency>

        <!-- springcloud alibaba nacos 依赖,Nacos Server 服务注册中心 -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <!-- open feign 服务调用 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-openfeign</artifactId>
        </dependency>

        <!-- springboot整合Web组件 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <!-- 持久层支持 -->
        <!--druid-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>1.1.10</version>
        </dependency>
        <!--mysql-connector-java-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <!--jdbc-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
        </dependency>
        <!-- mybatis -->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
        </dependency>

        <!--日常通用-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

YML 文件

server:
  port: 2001

spring:
  application:
    name: seata-order-service
  cloud:
    alibaba:
      seata:
        # 自定义事务组，需要和当时在 seata/conf/file.conf 中的一致
        tx-service-group: fsp_tx_group
    nacos:
      discovery:
        server-addr: localhost:8848
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://localhost:3306/seata_order
    username: root
    password: root

mybatis:
  mapper-locations: classpath:mapper/*.xml

# 日志
logging:
  level:
    io:
      seata: info

拷贝文件

将 seata/conf/ 下的 file.conf 和 registry.conf 两个文件拷贝到 resources 目录下，做一点修改。

domain 实体类

Order

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

CommonResult

/**
 * 如果前后端分离，这个是提供给前端信息和数据的类
 * @param <T>
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class CommonResult<T> implements Serializable {

    private Integer code;
    private String messgae;
    private T data;

    /**
     * 查询为空的时候使用的构造器
     *
     * @param code
     * @param messgae
     */
    public CommonResult(Integer code, String messgae) {
        this(code, messgae, null);
    }
}

Dao 接口

@Mapper
public interface OrderDao {

    //创建订单
    public void create(Order order);

    //修改订单状态
    public void update(@Param("userId") Long userId, @Param("status") Integer status);

}

Mapper 文件

在 resources 下新建 mapper 目录，新建 mapper 文件

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.itjing.springcloud.dao.OrderDao">
    <!-- 结果映射 -->
    <resultMap id="BaseResultMap" type="com.itjing.springcloud.domain.Order">
        <id column="id" property="id" jdbcType="BIGINT"></id>
        <result column="user_id" property="userId" jdbcType="BIGINT"></result>
        <result column="product_id" property="productId" jdbcType="BIGINT"></result>
        <result column="count" property="count" jdbcType="INTEGER"></result>
        <result column="money" property="money" jdbcType="DECIMAL"></result>
        <result column="status" property="status" jdbcType="INTEGER"></result>
    </resultMap>

    <insert id="create">
        insert into t_order(id, user_id, product_id, count, money, status)
        values (null, #{userId},#{productId},#{count},#{money},0)
    </insert>

    <update id="update">
        update t_order set status = 1 where user_id=#{userId} and status=#{status}
    </update>
</mapper>

Service 接口

创建 OrderService 、StorageService、AccountService 接口

注意，红框标记的是通过 open-feign 远程调用微服务的 service

OrderService

1
2
3

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

StorageService

@FeignClient(value = "seata-storage-service")
public interface StorageService {

    @PostMapping(value = "/storage/decrease")
    CommonResult decrease(@RequestParam("productId") Long productId, @RequestParam("count") Integer count);
}

AccountService

@FeignClient(value = "seata-account-service")
public interface AccountService {
    @PostMapping(value = "/account/decrease")
    CommonResult decrease(@RequestParam(value = "userId") Long userId, @RequestParam(value = "money") BigDecimal money);
}

Service 接口实现类

@Service
@Slf4j
public class OrderServiceImpl implements OrderService {

    @Resource
    private OrderDao orderDao;

    @Resource
    private StorageService storageService;

    @Resource
    private AccountService accountService;

    @Override
    public void create(Order order) {
        //1 新建订单
        log.info("--------> 开始创建订单");
        orderDao.create(order);
        log.info("--------> 订单微服务开始调用库存，做扣减---Count-");
        //2 扣减库存
        storageService.decrease(order.getProductId(), order.getCount());
        log.info("--------> 订单微服务开始调用库存，库存扣减完成！！");

        log.info("--------> 订单微服务开始调用账户，账户扣减---money-");
        //3 扣减账户
        accountService.decrease(order.getUserId(), order.getMoney());
        log.info("--------> 订单微服务开始调用账户，账户扣减完成!!");

        //4 修改订单状态，从0到1
        log.info("--------> 订单微服务修改订单状态，start");
        orderDao.update(order.getUserId(), 0);
        log.info("--------> 订单微服务修改订单状态，end");
    }
}

Controller

@RestController
public class OrderController {
    @Resource
    private OrderService orderService;

    @RequestMapping("/order/create")
    public CommonResult create(Order order) {
        orderService.create(order);
        return new CommonResult(200, "订单创建成功");
    }
}

配置类

MybatisConfig

@Configuration
@MapperScan({"com.itjing.springcloud.dao"}) //扫描mapper
public class MybatisConfig {
}

DataSourceProxyConfig

package com.itjing.springcloud.config;

import com.alibaba.druid.pool.DruidDataSource;
//这个是配置使用 seata 管理数据源，所以必须配置
import io.seata.rm.datasource.DataSourceProxy;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

import javax.sql.DataSource;

@Configuration
public class DataSourceProxyConfig {

    @Value("${mybatis.mapper-locations}")
    private String mapperLocations;

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource druidDataSource() {
        return new DruidDataSource();
    }

    @Bean
    public DataSourceProxy dataSourceProxy(DataSource dataSource) {
        return new DataSourceProxy(dataSource);
    }

    @Bean
    public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception {
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSourceProxy);
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(mapperLocations));
        sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory());
        return sqlSessionFactoryBean.getObject();
    }
}

主启动类

//这里必须排除数据源自动配置，因为写了配置类，让 seata 管理数据源
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableFeignClients
@EnableDiscoveryClient
public class SeataOrderMain2001 {
    public static void main(String[] args) {
        SpringApplication.run(SeataOrderMain2001.class, args);
    }
}

库存模块

新建模块

seata-storage-service2002

POM 文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>spring_cloud_atguigu_2020</artifactId>
        <groupId>com.itjing.springcloud</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>seata-storage-service2002</artifactId>

    <dependencies>
        <!-- seata -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
            <exclusions>
                <exclusion>
                    <artifactId>seata-all</artifactId>
                    <groupId>io.seata</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>io.seata</groupId>
            <artifactId>seata-all</artifactId>
            <version>0.9.0</version>
        </dependency>

        <!-- springcloud alibaba nacos 依赖,Nacos Server 服务注册中心 -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <!-- springboot整合Web组件 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <!-- 持久层支持 -->
        <!--druid-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>1.1.10</version>
        </dependency>
        <!--mysql-connector-java-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <!--jdbc-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
        </dependency>
        <!-- mybatis -->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
        </dependency>

        <!--日常通用-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

YML 文件

server:
  port: 2002

spring:
  application:
    name: seata-storage-service
  cloud:
    alibaba:
      seata:
        # 自定义事务组，需要和当时在 seata/conf/file.conf 中的一致
        tx-service-group: fsp_tx_group
    nacos:
      discovery:
        server-addr: localhost:8848
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://localhost:3306/seata_storage
    username: root
    password: root

mybatis:
  mapper-locations: classpath:mapper/*.xml

# 日志
logging:
  level:
    io:
      seata: info

拷贝文件

同订单模块。

domain 实体类

Storage

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Storage implements Serializable {
    private Long id;
    private Long productId;
    private Integer total;
    private Integer used;
    private Integer residue;
}

CommonResult，同订单模块

Dao 接口

@Mapper
public interface StorageDao {
    /*扣减库存*/
    void decrease(@Param("productId") Long productId, @Param("count") Integer count);
}

Mapper 文件

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.itjing.springcloud.dao.StorageDao">
    <resultMap id="BaseResultMap" type="com.itjing.springcloud.domain.Storage">
        <id column="id" property="id" jdbcType="BIGINT"></id>
        <result column="product_id" property="productId" jdbcType="BIGINT"></result>
        <result column="total" property="total" jdbcType="INTEGER"></result>
        <result column="used" property="used" jdbcType="INTEGER"></result>
        <result column="residue" property="residue" jdbcType="INTEGER"></result>
    </resultMap>

    <update id="decrease">
        UPDATE
            t_storage
        SET
            used = used + #{count},residue = residue - #{count}
        WHERE
            product_id = #{productId}
    </update>
</mapper>

Service 接口

1
2
3

public interface StorageService {
    void decrease(Long productId, Integer count);
}

Service 接口实现类

package com.itjing.springcloud.service.impl;


import com.itjing.springcloud.dao.StorageDao;
import com.itjing.springcloud.service.StorageService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class StorageServiceImpl implements StorageService {

    private static final Logger LOGGER = LoggerFactory.getLogger(StorageServiceImpl.class);

    @Resource
    private StorageDao storageDao;

    /**
     * 扣减库存
     *
     * @param productId
     * @param count
     */
    @Override
    public void decrease(Long productId, Integer count) {
        LOGGER.info("--------->storage-service扣减库存开始");
        storageDao.decrease(productId, count);
        LOGGER.info("--------->storage-service扣减库存结束");
    }
}

Controller

@RestController
public class StorageController {
    @Autowired
    private StorageService storageService;

    @RequestMapping("/storage/decrease")
    public CommonResult decrease(Long productId, Integer count) {
        storageService.decrease(productId, count);
        return new CommonResult(200, "扣减库存成功！");
    }
}

配置类

同订单模块。

主启动类

//这里必须排除数据源自动配置，因为写了配置类，让 seata 管理数据源
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableDiscoveryClient
public class SeataStorageMain2002 {
    public static void main(String[] args) {
        SpringApplication.run(SeataStorageMain2002.class, args);
    }
}

账户模块

新建模块

seata-account-service2003

POM 文件

同库存模块。

YML 文件

server:
  port: 2003

spring:
  application:
    name: seata-account-service
  cloud:
    alibaba:
      seata:
        # 自定义事务组，需要和当时在 seata/conf/file.conf 中的一致
        tx-service-group: fsp_tx_group
    nacos:
      discovery:
        server-addr: localhost:8848
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://localhost:3306/seata_account
    username: root
    password: root

mybatis:
  mapper-locations: classpath:mapper/*.xml

# 日志
logging:
  level:
    io:
      seata: info

拷贝文件

同库存模块。

domain 实体类

Account

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Account implements Serializable {
    private Long id;
    private Long userId;
    private BigDecimal total;
    private BigDecimal used;
    private BigDecimal residue;
}

CommonResult，同库存模块。

Dao 接口

@Mapper
public interface AccountDao {
    /*扣账户余额*/
    void decrease(@Param("userId") Long userId, @Param("money") BigDecimal money);
}

Mapper 文件

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.itjing.springcloud.dao.AccountDao">
    <resultMap id="BaseResultMap" type="com.itjing.springcloud.domain.Account">
        <id column="id" property="id" jdbcType="BIGINT"></id>
        <result column="user_id" property="userId" jdbcType="BIGINT"></result>
        <result column="total" property="total" jdbcType="DECIMAL"></result>
        <result column="used" property="used" jdbcType="DECIMAL"></result>
        <result column="residue" property="residue" jdbcType="DECIMAL"></result>
    </resultMap>

    <update id="decrease">
        UPDATE
            t_account
        SET
            used = used + #{money},residue = residue - #{money}
        WHERE
            user_id = #{userId}
    </update>

</mapper>

Service 接口

public interface AccountService {
    /*扣减账户余额*/
    void decrease(Long userId, BigDecimal money);
}

package com.itjing.springcloud.service.impl;


import com.itjing.springcloud.dao.AccountDao;
import com.itjing.springcloud.service.AccountService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;

@Service
public class AccountServiceImpl implements AccountService {

    private static final Logger LOGGER = LoggerFactory.getLogger(AccountServiceImpl.class);

    @Resource
    private AccountDao accountDao;


    /**
     * 扣减账户余额
     * @param userId
     * @param money
     */
    @Override
    public void decrease(Long userId, BigDecimal money) {
        LOGGER.info("--------->account-service扣减余额开始");
        accountDao.decrease(userId, money);
        LOGGER.info("--------->account-service扣减余额结束");
    }
}

Controller

@RestController
public class AccountController {
    @Autowired
    private AccountService accountService;

    @RequestMapping("/account/decrease")
    public CommonResult decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money) {
        accountService.decrease(userId, money);
        return new CommonResult(200, "扣减余额成功！");
    }
}

配置类

同库存模块。

主启动类

//这里必须排除数据源自动配置，因为写了配置类，让 seata 管理数据源
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableDiscoveryClient
public class SeataAccountMain2003 {
    public static void main(String[] args) {
        SpringApplication.run(SeataAccountMain2003.class, args);
    }
}

测试

先启动 nacos，再启动 seata，然后启动上面的 3 个微服务。

我这里直接用 GET 请求测试了，访问： http://localhost:2001/order/create?userId=1&productId=1&count=10&money=100

模拟超时异常

没加 @GlobalTransactional 注解，模拟账户超时异常

/**
 * 扣减账户余额
 * @param userId
 * @param money
 */
@Override
public void decrease(Long userId, BigDecimal money) {
    LOGGER.info("--------->account-service扣减余额开始");
    try {
        TimeUnit.SECONDS.sleep(20); //模拟超时，openfeign的默认超时为1s
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    accountDao.decrease(userId, money);
    LOGGER.info("--------->account-service扣减余额结束");
}

可以发现：

订单成功创建，但是没有成功支付。

但是库存却被扣掉了！

账户也被扣钱了。

而且由于 feign 的超时重试机制，账户余额还有可能被多次扣减，这太可怕了！

加 @GlobalTransactional 注解

还是模拟账户超时异常，这次在订单业务实现类 OrderServiceImpl 加上 @GlobalTransactional 注解

@Override
//只需要在业务类的方法上加上该注解，name值自定义唯一即可。 rollbackFor表示发生异常回滚
@GlobalTransactional(name = "fsp-create-order", rollbackFor = Exception.class)
public void create(Order order) {
    //1 新建订单
    log.info("--------> 开始创建订单");
    orderDao.create(order);
    log.info("--------> 订单微服务开始调用库存，做扣减---Count-");
    //2 扣减库存
    storageService.decrease(order.getProductId(), order.getCount());
    log.info("--------> 订单微服务开始调用库存，库存扣减完成！！");

    log.info("--------> 订单微服务开始调用账户，账户扣减---money-");
    //3 扣减账户
    accountService.decrease(order.getUserId(), order.getMoney());
    log.info("--------> 订单微服务开始调用账户，账户扣减完成!!");

    //4 修改订单状态，从0到1
    log.info("--------> 订单微服务修改订单状态，start");
    orderDao.update(order.getUserId(), 0);
    log.info("--------> 订单微服务修改订单状态，end");
}

再次访问，发现页面出现异常，数据库中的数据没有改变，记录都添加不进来，测试成功。

本文作者： LiJing
发布时间： 2021-01-23
最后更新： 2024-06-24
本文标题： SpringCloud Alibaba入门到精通（十六）- Seata处理分布式事务
本文链接： https://blog-yilia.xiaojingge.com/posts/cbafe4d9.html
版权声明： 本作品采用 CC BY-NC-SA 4.0 许可协议进行许可。转载请注明出处！