做应用开发时,要保证数据的一致性我们要对方法添加事务管理,最简单的处理方案是在方法上添加 @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=mysqldb.num=1db.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=truenacos.core.auth.server.identity.key=nacosTest
nacos.core.auth.server.identity.value=nacosTest2024nacos.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: seatapassword: seata
seata:config:type: nacosnacos:server-addr: 127.0.0.1:8848namespace:group: SEATA_GROUPcontext-path:username: nacospassword: nacosdata-id: seata-server.propertiesregistry:type: nacosnacos:application: seata-serverserver-addr: 127.0.0.1:8848group: SEATA_GROUPnamespace:cluster: defaultcontext-path:username: nacospassword: nacosstore:mode: dbdb:datasource: druiddb-type: mysqldriver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://192.168.56.101:3306/seata?rewriteBatchedStatements=true&characterEncoding=utf8&connectTimeout=30000&socketTimeout=30000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTCuser: rootpassword: nacos@Testmin-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,/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 = 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);
- 因为使用的是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 {@Autowiredprivate PayOrderClient payOrderClient;@Autowiredprivate AccountClient accountClient;@GetMapping("/pay")@GlobalTransactionalpublic 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);@Componentpublic static class PayOrderClientImpl implements FallbackFactory<PayOrderClient> {@Overridepublic PayOrderClient create(Throwable cause) {return new PayOrderClient() {@Overridepublic 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);@Componentpublic static class PayOrderClientImpl implements FallbackFactory<AccountClient> {@Overridepublic AccountClient create(Throwable cause) {return new AccountClient() {@Overridepublic BigDecimal add(int userId, BigDecimal pay) {return null;}@Overridepublic 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 {@Autowiredprivate AccountMapper accountMapper;@GetMapping("/addbalance")@GlobalTransactionalpublic 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 {@Autowiredprivate PayOrderMapper payOrderMapper;@GetMapping("/add")@GlobalTransactionalpublic 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 {@Autowiredprivate 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无法传递到下一个服务,导致分布式事务失效。