关于分布式事务:分布式事务二Seata-框架-AT-事务

44次阅读

共计 18213 个字符，预计需要花费 46 分钟才能阅读完成。

Seata 介绍

Seata 是一款开源的分布式事务解决方案，致力于提供高性能和简略易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式，为用户打造一站式的分布式解决方案。

2019 年 1 月，阿里巴巴中间件团队发动了开源我的项目 Fescar（Fast & EaSy Commit And Rollback），和社区一起共建开源分布式事务解决方案。Fescar 的愿景是让分布式事务的应用像本地事务的应用一样，简略和高效，并逐渐解决开发者们遇到的分布式事务方面的所有难题。

Fescar 开源后，蚂蚁金服退出 Fescar 社区参加共建，并在 Fescar 0.4.0 版本中奉献了 TCC 模式。

为了打造更中立、更凋谢、生态更加丰盛的分布式事务开源社区，通过社区核心成员的投票，大家决定对 Fescar 进行品牌降级，并更名为 Seata，意为：Simple Extensible Autonomous Transaction Architecture，是一套一站式分布式事务解决方案。

Seata 交融了阿里巴巴和蚂蚁金服在分布式事务技术上的积攒，并积淀了新批发、云计算和新金融等场景下丰盛的实践经验，但要实现实用于所有的分布式事务场景的愿景，仍有很长的路要走。

Seata AT 事务计划

Seata 的 AT 模式（Automatic Transaction）是一种 无侵入 的分布式事务解决方案。上面联合具体业务场景来剖析其执行的原理。

业务场景

订单零碎


当用户下订单时，执行以下三步流程：

  1. 订单零碎 保留订单
  2. 订单零碎调用 库存服务，缩小商品库存
  3. 订单零碎调用 账户服务，扣减用户金额

这三步要作为一个整体事务进行治理，要么整体胜利，要么整体失败。

Seata AT 基本原理

Seata AT 事务分两个阶段来治理全局事务：
第一阶段： 执行各分支事务
第二阶段： 管制全局事务最终提交或回滚

第一阶段：执行各分支事务

微服务零碎中，各服务之间无奈互相感知事务是否执行胜利，这时就须要一个专门的服务，来协调各个服务的运行状态。这个服务称为 TC（Transaction Coordinator），事务协调器。

订单零碎开始执行保留订单之前，首先启动 TM（Transaction Manager，事务管理器），由 TM 向 TC 申请开启一个全局事务：


这时 TC 会产生一个 全局事务 ID，称为 XID，并将 XID 传回 TM：

这样就开启了 全局事务！

全局事务开启后，开始执行创立订单的业务。首先执行保留订单，这时会先启动一个 RM（Resource Manager，资源管理器），并将 XID 传递给 RM。

RM 负责对分支事务（即微服务的本地事务）进行治理，并与 TC 通信，上报分支事务的执行状态、接管全局事务的提交或回滚指令。

RM 首先会应用 XID 向 TC 注册分支事务，将分支事务纳入对应的全局事务管辖。

当初能够执行保留订单的分支事务了。一旦分支事务执行胜利，RM 会上报事务状态：


TC 收到后，会将该状态信息传递到 TM：

到此，保留订单过程完结。上面是调用库存服务，缩小商品库存，与订单的执行过程雷同。

首先调用库存服务，启动 RM，并传递 XID：

库存服务的 RM 应用 XID 向 TC 进行注册，纳入全局事务管辖：

执行本地事务胜利后上报状态，TC 会将状态发送给 TM：

雷同的，实现账户分支事务：

第二阶段：管制全局事务最终提交

当初，TM（全局事务管理器）收集齐了全副分支事务的胜利状态，它会进行决策，确定全局事务胜利，向 TC 发送全局事务的提交申请：

而后，TC 会向所有 RM 发送提交操作指令，RM 会实现最终提交操作：


到此，全局事务全副提交实现！

第二阶段：管制全局事务最终回滚

下面是全局事务执行胜利的状况，上面再来看看事务执行失败的状况。

假如订单业务执行过程中，扣减账户金额这一步分支事务执行失败，那么失败状态对 TC 上报，而后再发送给 TM：


