分布式事务seata-at模式使用

1. seata安装

docker 安装

docker run --name seata-server \
        -p 8091:8091 \
        -p 7091:7091 \
        -e SEATA_IP=192.168.0.250 \
        -e SEATA_PORT=8091 \
        seataio/seata-server

将安装好的配置文件数据，拷贝一份到物理机

docker cp seata-serve:/seata-server/resources /User/seata/config

修改配置文件

server:
  port: 7091
  
spring:
  application:
    name: seata-server
    
logging:
  config: classpath:logback-spring.xml
  file:
    path: ${log.home:${user.home}/logs/seata}
  extend:
    logstash-appender:
      destination: 127.0.0.1:4560
    kafka-appender:
      bootstrap-servers: 127.0.0.1:9092
      topic: logback_to_logstash

console:
  user:
    username: seata
    password: seata
seata:
  config:
    # support: nacos, consul, apollo, zk, etcd3
    type: nacos
    nacos:
      server-addr: 192.168.0.250:8848
      namespace:
      group: SEATA_GROUP
      username: nacos
      password: nacos
      context-path:
      ##if use MSE Nacos with auth, mutex with username/password attribute
      #access-key:
      #secret-key:
      data-id: seataServer.properties
  registry:
    # support: nacos, eureka, redis, zk, consul, etcd3, sofa
    type: nacos
    nacos:
      application: seata-server
      server-addr: 192.168.0.250:8848
      group: SEATA_GROUP
      namespace:
      cluster: default
      username: nacos
      password: nacos
      context-path:
  store:
    # support: file 、 db 、 redis
    mode: db
    db:
      datasource: druid
      db-type: mysql
      driver-class-name: com.mysql.jdbc.Driver
      url: jdbc:mysql://192.168.0.250:3306/seata-work?rewriteBatchedStatements=true # 配置数据库服务端地址，提前初始化SQL
      user: root
      password: mysql_xMYB44
      min-conn: 10
      max-conn: 100
      global-table: global_table
      branch-table: branch_table
      lock-table: lock_table
      distributed-lock-table: distributed_lock
      query-limit: 1000
      max-wait: 5000
  security:
    secretKey: SeataSecretKey0c382ef121d778043159209298fd40bf3850a017
    tokenValidityInMilliseconds: 1800000
    ignore:
      urls: /,/**/*.css,/**/*.js,/**/*.html,/**/*.map,/**/*.svg,/**/*.png,/**/*.jpeg,/**/*.ico,/api/v1/auth/login

删除旧的seata-server，重新安装，并进行路径映射

docker run --name seata-server \
        -p 8091:8091 \
        -p 7091:7091 \
        -e SEATA_IP=192.168.0.250 \
        -e SEATA_PORT=8091 \
        -v /User/seata/config:/seata-server/resources  \
        seataio/seata-server

2. 数据库初始化

客户端

-- 注意此处0.3.0+ 增加唯一索引 ux_undo_log
CREATE TABLE `undo_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) NOT NULL,
  `context` varchar(128) NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime NOT NULL,
  `log_modified` datetime NOT NULL,
  `ext` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

服务端

-- -------------------------------- 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_status_gmt_modified` (`status` , `gmt_modified`),
    KEY `idx_transaction_id` (`transaction_id`)
) ENGINE = InnoDB
  DEFAULT CHARSET = utf8mb4;

-- 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 = utf8mb4;

-- the table to store lock data
CREATE TABLE IF NOT EXISTS `lock_table`
(
    `row_key`        VARCHAR(128) NOT NULL,
    `xid`            VARCHAR(128),
    `transaction_id` BIGINT,
    `branch_id`      BIGINT       NOT NULL,
    `resource_id`    VARCHAR(256),
    `table_name`     VARCHAR(32),
    `pk`             VARCHAR(36),
    `status`         TINYINT      NOT NULL DEFAULT '0' COMMENT '0:locked ,1:rollbacking',
    `gmt_create`     DATETIME,
    `gmt_modified`   DATETIME,
    PRIMARY KEY (`row_key`),
    KEY `idx_status` (`status`),
    KEY `idx_branch_id` (`branch_id`),
    KEY `idx_xid` (`xid`)
) ENGINE = InnoDB
  DEFAULT CHARSET = utf8mb4;

CREATE TABLE IF NOT EXISTS `distributed_lock`
(
    `lock_key`       CHAR(20) NOT NULL,
    `lock_value`     VARCHAR(20) NOT NULL,
    `expire`         BIGINT,
    primary key (`lock_key`)
) ENGINE = InnoDB
  DEFAULT CHARSET = utf8mb4;

INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('AsyncCommitting', ' ', 0);
INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('RetryCommitting', ' ', 0);
INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('RetryRollbacking', ' ', 0);
INSERT INTO `distributed_lock` (lock_key, lock_value, expire) VALUES ('TxTimeoutCheck', ' ', 0);

3. Pom配置

        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>io.seata</groupId>
                    <artifactId>seata-spring-boot-starter</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        
        <!-- 若版本加载不出来，则单独添加 --> 
        <dependency>
            <groupId>io.seata</groupId>
            <artifactId>seata-spring-boot-starter</artifactId>
            <version>1.7.1</version>  
        </dependency>

4. seata配置

# 配置seata
seata:
  enabled: true
  tx-service-group: xiaoqiu_tx_group # 事务组
  service:
    vgroup-mapping:
      xiaoqiu_tx_group: SEATA_GROUP # 事务组映射
    grouplist:
      SEATA_GROUP: 192.168.0.250:8091 # seata ip地址
  config:
    nacos:
      server-addr: 192.168.0.250:8848
      username: nacos
      password: nacos
  registry:
    nacos:
      server-addr: 192.168.0.250:8848
      username: nacos
      password: nacos

5. 自动回滚

直接把原有的@Transactional替换成@GlobalTransactional即可

1. 当**本机发生异常，且未被全局异常处理器捕获时**，即可直接回滚；
1. 当**外部调用发生异常时，返回的状态码非200**，也可以直接回滚。

6. 手动回滚

外部异常，但返回状态码还是200

调用外部接口时，外部接口自己捕获了异常，此时返回的http状态码为200，不会自动回滚，需要根据返回的数据单独判断，并进行手动回滚。

举例：

// 初始化简历
R<Void> r = workResumeServiceFeign.init(user.getId());
// 如果调用状态不是200，则手动回滚全局事务
if (r.getStatus() != HttpStatusEnum.SUCCESS.getCode()) {
    String xid = RootContext.getXID();
    if (StringUtils.isNotBlank(xid)) {
        try {
            GlobalTransactionContext.reload(xid).rollback();
        } catch (TransactionException e) {
            log.error("createUsers, 回滚全局事务失败！ mobile: {}", mobile, e);
        } finally {
            GraceException.display(ResponseStatusEnum.USER_REGISTER_ERROR);
        }
    }
}

本地异常，但异常被全局异常捕获

此时本地发生了异常，但是由于优雅处理异常的全局异常处理类，把异常吃掉了，所以需要单独处理。此时可以加一个AOP切面，对方法进行包装，手动的拦截要全局事务包裹的类，然后手动回滚异常。

举例：

异常代码：

		@GlobalTransactional
    @Override
    public Users createUsers(String mobile) {
        Users user = getDefaultUsers(mobile);
        user.setMobile(mobile);
        usersMapper.insert(user);

        // 初始化简历
        workResumeServiceFeign.init(user.getId());
        // 模拟除0异常
        int a = 1 / 0;
        return user;
    }

切面：

@Slf4j
@Component
@Aspect
public class SeataTransactionAspect {

    /**
     * 调用service之前，手动加入或者创建全局事务
     */
    @Before("execution(* com.xiaoqiu.service.impl.UsersServiceImpl.createUsers(..))")
    public void beginTransaction(JoinPoint joinPoint) throws TransactionException {
        log.info("手动开启全局事务");
        // 手动开启全局事务
        GlobalTransaction gt = GlobalTransactionContext.getCurrentOrCreate();
        gt.begin();
    }

    /**
     * 捕获异常，则手动回滚全局事务
     */
    @AfterThrowing(
            throwing = "throwable",
            pointcut = "execution(* com.xiaoqiu.service.impl.UsersServiceImpl.createUsers(..))"
    )
    public void seataRollback(Throwable throwable) throws Throwable {
        log.warn("捕获到异常信息，则回滚，异常信息为：{}", throwable.getMessage());

        // 从当前线程获得xid
        String xid = RootContext.getXID();
        if (StringUtils.isNotBlank(xid)) {
            GlobalTransactionContext.reload(xid).rollback();
        }
    }

}

如果觉得文章对你有用，请随意赞赏

微服务事务 SpringCloud

分布式事务seata-at模式使用

http://192.168.0.251:2081/archives/d834d277-7ce1-4388-91c0-81352ce26f97

作者

Administrator

发布于

2024-10-30

更新于

2024-10-30

许可协议

CC BY 4.0