当前位置：首页 > news >正文

分布式事务Seata-AT模式

news 2025/7/13 17:31:54

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: 7091spring:application:name: seata-serverlogging:config: classpath:logback-spring.xmlfile:path: ${log.home:${user.home}/logs/seata}extend:logstash-appender:destination: 127.0.0.1:4560kafka-appender:bootstrap-servers: 127.0.0.1:9092topic: logback_to_logstashconsole:user:username: seatapassword: seata
seata:config:# support: nacos, consul, apollo, zk, etcd3type: nacosnacos:server-addr: 192.168.0.250:8848namespace:group: SEATA_GROUPusername: nacospassword: nacoscontext-path:##if use MSE Nacos with auth, mutex with username/password attribute#access-key:#secret-key:data-id: seataServer.propertiesregistry:# support: nacos, eureka, redis, zk, consul, etcd3, sofatype: nacosnacos:application: seata-serverserver-addr: 192.168.0.250:8848group: SEATA_GROUPnamespace:cluster: defaultusername: nacospassword: nacoscontext-path:store:# support: file 、 db 、 redismode: dbdb:datasource: druiddb-type: mysqldriver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://192.168.0.250:3306/seata-work?rewriteBatchedStatements=true # 配置数据库服务端地址，提前初始化SQLuser: rootpassword: mysql_xMYB44min-conn: 10max-conn: 100global-table: global_tablebranch-table: branch_tablelock-table: lock_tabledistributed-lock-table: distributed_lockquery-limit: 1000max-wait: 5000security:secretKey: SeataSecretKey0c382ef121d778043159209298fd40bf3850a017tokenValidityInMilliseconds: 1800000ignore: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 = InnoDBDEFAULT 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 = InnoDBDEFAULT 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 = InnoDBDEFAULT 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 = InnoDBDEFAULT 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: truetx-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:8848username: nacospassword: nacosregistry:nacos:server-addr: 192.168.0.250:8848username: nacospassword: nacos

5. 自动回滚

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

1. 当**本机发生异常，且未被全局异常处理器捕获时**，即可直接回滚；
2. 当**外部调用发生异常时，返回的状态码非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@Overridepublic 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());// 从当前线程获得xidString xid = RootContext.getXID();if (StringUtils.isNotBlank(xid)) {GlobalTransactionContext.reload(xid).rollback();}}
}