TM 会进行决策，确定全局事务 失败 ，向 TC 发送全局事务的 回滚 申请：


而后，TC 会向所有 RM 发送回滚操作指令，RM 会实现最终回滚操作：

Seata AT 具体工作机制

以上理解了 Seata AT 的基本原理、工作流程，那么 Seata 具体是如何实现全局事务的提交和回滚操作呢？上面来剖析 Seata 的具体工作机制。

第一阶段：执行分支事务

以全面订单业务中的库存服务为例，库存表中存在一条商品的库存信息：


当初要执行业务操作缩小库存，从 50 件缩小到 40 件：

执行批改库存业务操作前，会先取出旧的库存信息：

当初能够批改库存了：


接着，取出更新后的新数据：

接下来，会把旧数据和新数据合并起来，保留到一个事务回滚日志表：undo_log 表：

至此，第一阶段，分支事务实现，将状态上报给 TC：

第二阶段：管制全局事务最终回滚

如果全局事务失败，那么第一阶段已提交的分支事务要执行回滚操作。

首先会收到来自 TC 的全局事务回滚指令：


接下来，依据事务回滚日志（undo_log）表的记录，将商品复原成旧的库存数据：

而后删除事务日志，最终实现第二阶段回滚操作：

第二阶段：管制全局事务最终提交

下面是全局事务回滚操作。如果全局事务胜利，要实现最终提交，AT 模式最终提交操作非常简单，只须要删除日志数据即可。

首先接管到 TC 的全局事务提交指令：


接着，间接删除事务日志，就实现了第二阶段提交操作：

导入订单我的项目，无事务版本

下载我的项目代码

  1. 拜访 git 仓库 https://gitee.com/benwang6/seata-samples
  2. 拜访我的项目标签
  3. 下载无事务版

订单案例波及四个数据库：


应用 db-init 工具，来重置所有数据库表，能够不便地把数据重置到初始状态。

Seata Server – TC 全局事务协调器

在后面介绍了 Seata AT 事务原理，介绍了 AT 事务的三个角色：TC（事务协调器）、TM（事务管理器）和 RM（资源管理器），其中 TM 和 RM 是嵌入在业务利用中的，而 TC 则是一个独立服务。

Seata Server 就是 TC，间接从官网仓库下载启动即可，下载地址：https://github.com/seata/seata/releases

Seata Server 配置

Seata Server 的配置文件有两个：

  • seata/conf/registry.conf
  • seata/conf/file.conf

registry.conf

Seata Server 要向注册核心进行注册，这样，其余服务就能够通过注册核心去发现 Seata Server，与 Seata Server 进行通信。

Seata 反对多款注册核心服务：nacos、eureka、redis、zk、consul、etcd3、sofa。

咱们我的项目中要应用 eureka 注册核心，eureka 服务的连贯地址、注册的服务名，这须要在 registry.conf 文件中进行配置：

