微服务集成Seata分布式事务

大大源码 • • 其他

Seata是什么

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

Seata术语

TC (Transaction Coordinator) - 事务协调者
维护全局和分支事务的状态，驱动全局事务提交或回滚。
TM (Transaction Manager) - 事务管理器
定义全局事务的范围：开始全局事务、提交或回滚全局事务。
RM (Resource Manager) - 资源管理器
管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

集成Seata

1、版本选择

因为Seatas是Spring Cloud Alibaba下面的一款开源分布式事务解决框架，所以按照Spring Cloud Alibaba提供的毕业版本对应关系，减少不必要的问题。
spring-cloud-alibaba版本依赖关系

Spring Cloud Alibaba 2.2.7.RELEASE
Spring Cloud Hoxton.SR12
Spring Boot 2.3.12.RELEASE
Seata 1.3.0

2、部署事务协调者TC

下载seata-server

官网下载地址 | github下载地址
这里选择seata 1.3.0版本的zip进行下载

配置

所有配置文件都在conf目录下

file.conf
  • 指定mode方式为db

file.conf配置seata-server数据存储方式，mode字段指定模式，有file、db、redis等。默认是file，这边选择db数据库模式。


## transaction log store, only used in seata-server
store {
## store mode: file、db、redis
mode = "db"

## database store property
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"
## mysql 8 要改驱动com.mysql.jdbc.Driver为com.mysql.cj.jdbc.Driver
driverClassName = "com.mysql.cj.jdbc.Driver"
## 下面数据库配置要改成自己的
url = "jdbc:mysql://127.0.0.1:3306/seata?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8"
user = "root"
password = "123456"
minConn = 5
maxConn = 30
globalTable = "global_table"
branchTable = "branch_table"
lockTable = "lock_table"
queryLimit = 100
maxWait = 5000
}
}
  • 初始化seata数据库

sql脚本地址：https://github.com/seata/seata/tree/1.3.0/script/server/db，这里选择mysql.sql

-- -------------------------------- The script used when storeMode is 'db' --------------------------------
-- the table to store GlobalSession data
CREATE TABLE IF NOT EXISTS `global_table`
(
    `xid`                       VARCHAR(128) NOT NULL,
    `transaction_id`            BIGINT,
    `status`                    TINYINT      NOT NULL,
    `application_id`            VARCHAR(32),
    `transaction_service_group` VARCHAR(32),
    `transaction_name`          VARCHAR(128),
    `timeout`                   INT,
    `begin_time`                BIGINT,
    `application_data`          VARCHAR(2000),
    `gmt_create`                DATETIME,
    `gmt_modified`              DATETIME,
    PRIMARY KEY (`xid`),
    KEY `idx_gmt_modified_status` (`gmt_modified`, `status`),
    KEY `idx_transaction_id` (`transaction_id`)
) ENGINE = InnoDB
  DEFAULT CHARSET = utf8;

-- the table to store BranchSession data
CREATE TABLE IF NOT EXISTS `branch_table`
(
    `branch_id`         BIGINT       NOT NULL,
    `xid`               VARCHAR(128) NOT NULL,
    `transaction_id`    BIGINT,
    `resource_group_id` VARCHAR(32),
    `resource_id`       VARCHAR(256),
    `branch_type`       VARCHAR(8),
    `status`            TINYINT,
    `client_id`         VARCHAR(64),
    `application_data`  VARCHAR(2000),
    `gmt_create`        DATETIME(6),
    `gmt_modified`      DATETIME(6),
    PRIMARY KEY (`branch_id`),
    KEY `idx_xid` (`xid`)
) ENGINE = InnoDB
  DEFAULT CHARSET = utf8;

-- the table to store lock data
CREATE TABLE IF NOT EXISTS `lock_table`
(
    `row_key`        VARCHAR(128) NOT NULL,
    `xid`            VARCHAR(96),
    `transaction_id` BIGINT,
    `branch_id`      BIGINT       NOT NULL,
    `resource_id`    VARCHAR(256),
    `table_name`     VARCHAR(32),
    `pk`             VARCHAR(36),
    `gmt_create`     DATETIME,
    `gmt_modified`   DATETIME,
    PRIMARY KEY (`row_key`),
    KEY `idx_branch_id` (`branch_id`)
) ENGINE = InnoDB
  DEFAULT CHARSET = utf8;
registry.conf

registry.conf配置seata-server的注册中心和配置中心地址。默认type是file，这里选择nacos作为注册和配置中心

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

  nacos {
    # 注册到nacos的服务名
    application = "seata-server"
    # nacos地址
    serverAddr = "127.0.0.1:8848"
    # 服务分组名称
    group = "SEATA_GROUP"
    # 服务所在的nacos命名空间
    namespace = ""
    cluster = "default"
    username = "nacos"
    password = "nacos"
  }
}

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

  nacos {
    serverAddr = "127.0.0.1:8848"
    namespace = ""
    group = "SEATA_GROUP"
    username = "nacos"
    password = "nacos"
  }

}

