做应用开发时，要保证数据的一致性我们要对方法添加事务管理，最简单的处理方案是在方法上添加 @Transactional 注解或者通过编程方式管理事务。但这种方案只适用于单数据源的关系型数据库，如果项目配置了多个数据源或者多个微服务的rpc调用，就会导致传统的事务管理方案失效，这就涉及到 分布式事务 管理，下面介绍一个非常好用的分布式事务管理框架：seata。通过seata了解如何在业务中管理分布式事务。
要使用seata首先需要到seata的官方网站下载服务端的安装包，当前最新的版本是2.1.0：apache-seata-2.1.0-incubating-bin.tar.gz 。seata支持多种部署方式：单机部署、高可用部署、docker部署、k8s部署、raft部署。根据自己的业务量选择一种部署方式，这里演示如何使用seata就选择最简单的单机部署方式，所有服务都在虚拟机中启动。
一、配置中心和注册中心
如果我们的项目是使用微服务架构，那么注册中心和配置中心是必不可少的两个组件，seata可以通过配置中心管理配置，为了简化代码选择nacos作为注册中心和配置中心，使用微服务模拟分布式事务场景，微服务选择SpringCloud，服务间调用使用OpenFeign。为了完成这个示例功能，首选就要搭建一个nacos服务：
首先下载nacos安装包：2.4.2.zip
将下载好的nacos解压到服务器的某个目录：
unzip nacos-server-2.4.2.zip
cd nacos
如果只是演示使用，不需要调整任何配置可以直接启动服务开始使用nacos了，启动服务命令是：
sh bin/startup.sh -m standalone
上面这种未做任何配置调整就启动服务的方式，nacos会使用内置的Derby数据库，但是考虑到将来服务的扩展，我们一般会使用关系型数据库管理服务数据，比如MySQL。为了安全我们也要开启nacos的鉴权功能，这就要对nacos的配置文件进行调整，调整配置文件信息：application.properties。
- 修改数据库配置，所有的配置信息都保存到数据库中，nacos开启高可用后就可以使用统一的配置信息：
spring.datasource.platform=mysql
spring.sql.init.platform=mysql
db.num=1
db.url.0=jdbc:mysql://192.168.56.101:3306/nacos?characterEncoding=utf8&connectTimeout=30000&socketTimeout=30000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC
db.user.0=root
db.password.0=nacos@Test
- 开启鉴权，其他的服务端注册或者拉取配置信息都需要身份认证，保证系统的安全：
nacos.core.auth.enabled=true
nacos.core.auth.server.identity.key=nacosTest
nacos.core.auth.server.identity.value=nacosTest2024
nacos.core.auth.plugin.nacos.token.secret.key=SecretKeyh6k53by4u8ct9gd6of5vvgcvgpxf4u5g605ks7hzkknd7uiuim00co9gqocf
- 初始化数据库脚本：
CREATE TABLE `config_info` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id',
`data_id` varchar(255) NOT NULL COMMENT 'data_id',
`group_id` varchar(128) DEFAULT NULL COMMENT 'group_id',
`content` longtext NOT NULL COMMENT 'content',
`md5` varchar(32) DEFAULT NULL COMMENT 'md5',
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间',
`src_user` text COMMENT 'source user',
`src_ip` varchar(50) DEFAULT NULL COMMENT 'source ip',
`app_name` varchar(128) DEFAULT NULL COMMENT 'app_name',
`tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段',
`c_desc` varchar(256) DEFAULT NULL COMMENT 'configuration description',
`c_use` varchar(64) DEFAULT NULL COMMENT 'configuration usage',
`effect` varchar(64) DEFAULT NULL COMMENT '配置生效的描述',
`type` varchar(64) DEFAULT NULL COMMENT '配置的类型',
`c_schema` text COMMENT '配置的模式',
`encrypted_data_key` varchar(1024) NOT NULL DEFAULT '' COMMENT '密钥',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_configinfo_datagrouptenant` (`data_id`,`group_id`,`tenant_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_info';
CREATE TABLE `config_info_aggr` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id',
`data_id` varchar(255) NOT NULL COMMENT 'data_id',
`group_id` varchar(128) NOT NULL COMMENT 'group_id',
`datum_id` varchar(255) NOT NULL COMMENT 'datum_id',
`content` longtext NOT NULL COMMENT '内容',
`gmt_modified` datetime NOT NULL COMMENT '修改时间',
`app_name` varchar(128) DEFAULT NULL COMMENT 'app_name',
`tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_configinfoaggr_datagrouptenantdatum` (`data_id`,`group_id`,`tenant_id`,`datum_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='增加租户字段';
CREATE TABLE `config_info_beta` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id',
`data_id` varchar(255) NOT NULL COMMENT 'data_id',
`group_id` varchar(128) NOT NULL COMMENT 'group_id',
`app_name` varchar(128) DEFAULT NULL COMMENT 'app_name',
`content` longtext NOT NULL COMMENT 'content',
`beta_ips` varchar(1024) DEFAULT NULL COMMENT 'betaIps',
`md5` varchar(32) DEFAULT NULL COMMENT 'md5',
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间',
`src_user` text COMMENT 'source user',
`src_ip` varchar(50) DEFAULT NULL COMMENT 'source ip',
`tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段',
`encrypted_data_key` varchar(1024) NOT NULL DEFAULT '' COMMENT '密钥',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_configinfobeta_datagrouptenant` (`data_id`,`group_id`,`tenant_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_info_beta';
CREATE TABLE `config_info_tag` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id',
`data_id` varchar(255) NOT NULL COMMENT 'data_id',
`group_id` varchar(128) NOT NULL COMMENT 'group_id',
`tenant_id` varchar(128) DEFAULT '' COMMENT 'tenant_id',
`tag_id` varchar(128) NOT NULL COMMENT 'tag_id',
`app_name` varchar(128) DEFAULT NULL COMMENT 'app_name',
`content` longtext NOT NULL COMMENT 'content',
`md5` varchar(32) DEFAULT NULL COMMENT 'md5',
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间',
`src_user` text COMMENT 'source user',
`src_ip` varchar(50) DEFAULT NULL COMMENT 'source ip',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_configinfotag_datagrouptenanttag` (`data_id`,`group_id`,`tenant_id`,`tag_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_info_tag';
CREATE TABLE `config_tags_relation` (
`id` bigint(20) NOT NULL COMMENT 'id',
`tag_name` varchar(128) NOT NULL COMMENT 'tag_name',
`tag_type` varchar(64) DEFAULT NULL COMMENT 'tag_type',
`data_id` varchar(255) NOT NULL COMMENT 'data_id',
`group_id` varchar(128) NOT NULL COMMENT 'group_id',
`tenant_id` varchar(128) DEFAULT '' COMMENT 'tenant_id',
`nid` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'nid, 自增长标识',
PRIMARY KEY (`nid`),
UNIQUE KEY `uk_configtagrelation_configidtag` (`id`,`tag_name`,`tag_type`),
KEY `idx_tenant_id` (`tenant_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='config_tag_relation';
CREATE TABLE `group_capacity` (
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID',
`group_id` varchar(128) NOT NULL DEFAULT '' COMMENT 'Group ID，空字符表示整个集群',
`quota` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '配额，0表示使用默认值',
`usage` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '使用量',
`max_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个配置大小上限，单位为字节，0表示使用默认值',
`max_aggr_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '聚合子配置最大个数，，0表示使用默认值',
`max_aggr_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个聚合数据的子配置大小上限，单位为字节，0表示使用默认值',
`max_history_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '最大变更历史数量',
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_group_id` (`group_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='集群、各Group容量信息表';
CREATE TABLE `his_config_info` (
`id` bigint(20) unsigned NOT NULL COMMENT 'id',
`nid` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'nid, 自增标识',
`data_id` varchar(255) NOT NULL COMMENT 'data_id',
`group_id` varchar(128) NOT NULL COMMENT 'group_id',
`app_name` varchar(128) DEFAULT NULL COMMENT 'app_name',
`content` longtext NOT NULL COMMENT 'content',
`md5` varchar(32) DEFAULT NULL COMMENT 'md5',
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间',
`src_user` text COMMENT 'source user',
`src_ip` varchar(50) DEFAULT NULL COMMENT 'source ip',
`op_type` char(10) DEFAULT NULL COMMENT 'operation type',
`tenant_id` varchar(128) DEFAULT '' COMMENT '租户字段',
`encrypted_data_key` varchar(1024) NOT NULL DEFAULT '' COMMENT '密钥',
PRIMARY KEY (`nid`),
KEY `idx_gmt_create` (`gmt_create`),
KEY `idx_gmt_modified` (`gmt_modified`),
KEY `idx_did` (`data_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='多租户改造';
CREATE TABLE `tenant_capacity` (
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID',
`tenant_id` varchar(128) NOT NULL DEFAULT '' COMMENT 'Tenant ID',
`quota` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '配额，0表示使用默认值',
`usage` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '使用量',
`max_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个配置大小上限，单位为字节，0表示使用默认值',
`max_aggr_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '聚合子配置最大个数',
`max_aggr_size` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '单个聚合数据的子配置大小上限，单位为字节，0表示使用默认值',
`max_history_count` int(10) unsigned NOT NULL DEFAULT '0' COMMENT '最大变更历史数量',
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tenant_id` (`tenant_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='租户容量信息表';
CREATE TABLE `tenant_info` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id',
`kp` varchar(128) NOT NULL COMMENT 'kp',
`tenant_id` varchar(128) default '' COMMENT 'tenant_id',
`tenant_name` varchar(128) default '' COMMENT 'tenant_name',
`tenant_desc` varchar(256) DEFAULT NULL COMMENT 'tenant_desc',
`create_source` varchar(32) DEFAULT NULL COMMENT 'create_source',
`gmt_create` bigint(20) NOT NULL COMMENT '创建时间',
`gmt_modified` bigint(20) NOT NULL COMMENT '修改时间',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tenant_info_kptenantid` (`kp`,`tenant_id`),
KEY `idx_tenant_id` (`tenant_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='tenant_info';
CREATE TABLE `users` (
`username` varchar(50) NOT NULL PRIMARY KEY COMMENT 'username',
`password` varchar(500) NOT NULL COMMENT 'password',
`enabled` boolean NOT NULL COMMENT 'enabled'
);
CREATE TABLE `roles` (
`username` varchar(50) NOT NULL COMMENT 'username',
`role` varchar(50) NOT NULL COMMENT 'role',
UNIQUE INDEX `idx_user_role` (`username` ASC, `role` ASC) USING BTREE
);
CREATE TABLE `permissions` (
`role` varchar(50) NOT NULL COMMENT 'role',
`resource` varchar(128) NOT NULL COMMENT 'resource',
`action` varchar(8) NOT NULL COMMENT 'action',
UNIQUE INDEX `uk_role_permission` (`role`,`resource`,`action`) USING BTREE
);
以上三个步骤执行完成后，就可以启动服务了，服务启动后看到控制台输出信息：
通过浏览器访问：http://192.168.56.101:8848/nacos 看到下面这个信息就表示nacos配置成功了！！！
二、seata服务端
将上面下载好的seata包解压：
tar -zvxf apache-seata-2.1.0-incubating-bin.tar.gz
接下来进入解压后的目录修改seata配置：
- 修改seata的注册中心和配置中心，存储就选择MySQL：
console:
user:
username: seata
password: seata
seata:
config:
type: nacos
nacos:
server-addr: 127.0.0.1:8848
namespace:
group: SEATA_GROUP
context-path:
username: nacos
password: nacos
data-id: seata-server.properties
registry:
type: nacos
nacos:
application: seata-server
server-addr: 127.0.0.1:8848
group: SEATA_GROUP
namespace:
cluster: default
context-path:
username: nacos
password: nacos
store:
mode: db
db:
datasource: druid
db-type: mysql
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://192.168.56.101:3306/seata?rewriteBatchedStatements=true&characterEncoding=utf8&connectTimeout=30000&socketTimeout=30000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC
user: root
password: nacos@Test
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,/version.json,/health,/error
- 运行script目录下的sql脚本，这里支持多种数据库，由于我使用的是mysql，下面执行的就是mysql的脚本：
-- 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);
- 因为使用的是nacos注册中心，我们还需要执行一下配置中心的初始化脚本文件，文件在解压后目录下的 script/config-center/nacos 子目录下，在执行脚本之前需要先调整上一层目录下的文件
config.txt根据自己使用的配置中心和存储方式调整相关内容：
调整存储方式：
store.mode=db
store.lock.mode=db
store.session.mode=db
调整数据库连接：
store.db.datasource=druid
store.db.dbType=mysql
store.db.driverClassName=com.mysql.jdbc.Driverstore.db.url=jdbc:mysql://192.168.56.101:3306/seata?useUnicode=true&rewriteBatchedStatements=true&allowMultiQueries=true&characterEncoding=utf8&useSSL=false&serverTimezone=GMT%2B8
store.db.user=root
store.db.password=nacos@Test
store.db.minConn=5
store.db.maxConn=30
store.db.globalTable=global_table
store.db.branchTable=branch_table
store.db.distributedLockTable=distributed_lock
store.db.queryLimit=100
store.db.lockTable=lock_table
store.db.maxWait=5000
执行脚本之前需要删除未使用的空配置，比如不使用redis就删除相关空配置，其他的配置可以保留。进入nacos脚本目录执行初始化脚本：
sh nacos-config.sh -h 192.168.56.101 -p 8848 -g SEATA_GROUP -u nacos -w nacos
脚本执行完成后可以看到下面的输出：
- 以上所有都完成后，启动服务：
sh seata-server.sh -m db
如果看到控制台输出下面内容表示服务启动成功！！！
三、事务模式使用
seata官方文档介绍了它支持四种事务：AT、XA、TCC、SAGA。下面分别简单示例一下每种模式的使用方式和适用场景。
一般讲到事务就会涉及到付款场景，示例也是通过支付和扣款这两种场景介绍分布式事务的使用，假定一共有三个微服务：seata-demo-account 用于处理账户数据； seata-demo-order 用于处理订单数据； seata-demo 是前置接口服务，通过调用上面两个服务处理数据，三个服务都启动成功后nacos注册中心将三个服务都注册进来：
1、AT模式
seata分布式事务默认开启的模式，有关AT模式实现原理参考seata官方介绍的非常详细，这里只是简单说一下过程：首先要在操作的数据库下面创建一个 undo_log 表，将被执行的sql语句修改前的状态保存到这个 undo_log 表，然后执行每一个子事务，子事务执行完后就提交事务，当每个子事务都执行成功了直接删除 undo_log 表记录；如果其中某一个子事务失败，则所有子事务都按照 undo_log 记录的sql回滚执行。
- 在要使用的数据库下创建
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,
PRIMARY KEY (`id`),
UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
- 在需要事务控制的方法上添加
@GlobalTransactional注解。
以支付场景为例，在我的demo示例代码中是需要 seata-demo 服务分别调用 seata-demo-account 和 seata-demo-order ，这样要保证账户余额和支付订单两个服务数据一致性，就必须要对两个操作添加事务控制，由于两个操作分别在两个服务中，那么就需要使用分布式事务实现控制：
seata-demo中的代码：
import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.xingo.client.AccountClient;
import org.xingo.client.PayOrderClient;
import java.math.BigDecimal;
/**
* @Author xingo
* @Date 2024/9/27
*/
@RestController
public class PayController {
@Autowired
private PayOrderClient payOrderClient;
@Autowired
private AccountClient accountClient;
@GetMapping("/pay")
@GlobalTransactional
public String pay(@RequestParam(value = "userId") int userId,
@RequestParam(value = "pay") BigDecimal pay) {
String rs1 = accountClient.addBalance(userId, pay);
System.out.println("acount: " + rs1);
String rs2 = payOrderClient.payOrder(userId, pay);
System.out.println("order: " + rs2);
return "ok";
}
}
两个远程调用是我使用open-feign封装的客户端：
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import java.math.BigDecimal;
import java.sql.Timestamp;
/**
* @Author xingo
* @Date 2024/9/27
*/
@Component
@CircuitBreaker(name = "default")
@FeignClient(contextId = "payOrderClient", name = "seata-demo-order", path = "/payorder", fallbackFactory = PayOrderClient.PayOrderClientImpl.class)
public interface PayOrderClient {
@GetMapping("/add")
String payOrder(@RequestParam(value = "userId") int userId,
@RequestParam(value = "pay") BigDecimal pay);
@Component
public static class PayOrderClientImpl implements FallbackFactory<PayOrderClient> {
@Override
public PayOrderClient create(Throwable cause) {
return new PayOrderClient() {
@Override
public String payOrder(int userId, BigDecimal pay) {
return this.fail();
}
private String fail() {
return "error, ts: " + new Timestamp(System.currentTimeMillis());
}
};
}
}
}
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import java.math.BigDecimal;
import java.sql.Timestamp;
/**
* @Author xingo
* @Date 2024/9/27
*/
@Component
@CircuitBreaker(name = "default")
@FeignClient(contextId = "accountClient", name = "seata-demo-account", path = "/account", fallbackFactory = AccountClient.PayOrderClientImpl.class)
public interface AccountClient {
@GetMapping("/add")
BigDecimal add(@RequestParam(value = "userId") int userId,
@RequestParam(value = "pay") BigDecimal pay);
@GetMapping("/addbalance")
String addBalance(@RequestParam(value = "userId") int userId,
@RequestParam(value = "pay") BigDecimal pay);
@Component
public static class PayOrderClientImpl implements FallbackFactory<AccountClient> {
@Override
public AccountClient create(Throwable cause) {
return new AccountClient() {
@Override
public BigDecimal add(int userId, BigDecimal pay) {
return null;
}
@Override
public String addBalance(int userId, BigDecimal pay) {
return this.fail();
}
private String fail() {
return "error, ts: " + new Timestamp(System.currentTimeMillis());
}
};
}
}
}
seata-demo-account中实现账户管理
import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.xingo.mapper.AccountMapper;
import java.math.BigDecimal;
/**
* @Author xingo
* @Date 2024/9/27
*/
@RestController
@RequestMapping("/account")
public class AccountController {
@Autowired
private AccountMapper accountMapper;
@GetMapping("/addbalance")
@GlobalTransactional
public String addBalance(@RequestParam(value = "userId") int userId,
@RequestParam(value = "pay") BigDecimal pay) {
int rs = accountMapper.addAmount(pay, userId);
return rs > 0 ? "ok|" + accountMapper.selectBalance(userId) : "fail";
}
}
seata-demo-order中实现订单管理
import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.xingo.common.JacksonUtils;
import org.xingo.domain.SeataPayOrder;
import org.xingo.mapper.PayOrderMapper;
import java.math.BigDecimal;
import java.time.LocalDateTime;
/**
* @Author xingo
* @Date 2024/9/27
*/
@RestController
@RequestMapping("/payorder")
public class PayOrderController {
@Autowired
private PayOrderMapper payOrderMapper;
@GetMapping("/add")
@GlobalTransactional
public String payOrder(@RequestParam(value = "userId") int userId,
@RequestParam(value = "pay") BigDecimal pay) {
SeataPayOrder order = SeataPayOrder.builder()
.userId(userId)
.pay(pay)
.createTime(LocalDateTime.now())
.modifyTime(LocalDateTime.now())
.build();
payOrderMapper.insert(order);
return "ok|" + JacksonUtils.toJSONString(order);
}
}
上面这些方法就已经实现了AT模式的分布式事务控制，下面是我验证事务控制的过程：
当两个子事务都正常执行时数据库会按照请求更新数据，如果代码中有异常时会回滚数据库里面的数据：
（1）异常发生前子事务都提交，同时记录提交前的sql语句到undo_log表：
（2）代码有异常发生后发现数据库里面的数据回滚：
AT模式每个子事务单独控制，这样就不会长时间占用数据库资源，AT模式需要关系性数据库的支持。
2、XA模式
AT模式是事务的最终一致性，这样就会导致数据的中间状态可以被其他线程读到；如果要实现事务的强一致性，那么就要使用数据库都支持的二阶段提交，整个过程如下图：
预提交阶段检查资源是否都已经准备好，如果各个子事务都已经反馈能够提交事务了，在由TC告知各个子事务统一提交或回滚操作，由于在这个过程中会占用数据库资源，性能不如AT模式高，但是这种模式会保证事务的强一致性。
3、TCC模式
无论是AT或者XA模式，都需要底层关系型数据库的事务支持，对于不支持事务的数据库，我们要保证数据的一致性，就需要自己实现事务的管理，在seata中可以通过TCC实现，要实现TCC事务管理，首先要定义一个接口，在接口上面添加 @LocalTCC 注解，并且声明三个方法，分别对应处理数据的方法，提交数据的方法，回滚数据的方法，使用注解 @TwoPhaseBusinessAction(name = "addPreOrder", commitMethod = "confirmPreOrder", rollbackMethod = "cancelPreOrder") 标记，大致定义如下：
import io.seata.rm.tcc.api.BusinessActionContext;
import io.seata.rm.tcc.api.LocalTCC;
import io.seata.rm.tcc.api.TwoPhaseBusinessAction;
import java.math.BigDecimal;
/**
* @Author xingo
* @Date 2024/9/27
*/
@LocalTCC
public interface PreOrderService {
/**
* 订单预付
* @param userId
* @param pay
* @param orderNo
* @return
*/
@TwoPhaseBusinessAction(name = "addPreOrder", commitMethod = "confirmPreOrder", rollbackMethod = "cancelPreOrder")
boolean addPreOrder(int userId, BigDecimal pay, String orderNo);
/**
* 事务的commit方法
* @param ctx
* @return
*/
boolean confirmPreOrder(BusinessActionContext ctx);
/**
* 事务的rollback方法
* @param ctx
* @return
*/
boolean cancelPreOrder(BusinessActionContext ctx);
}
接下来就是要实现上面定义的接口：
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import io.seata.core.context.RootContext;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.xingo.domain.SeataPreOrder;
import org.xingo.mapper.PreOrderMapper;
import org.xingo.service.PreOrderService;
import java.math.BigDecimal;
import java.time.LocalDateTime;
/**
* @Author xingo
* @Date 2024/9/27
*/
@Service
public class PreOrderServiceImpl implements PreOrderService {
@Autowired
private PreOrderMapper preOrderMapper;
@Override
@Transactional(rollbackFor = Exception.class)
public boolean addPreOrder(int userId, BigDecimal pay, String orderNo) {
System.out.println("try-xid: " + RootContext.getXID());
LambdaQueryWrapper<SeataPreOrder> wrapper = Wrappers.<SeataPreOrder>lambdaQuery().eq(SeataPreOrder::getOrderNo, orderNo);
SeataPreOrder find = preOrderMapper.selectOne(wrapper);
if(find == null) {
SeataPreOrder order = SeataPreOrder.builder()
.userId(userId)
.pay(pay)
.createTime(LocalDateTime.now())
.modifyTime(LocalDateTime.now())
.orderNo(orderNo)
.xid(RootContext.getXID())
.build();
preOrderMapper.insert(order);
return true;
}
return false;
}
@Override
@Transactional(rollbackFor = Exception.class)
public boolean confirmPreOrder(BusinessActionContext ctx) {
System.out.println("confirm-xid: " + ctx.getXid());
int rs = preOrderMapper.clearXid(ctx.getXid());
return rs > 0;
}
@Override
@Transactional(rollbackFor = Exception.class)
public boolean cancelPreOrder(BusinessActionContext ctx) {
System.out.println("cancel-xid: " + ctx.getXid());
int rs = preOrderMapper.delByXid(ctx.getXid());
return rs > 0;
}
}
定义好上面的类信息后就声明了一个TCC事务，调用方式跟上面使用AT或XA一样，只需要在请求方法上面添加 @GlobalTransactional 注解就可以管理事务了，如果没有异常情况输出日志大致如下：
如果代码中有异常抛出，那么事务回滚：
TCC模式适用性非常好，因为事务的控制都是我们自己代码实现，他并不依赖底层数据库支持，但是它需要我们自己编写大量的代码实现事务控制。
4、SAGA模式
saga事务可以分为两部分：正向服务和反向补偿服务，正向是事务正常执行时，每一个节点提交本地事务；一旦某个节点失败了，就反向执行补偿。
seata中的saga是通过状态机来管理的，个人感觉这个模式比较复杂，并且在项目中很少用到，基本上前三种模式已经足够了。
在使用OpenFeign时，注意要关闭熔断和降级CircuitBreaker，因为在不做处理的情况下会导致分布式事务ID无法传递到下一个服务，导致分布式事务失效。