registry {
  # file、nacos、eureka、redis、zk、consul、etcd3、sofa
  # 这里抉择 eureka 注册配置
  type = "eureka"

  nacos {......}

  # eureka 的注册配置
  eureka {
    # 注册核心地址
    serviceUrl = "http://localhost:8761/eureka"
    # 注册的服务 ID
    application = "seata-server"
    weight = "1"
  }
  
  redis {......}
  ......

file.conf

Seata 须要存储全局事务信息、分支事务信息、全局锁信息，这些数据存储到什么地位？

针对存储地位的配置，反对放在配置核心，或者也能够放在本地文件。Seata Server 反对的配置核心服务有：nacos、apollo、zk、consul、etcd3。

这里咱们抉择最简略的，应用本地文件，这须要在 registry.conf 配置文件中来指定：

......

config {
  # file、nacos、apollo、zk、consul、etcd3
  # 在这里抉择应用本地文件来保留配置
  type = "file"

......

  etcd3 {serverAddr = "http://localhost:2379"}
  
  file {
    # 在这里设置配置文件的文件名
    name = "file.conf"
  }
}

file.conf 中对事务信息的存储地位进行配置，存储地位反对：file、db、redis。

这里咱们抉择数据库作为存储地位，这须要在 file.conf 中进行配置：

store {
  ## store mode: file、db、redis
  # 这里抉择数据库存储
  mode = "db"

  ## file store property
  file {......}

  # 数据库存储
  db {## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp)/HikariDataSource(hikari) etc.
    datasource = "druid"
    ## mysql/oracle/postgresql/h2/oceanbase etc.
    dbType = "mysql"
    driverClassName = "com.mysql.jdbc.Driver"

    # 数据库连贯配置
    url = "jdbc:mysql://127.0.0.1:3306/seata?useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8"
    user = "root"
    password = "root"
    minConn = 5
    maxConn = 30

    # 事务日志表表名设置
    globalTable = "global_table"
    branchTable = "branch_table"
    lockTable = "lock_table"

    queryLimit = 100
    maxWait = 5000
  }

  ## redis store property
  redis {......}
}

启动参数设置

启动文件：seata-server.bat

用文本编辑器关上文件，找到文件中这一行：

%JAVACMD% %JAVA_OPTS% -server -Xmx2048m -Xms2048m -Xmn1024m -Xss512k -XX:Sur......

看到 Seata Server 默认应用 2G 内存，测试环境咱们能够把内存调低：

%JAVACMD% %JAVA_OPTS% -server -Xmx256m -Xms256m -Xmn128m -Xss512k -XX:Sur......

启动 Seata Server

双击 seata-server.bat 启动 Seata Server。

查看 Eureka 注册核心 Seata Server 的注册信息：

order 订单服务增加 Seata AT 事务

订单调用库存和账户，咱们先从后面的订单开始。

在订单我的项目中要启动 全局事务 ，还要执行订单保留的 分支事务。

order-parent 增加 seata 依赖

order-parent 的 pom.xml 文件中有一段正文掉的 seata 依赖，当初能够关上它了：

<?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">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>cn.tedu</groupId>
    <artifactId>order-parent</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>pom</packaging>
    <name>order-parent</name>

    <properties>
        <mybatis-plus.version>3.3.2</mybatis-plus.version>
        <druid-spring-boot-starter.version>1.1.23</druid-spring-boot-starter.version>
        <seata.version>1.3.0</seata.version>
        <spring-cloud-alibaba-seata.version>2.0.0.RELEASE</spring-cloud-alibaba-seata.version>
        <spring-cloud.version>Hoxton.SR6</spring-cloud.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-openfeign</artifactId>
        </dependency>

        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>${mybatis-plus.version}</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>${druid-spring-boot-starter.version}</version>
        </dependency>

        <!-- 关上上面这段 seata 依赖 -->
        <dependency>
          <groupId>com.alibaba.cloud</groupId>
          <artifactId>spring-cloud-alibaba-seata</artifactId>
          <version>${spring-cloud-alibaba-seata.version}</version>
          <exclusions>
            <exclusion>
              <artifactId>seata-all</artifactId>
              <groupId>io.seata</groupId>
            </exclusion>
          </exclusions>
        </dependency>
        <dependency>
          <groupId>io.seata</groupId>
          <artifactId>seata-all</artifactId>
          <version>${seata.version}</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

配置

application.yml

TC 事务协调器通过“事务组”的形式将多个服务组织成一个全局事务。每个服务启动时要向 TC 注册，退出同一个事务组。

spring:
  ......
  
  cloud:
    alibaba:
      seata:
        tx-service-group: order_tx_group

......

registry.conf

须要从注册核心取得 TC 的地址，这里配置注册核心的地址。

TC 在注册核心注册的服务 ID 在上面 file.conf 中指定。

registry {
  # file、nacos、eureka、redis、zk、consul、etcd3、sofa
  type = "eureka"

  nacos {
    serverAddr = "localhost"
    namespace = ""cluster ="default"
  }
  eureka {
    serviceUrl = "http://localhost:8761/eureka"
    # application = "default"
    # weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = "0"
    password = ""cluster ="default"timeout ="0"
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
    username = ""password =""
  }
  consul {
    cluster = "default"
    serverAddr = "127.0.0.1:8500"
  }
  etcd3 {
    cluster = "default"
    serverAddr = "http://localhost:2379"
  }
  sofa {
    serverAddr = "127.0.0.1:9603"
    application = "default"
    region = "DEFAULT_ZONE"
    datacenter = "DefaultDataCenter"
    cluster = "default"
    group = "SEATA_GROUP"
    addressWaitTime = "3000"
  }
  file {name = "file.conf"}
}

config {
  # file、nacos、apollo、zk、consul、etcd3、springCloudConfig
  type = "file"

  nacos {
    serverAddr = "localhost"
    namespace = ""group ="SEATA_GROUP"
  }
  consul {serverAddr = "127.0.0.1:8500"}
  apollo {
    app.id = "seata-server"
    apollo.meta = "http://192.168.1.204:8801"
    namespace = "application"
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
    username = ""password =""
  }
  etcd3 {serverAddr = "http://localhost:2379"}
  file {name = "file.conf"}
}

file.conf

在这里咱们指定 TC 的服务 ID seata-server：

vgroupMapping.order_tx_group = "seata-server"

order_tx_group 对应 application.yml 中注册的事务组名。

transport {
  # tcp udt unix-domain-socket
  type = "TCP"
  #NIO NATIVE
  server = "NIO"
  #enable heartbeat
  heartbeat = true
  # the client batch send request enable
  enableClientBatchSendRequest = true
  #thread factory for netty
  threadFactory {
    bossThreadPrefix = "NettyBoss"
    workerThreadPrefix = "NettyServerNIOWorker"
    serverExecutorThread-prefix = "NettyServerBizHandler"
    shareBossWorker = false
    clientSelectorThreadPrefix = "NettyClientSelector"
    clientSelectorThreadSize = 1
    clientWorkerThreadPrefix = "NettyClientWorkerThread"
    # netty boss thread size,will not be used for UDT
    bossThreadSize = 1
    #auto default pin or 8
    workerThreadSize = "default"
  }
  shutdown {
    # when destroy server, wait seconds
    wait = 3
  }
  serialization = "seata"
  compressor = "none"
}
service {
  #transaction service group mapping
  # order_tx_group 与 yml 中的“tx-service-group: order_tx_group”配置统一
  #“seata-server”与 TC 服务器的注册名统一
  # 从 eureka 获取 seata-server 的地址，再向 seata-server 注册本人，设置 group
  vgroupMapping.order_tx_group = "seata-server"
  #only support when registry.type=file, please don't set multiple addresses
  order_tx_group.grouplist = "127.0.0.1:8091"
  #degrade, current not support
  enableDegrade = false
  #disable seata
  disableGlobalTransaction = false
}

client {
  rm {
    asyncCommitBufferLimit = 10000
    lock {
      retryInterval = 10
      retryTimes = 30
      retryPolicyBranchRollbackOnConflict = true
    }
    reportRetryCount = 5
    tableMetaCheckEnable = false
    reportSuccessEnable = false
  }
  tm {
    commitRetryCount = 5
    rollbackRetryCount = 5
  }
  undo {
    dataValidation = true
    logSerialization = "jackson"
    logTable = "undo_log"
  }
  log {exceptionRate = 100}
}

创立 seata 数据源代理

Seata AT 事务对业务代码无侵入，全自动化解决全局事务，其性能是靠 Seata 的数据源代理工具实现的。

这里咱们创立 Seata 的数据源代理，并排除 Spring 默认的数据源。

package cn.tedu.order;

import com.alibaba.druid.pool.DruidDataSource;
import io.seata.rm.datasource.DataSourceProxy;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import javax.sql.DataSource;

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

    @Primary
    @Bean("dataSource")
    public DataSourceProxy dataSource(DataSource druidDataSource){return new DataSourceProxy(druidDataSource);
    }
}