初始化配置中心，把https://github.com/seata/seata/tree/1.3.0/ 下的整个文件夹script文件夹拷贝到seata的目录下。
在这里插入图片描述

编辑script/config-center目录下的config.txt文件

transport.type=TCP
transport.server=NIO
transport.heartbeat=true
transport.enableClientBatchSendRequest=false
transport.threadFactory.bossThreadPrefix=NettyBoss
transport.threadFactory.workerThreadPrefix=NettyServerNIOWorker
transport.threadFactory.serverExecutorThreadPrefix=NettyServerBizHandler
transport.threadFactory.shareBossWorker=false
transport.threadFactory.clientSelectorThreadPrefix=NettyClientSelector
transport.threadFactory.clientSelectorThreadSize=1
transport.threadFactory.clientWorkerThreadPrefix=NettyClientWorkerThread
transport.threadFactory.bossThreadSize=1
transport.threadFactory.workerThreadSize=default
transport.shutdown.wait=3
# 注意这里是事务分组，默认分组名称是my_test_tx_group，
# 这里的default对应的是刚刚配置的nacos注册中心cluster = "default" 集群名称
service.vgroupMapping.my_test_tx_group=default
service.default.grouplist=127.0.0.1:8091
service.enableDegrade=false
service.disableGlobalTransaction=false
client.rm.asyncCommitBufferLimit=10000
client.rm.lock.retryInterval=10
client.rm.lock.retryTimes=30
client.rm.lock.retryPolicyBranchRollbackOnConflict=true
client.rm.reportRetryCount=5
client.rm.tableMetaCheckEnable=false
client.rm.sqlParserType=druid
client.rm.reportSuccessEnable=false
client.rm.sagaBranchRegisterEnable=false
client.tm.commitRetryCount=5
client.tm.rollbackRetryCount=5
client.tm.degradeCheck=false
client.tm.degradeCheckAllowTimes=10
client.tm.degradeCheckPeriod=2000
# 这里mode指定为db, 默认是file
store.mode=db
store.db.datasource=druid
store.db.dbType=mysql
# 数据库连接配置这里也改成自己的
store.db.driverClassName=com.mysql.cj.jdbc.Driver
store.db.url=jdbc:mysql://127.0.0.1:3306/seata?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8
store.db.user=root
store.db.password=123456

store.db.minConn=5
store.db.maxConn=30
store.db.globalTable=global_table
store.db.branchTable=branch_table
store.db.queryLimit=100
store.db.lockTable=lock_table
store.db.maxWait=5000
store.redis.host=127.0.0.1
store.redis.port=6379
store.redis.maxConn=10
store.redis.minConn=1
store.redis.database=0
store.redis.password=null
store.redis.queryLimit=100
server.recovery.committingRetryPeriod=1000
server.recovery.asynCommittingRetryPeriod=1000
server.recovery.rollbackingRetryPeriod=1000
server.recovery.timeoutRetryPeriod=1000
server.maxCommitRetryTimeout=-1
server.maxRollbackRetryTimeout=-1
server.rollbackRetryTimeoutUnlockEnable=false
client.undo.dataValidation=true
client.undo.logSerialization=jackson
client.undo.onlyCareUpdateColumns=true
server.undo.logSaveDays=7
server.undo.logDeletePeriod=86400000
client.undo.logTable=undo_log
client.log.exceptionRate=100
transport.serialization=seata
transport.compressor=none
metrics.enabled=false
metrics.registryType=compact
metrics.exporterList=prometheus
metrics.exporterPrometheusPort=9898

将config.txt的配置推送到nacos的配置中心，在script/config-center/nacos目录下有两个脚本文件，nacos-config.py和nacos-config.sh，这边选择sh文件执行，因为有安装git有git bash可以执行sh文件，命令如下：

sh nacos-config.sh -h 127.0.0.1 -p 8848 -g SEATA_GROUP -t 30fe3db1-8205-4ee8-86be-be79aa67a1b2 -u nacos -w nacos
  • -h：nacos的IP地址
  • -p：nacos的端口
  • -g：指定要推送group
  • -t： 指定要推送的命名空间，没有创建默认public
  • -u：nacos服务的账号
  • -w：nacos服务的密码
    推送结果
启动seate-server

bin目录下的seata-server.bat

在这里插入图片描述

查看nacos服务

在这里插入图片描述

3、编写seata的client端

简单设计