主程序中排除 Springboot 的默认数据源：

package cn.tedu.order;

import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.cloud.openfeign.EnableFeignClients;

@EnableFeignClients
@MapperScan("cn.tedu.order.mapper")
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)// 不应用 springboot 本人的数据源主动配置
public class OrderApplication {public static void main(String[] args) {SpringApplication.run(OrderApplication.class, args);
    }

}

启动全局事务

Seata AT 对业务无侵入，所以启动全局事务非常简单，只须要增加一个 @GlobalTransactional 注解即可。

另外咱们一步一步地增加全局事务并测试，这里先把 storage 和 account 调用注掉。

package cn.tedu.order.service;

import cn.tedu.order.entity.Order;
import cn.tedu.order.feign.AccountClient;
import cn.tedu.order.feign.EasyIdGeneratorClient;
import cn.tedu.order.feign.StorageClient;
import cn.tedu.order.mapper.OrderMapper;
import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderServiceImpl implements OrderService {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    EasyIdGeneratorClient easyIdGeneratorClient;
    @Autowired
    private AccountClient accountClient;
    @Autowired
    private StorageClient storageClient;
    
    // 由 TM（Transaction Manager），由 TM 向协调器申请开启全局事务
    @GlobalTransactional
    @Override
    public void create(Order order) {
        // 从全局惟一 id 发号器取得 id
       Long orderId = Long.valueOf(easyIdClient.nextId("order_business"));
        order.setId(orderId);

        orderMapper.create(order);

        // 批改库存
        //storageClient.decrease(order.getProductId(), order.getCount());

        // 批改账户余额
        //accountClient.decrease(order.getUserId(), order.getMoney());

    }
}

启动 order 我的项目进行测试

按程序启动服务：

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

调用保留订单，地址：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

察看控制台，看到全局事务和订单的分支事务曾经启动，并能够看到全局事务 ID（XID）和分支事务 ID（Branch ID）：

而后察看数据库中新增加的订单数据：

测试出现异常，回滚的状况

在业务代码中加一个模仿异样再试一下：

package cn.tedu.order.service;

import cn.tedu.order.entity.Order;
import cn.tedu.order.feign.AccountClient;
import cn.tedu.order.feign.EasyIdGeneratorClient;
import cn.tedu.order.feign.StorageClient;
import cn.tedu.order.mapper.OrderMapper;
import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderServiceImpl implements OrderService {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    EasyIdGeneratorClient easyIdGeneratorClient;
    @Autowired
    private AccountClient accountClient;
    @Autowired
    private StorageClient storageClient;

    @GlobalTransactional
    @Override
    public void create(Order order) {
        // 从全局惟一 id 发号器取得 id
        Long orderId = easyIdGeneratorClient.nextId("order_business");
        order.setId(orderId);

        orderMapper.create(order);

        if (Math.random() < 0.5) {throw new RuntimeException("模仿异样");
        }

        // 批改库存
        //storageClient.decrease(order.getProductId(), order.getCount());

        // 批改账户余额
        //accountClient.decrease(order.getUserId(), order.getMoney());

    }
}

重启 order 我的项目，并调用保留订单：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

能够看到全局事务回滚的日志：

订单启动全局事务局部实现，在持续之前，先把模仿异样正文掉：

 ......

        //if (Math.random() < 0.5) {//    throw new RuntimeException("模仿异样");
        //}

        ......

storage 库存服务增加 Seata AT 事务

配置

与订单我的项目中增加的配置完全相同，请参考订单配置章节配置上面三个文件：

  • application.yml
  • registry.conf
  • file.conf

创立 seata 数据源代理

与订单我的项目中数据源代理完全相同，请参考订单中数据源代理章节，在 cn.tedu.storage 包下创立数据源配置类 DatasourceConfiguration。主程序注解排除 DataSourceAutoConfiguration 主动配置类。

启动分支事务

在业务办法上增加 @Transactional 注解启动本地事务：

package cn.tedu.storage.service;

import cn.tedu.storage.mapper.StorageMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class StorageServiceImpl implements StorageService {
    @Autowired
    private StorageMapper storageMapper;

    @Transactional
    @Override
    public void decrease(Long productId, Integer count) throws Exception {storageMapper.decrease(productId,count);
    }
}

order 的业务类中调用缩小商品库存

后面咱们把调用商品库存正文掉了，现把正文关上：

 // 批改库存
        storageClient.decrease(order.getProductId(), order.getCount());

启动 storage 我的项目进行测试

按程序启动我的项目：

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

调用保留订单，地址：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

订单会调用库存，这两个服务会别离启动一个分支事务，两个分支事务一起组成一个全局事务：