client端实现

1、初始化数据库
-- 新建order数据库 添加order_tbl表
DROP TABLE IF EXISTS `order_tbl`;
CREATE TABLE `order_tbl` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `user_id` varchar(255) DEFAULT NULL,
  `commodity_code` varchar(255) DEFAULT NULL,
  `count` int(11) DEFAULT 0,
  `money` int(11) DEFAULT 0,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- 新建account数据库 添加account_tbl表
DROP TABLE IF EXISTS `account_tbl`;
CREATE TABLE `account_tbl` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `user_id` varchar(255) DEFAULT NULL,
  `money` int(11) DEFAULT 0,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

两个库要添加undo_log表，用于记录数据在本地事务前后的状态，seata默认使用AT模式，所以使用https://github.com/seata/seata/tree/1.3.0/script/client/at/db 下的建表语句mysql.sql

-- for AT mode you must to init this sql for you business database. the seata server not need it.
CREATE TABLE IF NOT EXISTS `undo_log`
(
    `branch_id`     BIGINT(20)   NOT NULL COMMENT 'branch transaction id',
    `xid`           VARCHAR(100) NOT NULL COMMENT 'global transaction id',
    `context`       VARCHAR(128) NOT NULL COMMENT 'undo_log context,such as serialization',
    `rollback_info` LONGBLOB     NOT NULL COMMENT 'rollback info',
    `log_status`    INT(11)      NOT NULL COMMENT '0:normal status,1:defense status',
    `log_created`   DATETIME(6)  NOT NULL COMMENT 'create datetime',
    `log_modified`  DATETIME(6)  NOT NULL COMMENT 'modify datetime',
    UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`)
) ENGINE = InnoDB
  AUTO_INCREMENT = 1
  DEFAULT CHARSET = utf8 COMMENT ='AT transaction mode undo table';
2、业务代码
  • maven依赖
<properties>
  <java.version>1.8</java.version>
  <maven.compiler.source>8</maven.compiler.source>
  <maven.compiler.target>8</maven.compiler.target>
  <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <spring.cloud.alibaba>2.2.7.RELEASE</spring.cloud.alibaba>
  <spring.cloud.version>Hoxton.SR12</spring.cloud.version>
</properties>

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <!--   nacos注册中心和配置中心     -->
  <dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
  </dependency>
  <dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
  </dependency>
  <!--   openfeign服务调用     -->
  <dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-openfeign</artifactId>
  </dependency>
  <!--   druid数据源     -->
  <dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid-spring-boot-starter</artifactId>
  </dependency>
  <!--   mysql驱动     -->
  <dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
  </dependency>
  <!--   JPA     -->
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-jpa</artifactId>
  </dependency>
  <!--   lombok     -->
  <dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
  </dependency>
  <!--  seata依赖      -->
  <dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
  </dependency>

</dependencies>

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

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

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

这里引入的是spring-cloud-starter-alibaba-seata,里面帮我们解决了feign调用XID参数传递的问题，不然子事务获取不到全局事务的XID。

在这里插入图片描述

  • account服务
server:
  port: 8001

# 数据源配置
spring:
  application:
    name: seata-account
  datasource:
    url: jdbc:mysql://localhost:3306/seata-account?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8
    driver-class-name: com.mysql.cj.jdbc.Driver
    type: com.alibaba.druid.pool.DruidDataSource # 自定义数据源
    username: root
    password: 123456

# 服务注册中心
  cloud:
    nacos:
      server-addr: localhost:8848
# seata配置，与seata-server的registry.conf一致
seata:
  enabled: true
  tx-service-group: my_test_tx_group
  registry:
    type: nacos
    nacos:
      application: seata-server
      server-addr: 127.0.0.1:8848
      group: SEATA_GROUP
      namespace: 
      username: nacos
      password: nacos
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      group: SEATA_GROUP
      namespace: 
      username: nacos
      password: nacos

@EnableFeignClients
@EnableDiscoveryClient
@SpringBootApplication
public class SeataAccountApp {

    public static void main(String[] args) {
        SpringApplication.run(SeataAccountApp.class, args);
    }
}

// entity
@Data
@Accessors(chain = true)
@Entity
@Table(name="account_tbl")
public class Account implements Serializable {

    @Id
    private int id;
    private String userId;
    private int money;

}

// DAO层
public interface AccountDao extends JpaRepository<Account, Integer> {
}

// Service接口
public interface AccountService {

    /**
     * 账户扣款
     */
    void debit(String userId, int money);
}

// AccountService实现
@Service
public class AccountServiceImpl implements AccountService {
    @Autowired
    private AccountDao accountDao;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void debit(String userId, int money) {
        System.out.println("SEATA全局事务XID=================>" +  RootContext.getXID());
        Account account = accountDao.getOne(Integer.valueOf(userId));
        account.setMoney(account.getMoney() - money);
        accountDao.save(account);
    }
}

// controller层
@RestController
public class AccountController {
    @Autowired
    private AccountService accountService;

    @GetMapping("debit")
    public String debit(@RequestParam("userId") String userId, @RequestParam("money") int money) {
        accountService.debit(userId, money);
        return "ok";
    }

}
  • order服务
server:
  port: 8002

# 数据源配置
spring:
  application:
    name: seata-order
  datasource:
    url: jdbc:mysql://localhost:3306/seata-order?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8
    driver-class-name: com.mysql.cj.jdbc.Driver
    type: com.alibaba.druid.pool.DruidDataSource # 自定义数据源
    username: root
    password: 123456

# 服务注册中心
  cloud:
    nacos:
      server-addr: localhost:8848
# seata配置，与seata-server的registry.conf一致
seata:
  enabled: true
  tx-service-group: my_test_tx_group
  registry:
    type: nacos
    nacos:
      application: seata-server
      server-addr: 127.0.0.1:8848
      group: SEATA_GROUP
      namespace: 
      username: nacos
      password: nacos
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      group: SEATA_GROUP
      namespace: 
      username: nacos
      password: nacos

编写AccountFeign.java

@FeignClient(name = "seata-account")
public interface AccountFeign {

    @GetMapping("debit")
    String debit(@RequestParam("userId") String userId, @RequestParam("money") int money);

}

@Entity
@Data
@Accessors(chain = true)
@Table(name="order_tbl")
public class Order implements Serializable {

    @Id
    private int id;
    private String userId;
    private String commodityCode;
    private int count;
    private int money;
}

public interface OrderDao extends JpaRepository<Order, Integer> {
}

public interface OrderService {

    void createOrder(String userId, String commodityCode, int orderCount);
}

@Service
public class OrderServiceImpl implements OrderService {

    @Autowired
    private OrderDao orderDao;
    @Autowired
    private AccountFeign accountFeign;


    @Override
    @Transactional(rollbackFor = Exception.class)
    public void createOrder(String userId, String commodityCode, int orderCount) {
        System.out.println("SEATA全局事务id=================>" +  RootContext.getXID());
        // 调用account服务
        accountFeign.debit(userId, 10);

        Order order = new Order().setUserId(userId).setCommodityCode(commodityCode)
                .setCount(orderCount).setMoney(10);
        // 保存订单
        orderDao.save(order);
    }
}

@RestController
public class OrderController {

    @Autowired
    private OrderService orderService;

    @GetMapping("createOrder")
    public String createOrder(@RequestParam("userId") String userId, @RequestParam("commodityCode") String commodityCode, @RequestParam("orderCount") int orderCount) {
        orderService.createOrder(userId, commodityCode, orderCount);
        return "创建订单完成！";
    }
}
3、调试

正常流程是账户扣款成功 -> 创建订单成功
制造一个异常

@Service
public class OrderServiceImpl implements OrderService {

    @Autowired
    private OrderDao orderDao;
    @Autowired
    private AccountFeign accountFeign;


    @Override
    @Transactional(rollbackFor = Exception.class)
    public void createOrder(String userId, String commodityCode, int orderCount) {
        System.out.println("SEATA全局事务id=================>" +  RootContext.getXID());
        // 调用account服务
        accountFeign.debit(userId, 10);

        Order order = new Order().setUserId(userId).setCommodityCode(commodityCode)
                .setCount(orderCount).setMoney(10);
        // 保存订单
        orderDao.save(order);
        // ArithmeticException 异常
        System.out.println(100/0);
    }
}

此时，账户扣款成功 -> 订单创建失败，两个服务的事务没有一起回滚。
创建全局事务

@Service
public class OrderServiceImpl implements OrderService {

    @Autowired
    private OrderDao orderDao;
    @Autowired
    private AccountFeign accountFeign;


    @Override
    @GlobalTransactional
    public void createOrder(String userId, String commodityCode, int orderCount) {
        System.out.println("SEATA全局事务id=================>" +  RootContext.getXID());
        // 调用account服务
        accountFeign.debit(userId, 10);

        Order order = new Order().setUserId(userId).setCommodityCode(commodityCode)
                .setCount(orderCount).setMoney(10);
        // 保存订单
        orderDao.save(order);
        // ArithmeticException 异常
        System.out.println(100/0);
    }
}

把@Transactional换成@GlobalTransactional表示开启全局事务，一旦有子事务失败，则全局事务回发起回滚，回滚所有子事务。

版权声明：本文为qq_36700462原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
THE END
< <上一篇
下一篇>>