察看两个我的项目的控制台都有 Seata AT 事务的日志，Storage 我的项目控制台如下：

而后察看数据库中新增加的订单和缩小的库存：

缩小的库存：

测试出现异常，回滚的状况

在业务代码中加一个模仿异样再试一下：

package cn.tedu.storage.service;

import cn.tedu.storage.mapper.StorageMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class StorageServiceImpl implements StorageService {
    @Autowired
    private StorageMapper storageMapper;

    @Transactional
    @Override
    public void decrease(Long productId, Integer count) throws Exception {storageMapper.decrease(productId,count);

        if (Math.random() < 0.5) {throw new RuntimeException("模仿异样");
        }
    }
}

重启 storage 我的项目，并调用保留订单：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

查看数据库表 order 和 storage，如果执行胜利会新增订单、缩小库存，如果执行失败则数据没有变动，被回滚了。

storage 分支事务局部实现，在持续之前，先把模仿异样正文掉：

 ......

        //if (Math.random() < 0.5) {//    throw new RuntimeException("模仿异样");
        //}

        ......

account 账户服务增加 Seata AT 事务

配置

与订单我的项目中增加的配置完全相同，请参考订单配置章节配置上面三个文件：

  • application.yml
  • registry.conf
  • file.conf

创立 seata 数据源代理

与订单我的项目中数据源代理完全相同，请参考订单中数据源代理章节，在 cn.tedu.account 包下创立数据源配置类 DatasourceConfiguration。主程序注解排除 DataSourceAutoConfiguration 主动配置类。

启动分支事务

在业务办法上增加 @Transactional 注解启动本地事务：

package cn.tedu.account.service;

import cn.tedu.account.mapper.AccountMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
@Service
public class AccountServiceImpl implements AccountService {
    @Autowired
    private AccountMapper accountMapper;

    @Transactional
    @Override
    public void decrease(Long userId, BigDecimal money) {accountMapper.decrease(userId,money);
    }
}

order 的业务类中调用扣减账户金额

后面咱们把调用账户正文掉了，现把正文关上：

 // 批改账户余额
        accountClient.decrease(order.getUserId(), order.getMoney());

启动 account 我的项目进行测试

按程序启动我的项目：

  1. Eureka
  2. Seata Server
  3. Easy Id Generator
  4. Storage
  5. Account
  6. Order

调用保留订单，地址：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

订单会调用库存和账户，这三个服务会别离启动一个分支事务，三个分支事务一起组成一个全局事务：

察看三个我的项目的控制台都有 Seata AT 事务的日志，account 我的项目控制台如下：

而后察看数据库中的订单表、库存表和账户表。

这是账户表，看到金额曾经被扣减：

测试出现异常，回滚的状况

在业务代码中加一个模仿异样再试一下：

package cn.tedu.account.service;

import cn.tedu.account.mapper.AccountMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
@Service
public class AccountServiceImpl implements AccountService {
    @Autowired
    private AccountMapper accountMapper;

    @Transactional
    @Override
    public void decrease(Long userId, BigDecimal money) {accountMapper.decrease(userId,money);

        if (Math.random() < 0.5) {throw new RuntimeException("模仿异样");
        }
    }
}

重启 account 我的项目，并调用保留订单：
http://localhost:8083/create?userId=1&productId=1&count=10&money=100

查看数据库表 order、storage 和 account，如果执行胜利会新增订单、缩小库存、扣减金额，如果执行失败则数据没有变动，被回滚了。

失败时，在 order 和 storage 控制台能够看到回滚日志。

这是 storage 的回滚日志：

account 分支事务局部实现，最初把模仿异样正文掉：

 ......

        //if (Math.random() < 0.5) {//    throw new RuntimeException("模仿异样");
        //}

        ......

正文完
分布式事务
发表至： 分布式事务
2020-12-14
 0
关于分布式事务:分布式事务实践-解决数据一致性
关于分布式事务:不就是分布式事务这下彻底清楚了😎
关于分布式事务:分布式事务系统之底层原理揭秘
关于分布式事务:分布式锁分析使用Redis实现分布式事务中的锁机制