分布式事务
1、什么是分布式事务，分布式事务与本地事务的区别？
本地事务，就是只对本项目模块事务进行管理的，对于其他的数据库事务不做任何的改变，在项目编写中，经常使用@Transactional(rollBackFor=Exception.class)实现当前方法如果出现任何的异常就会使用数据库的redo.log日志实现事务的回滚。
分布式事务，它是在全局的角度看待问题，在多个模块之间发挥自己的作用，如果其中的一个模块出现了问题，那么就会将涉及到的多个模块的事务都会进行回滚，他是跨数据库的，跨项目模块的。
2、分布式事务的解决方案？
分布式事务的解决方案可以使用Seata实现。
什么是Seata？
Seata是一个开源的分布式事务的解决方案，它主要提供一个高性能和简单实用的分布式事务服务。
Seata为用户提供了四种模式：AT、TCC、SAGA、XA模式。
后面。将主要使用XA模式实现分布式事务问题。
3、实战案例一（微服务系统，不使用分布式事务）
第一、创建数据库bank1
CREATE DATABASE BANK1; CREATE TABLE `account_info` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `account_name` varchar(100) COLLATE utf8_bin DEFAULT NULL COMMENT '户主姓名', `account_no` varchar(100) COLLATE utf8_bin DEFAULT NULL COMMENT '银行卡号', `account_password` varchar(100) COLLATE utf8_bin DEFAULT NULL COMMENT '帐户密码', `account_balance` double DEFAULT NULL COMMENT '帐户余额', PRIMARY KEY (`id`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 COLLATE=utf8_bin ROW_FORMAT=DYNAMIC;第二、创建数据库bank2
CREATE DATABASE BANK2; CREATE TABLE `account_info` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `account_name` varchar(100) COLLATE utf8_bin DEFAULT NULL COMMENT '户主姓名', `account_no` varchar(100) COLLATE utf8_bin DEFAULT NULL COMMENT '银行卡号', `account_password` varchar(100) COLLATE utf8_bin DEFAULT NULL COMMENT '帐户密码', `account_balance` double DEFAULT NULL COMMENT '帐户余额', PRIMARY KEY (`id`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 COLLATE=utf8_bin ROW_FORMAT=DYNAMIC;第三、分别向数据库bank1和bank2插入数据
BANK1: INSERT INTO ACCOUNT_INFO VALUES(2,"张三",1,123456,1000); BANK2: INSERT INTO ACCOUNT_INFO VALUES(3,"李四",2,123456,0);第四、搭建项目环境（BANK1）
pom依赖
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.hui</groupId> <artifactId>bank1</artifactId> <version>1.0-SNAPSHOT</version> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.6.3</version> </parent> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <!-- web --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- mysql-connector --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.28</version> </dependency> <!-- mybatis-plus --> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-boot-starter</artifactId> <version>3.3.0</version> </dependency> <!-- nacos --> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> <version>2021.0.1.0</version> </dependency> <!-- open-feign --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-openfeign</artifactId> <version>3.1.0</version> </dependency> <!-- lombok --> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.22</version> </dependency> <!-- loadbanancer --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-loadbalancer</artifactId> <version>3.1.2</version> </dependency> <!-- 逆向工程数据库 --> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-generator</artifactId> <version>3.5.2</version> </dependency> <dependency> <groupId>org.apache.velocity</groupId> <artifactId>velocity-engine-core</artifactId> <version>2.0</version> </dependency> </dependencies> </project>Controller
/**
* @Description: 用户转账操作
* @Author: huidou 惠豆
* @CreateTime: 2022/6/14 23:19
*/
@Controller
@RequestMapping("/bank1")
public class AccountInfoController {
@Autowired
private IAccountInfoService iAccountInfoService;
/**
* 向李四赚钱
* @param accountNo
* @param amount
* @return
*/
@ResponseBody
@RequestMapping("/transfer")
public String account(String accountNo, Double amount) {
Integer accountBalance = iAccountInfoService.updateAccountBalance(accountNo, amount);
if (accountBalance >= 1) {
return "转账成功，向李四转账成功";
} else {
return "转账失败，没能向李四转账成功";
}
}
}
service
public interface IAccountInfoService extends IService<AccountInfo> {
/**
* 向李四转账
* @param accountNo 银行卡号
* @param amount 金额
*/
Integer updateAccountBalance(String accountNo, Double amount);
}
@Slf4j
@Service
public class AccountInfoServiceImpl extends ServiceImpl<AccountInfoMapper, AccountInfo> implements IAccountInfoService {
@Autowired
private AccountInfoMapper accountInfoMapper;
// 远程服务接口
@Autowired
private Bank2Client bank2Client;
/**
* 向李四转账
* 从异常捕获来分析，bank1和bank2都设置了本地事务，这是后就会出现两种情况：
* 1、bank1在调用bank2接口之后出现了异常。
* 2、bank1正常调用bank2，但是bank2出现了异常。
*
* 结果分别是：
* 第一、bank1出现了异常会根据本地事务的特点，遇见异常就进行回滚，但是bank2是正常的，就会出现bank1没有扣钱，但是bank2已经成功的加钱了。
* 第二、bank1没有出现问题，但是bank2出现了问题，由于bank1调用bank2，benk2出现了Exception就会抛给bank1，所以最后bank1和bank2都会根据本地事务进行回滚。
* @param accountNo 银行卡号
* @param amount 金额
*/
@Override
@Transactional(rollbackFor = Exception.class)
public Integer updateAccountBalance(String accountNo, Double amount) {
// 1、首先查询自己的账户信息
AccountInfo accountInfo = accountInfoMapper.selectById(2);
if (accountInfo == null) {
return 0;
}
// 2、判断自己的余额信息
if (accountInfo.getAccountBalance() < amount) {
return 0;
}
// 3、修改自己的余额信息
accountInfo.setAccountBalance(accountInfo.getAccountBalance() - amount);
int update = accountInfoMapper.updateById(accountInfo);
// 4、测试以下远程调用的异常会不会引起当前方法的异常
try {
bank2Client.transfer(accountNo,amount);
} catch (Exception e) {
System.out.println("bank2出现错误了");
System.out.println(e.getMessage());
}
return 1;
}
}
mapper
@Repository
public interface AccountInfoMapper extends BaseMapper<AccountInfo> {
}
pojo
@Data
@TableName("account_info")
public class AccountInfo implements Serializable {
private static final long serialVersionUID = 1L;
@TableId(value = "id", type = IdType.AUTO)
private Long id;
private String accountName;
private String accountNo;
private String accountPassword;
private Double accountBalance;
}
openfeign
/**
* @Description: openfeign 远程调用服务接口
* @Author: huidou 惠豆
* @CreateTime: 2022/6/14 23:49
*/
@Component
@FeignClient(value = "seata-bank2")
public interface Bank2Client {
// 调用bank2的服务接口，实现向李四转账
@RequestMapping("/bank2/transfer")
String transfer(@RequestParam("accountNo") String accountNo,@RequestParam("amount") Double amount);
}
搭建项目环境（BANK2）
pom
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.hui</groupId> <artifactId>bank2</artifactId> <version>1.0-SNAPSHOT</version> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.6.3</version> </parent> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <!-- web --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- mysql-connector --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.28</version> </dependency> <!-- mybatis-plus --> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-boot-starter</artifactId> <version>3.3.0</version> </dependency> <!-- nacos --> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> <version>2021.0.1.0</version> </dependency> <!-- lombok --> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.22</version> </dependency> <!-- loadbanaer --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-loadbalancer</artifactId> <version>3.1.2</version> </dependency> <!-- 逆向工程数据库 --> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-generator</artifactId> <version>3.5.2</version> </dependency> <dependency> <groupId>org.apache.velocity</groupId> <artifactId>velocity-engine-core</artifactId> <version>2.0</version> </dependency> </dependencies> </project>controller
/**
* @Description: 用户转账操作
* @Author: huidou 惠豆
* @CreateTime: 2022/6/14 23:19
*/
@Controller
@RequestMapping("/bank2")
public class AccountInfoController {
@Autowired
private IAccountInfoService iAccountInfoService;
/**
* 接收来自张三的钱
* @param accountNo
* @param amount
* @return
*/
@ResponseBody
@RequestMapping("/transfer")
public String account(String accountNo, Double amount) {
iAccountInfoService.updateAccountBalance(accountNo, amount);
return "转账成功，接收来自张三的钱";
}
}
service
public interface IAccountInfoService extends IService<AccountInfo> {
/**
* 李四账户增加金额
* @param accountNo 银行卡号
* @param amount 金额
*/
void updateAccountBalance(String accountNo, Double amount);
}
@Slf4j
@Service
public class AccountInfoServiceImpl extends ServiceImpl<AccountInfoMapper, AccountInfo> implements IAccountInfoService {
@Autowired
private AccountInfoMapper accountInfoMapper;
/**
* 李四增加金额
* @param accountNo 银行卡号
* @param amount 金额
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void updateAccountBalance(String accountNo, Double amount) {
try {
System.out.println(10/0);
// 1、根据账户ID，查询对应的账号信息（目前只有一人，就给死ID）
QueryWrapper<AccountInfo> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("account_no",accountNo);
AccountInfo accountInfo = accountInfoMapper.selectOne(queryWrapper);
// 2、判断该账户是不是为Null
if (accountInfo == null) {
return;
}
// 3、给该账户添加钱（来自张三的）
accountInfo.setAccountBalance(accountInfo.getAccountBalance() + amount);
// 4、更新数据
int updateById = accountInfoMapper.updateById(accountInfo);
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException("出现了10/0的异常");
}
}
}
pojo
@Data
@TableName("account_info")
public class AccountInfo implements Serializable {
private static final long serialVersionUID = 1L;
@TableId(value = "id", type = IdType.AUTO)
private Long id;
private String accountName;
private String accountNo;
private String accountPassword;
private Double accountBalance;
}
项目的配置文件
server: port: 6001 spring: application: name: seata-bank1 cloud: nacos: discovery: server-addr: 192.168.205.150:8848 datasource: username: root password: root url: jdbc:mysql://localhost:3306/bank1?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC driver-class-name: com.mysql.cj.jdbc.Driver server: port: 6002 spring: application: name: seata-bank2 cloud: nacos: discovery: server-addr: 192.168.205.150:8848 datasource: username: root password: root url: jdbc:mysql://localhost:3306/bank2?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC driver-class-name: com.mysql.cj.jdbc.DriverNacos服务列表
-
项目测试
1、两个项目中都不加@Transactional(rollbackFor = Exception.class)，逻辑正常调用。
2、两个项目中都加上@Transactional(rollbackFor = Exception.class)，逻辑正常调用。
3、两个项目中都加上@Transactional(rollbackFor = Exception.class)，bank1在调用bank2接口之后设置异常查看是否出现数据不一致（bank2逻辑正常）。
4、两个项目中都加上@Transactional(rollbackFor = Exception.class)，bank1在调用bank2接口之后（bank1逻辑正常），但是bank2逻辑设置错误。
结果：
1、对于第一种情况，逻辑是正确的，数据符合一致性。
2、对于第二种情况，逻辑是正确的，数据符合一致性。
3、对于第三种情况，由于bank1在调用完bank2之后出现了异常，bank2的执行逻辑正确，再加上都有
@Transactional(rollbackFor = Exception.class)注解，所以导致bank1数据回滚，但是bank2数据还是修改了，因为这个注解是本地注解，不能管到其他的事务模块中。
4、对于第四种情况容易出现误解，由于bank1的执行逻辑是正确的，但是bank2的逻辑是错误的，是会报出异常的，加上bank1调用了bank2，这时bank1就会接收到bank2抛出的异常，由于两个接口都有@Transactional(rollbackFor = Exception.class)注解，所以最后都会进行回滚，这个时候，不清楚异常接收的原理的时候，还以为两个本地事务注解就能实现分布式事务，其实不是的，要真是，那还要分布式事务的理由在那里，第三种情况出现了又该如何处理？
4、实战案例二（微服务系统，使用分布式事务XA模式）
第一、添加Seata依赖<!-- seata --> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-seata</artifactId> <version>2021.0.1.0</version> </dependency>第二、数据库添加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;第三、BANK1（在业务层调用BANK2接口之后设置异常）
service
@Slf4j
@Service
public class AccountInfoServiceImpl extends ServiceImpl<AccountInfoMapper, AccountInfo> implements IAccountInfoService {
@Autowired
private AccountInfoMapper accountInfoMapper;
// 远程服务接口
@Autowired
private Bank2Client bank2Client;
/**
* 向李四转账
* 从异常捕获来分析，bank1和bank2都设置了本地事务，这是后就会出现两种情况：
* 1、bank1在调用bank2接口之后出现了异常。
* 2、bank1正常调用bank2，但是bank2出现了异常。
*
* 结果分别是：
* 第一、bank1出现了异常会根据本地事务的特点，遇见异常就进行回滚，但是bank2是正常的，就会出现bank1没有扣钱，但是bank2已经成功的加钱了。
* 第二、bank1没有出现问题，但是bank2出现了问题，由于bank1调用bank2，benk2出现了Exception就会抛给bank1，所以最后bank1和bank2都会根据本地事务进行回滚。
* @param accountNo 银行卡号
* @param amount 金额
*/
@GlobalTransactional // 这个注解只需要在其中的一个模块方法上添加即可，因为是全局，所以一个就够了
@Override
@Transactional(rollbackFor = Exception.class)
public Integer updateAccountBalance(String accountNo, Double amount) {
// 1、首先查询自己的账户信息
AccountInfo accountInfo = accountInfoMapper.selectById(2);
if (accountInfo == null) {
return 0;
}
// 2、判断自己的余额信息
if (accountInfo.getAccountBalance() < amount) {
return 0;
}
// 3、修改自己的余额信息
accountInfo.setAccountBalance(accountInfo.getAccountBalance() - amount);
int update = accountInfoMapper.updateById(accountInfo);
// 4、测试以下远程调用的异常会不会引起当前方法的异常
bank2Client.transfer(accountNo,amount);
/**
* 在这里设置错误，检测是否出现不一致性
* 错误情况就是：bank1数据回滚，不改变，但是bank2数据改变
* 正常情况就是：出现异常，大家都不改变。
*/
System.out.println(10/0);
return 1;
}
}
第四、BANK2（业务层逻辑正确）
service
@Slf4j
@Service
public class AccountInfoServiceImpl extends ServiceImpl<AccountInfoMapper, AccountInfo> implements IAccountInfoService {
@Autowired
private AccountInfoMapper accountInfoMapper;
/**
* 李四增加金额
* @param accountNo 银行卡号
* @param amount 金额
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void updateAccountBalance(String accountNo, Double amount) {
// 1、根据账户ID，查询对应的账号信息（目前只有一人，就给死ID）
QueryWrapper<AccountInfo> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("account_no",accountNo);
AccountInfo accountInfo = accountInfoMapper.selectOne(queryWrapper);
// 2、判断该账户是不是为Null
if (accountInfo == null) {
return;
}
// 3、给该账户添加钱（来自张三的）
accountInfo.setAccountBalance(accountInfo.getAccountBalance() + amount);
// 4、更新数据
int updateById = accountInfoMapper.updateById(accountInfo);
}
}
结果：
数据库初始数据图：
数据库执行之后的结果还是一样的，所以说明分布式事务Seata生效了。
5、使用分布式事务解决的原理
分布式事务的核心思想其实就是一个投票的过程，对于一个决定，如果没有一个人有问题，那么就会执行，反之，只要有一个人有问题，就会进行回退取消。
从上面的图可以看出，当用户服务的事务管理器TM会向全局事务控制申请开启分布式事务，这时会产生一个全局唯一的XID，然后，用户服务会向全局事务进行注册自己的本地事务，注册完之后，会进行执行自己的逻辑，之后在条用积分服务，积分服务也是一样的会进行注册本地事务，由于两个服务都有自己的本地服务，倘若都成功了，就都会向TC提交一个投票也就是Commit，这时TC就会向两个本地事务发送Commit的命令；但是若有一个出现了问题那么@Transactional(rollbackFor = Exception.class)注解就会进行回滚，由于TC管辖了该本地事务，就会知道该事务想自己提交了rollback的请求，这时TC就会向所有的本地事务发送rollback的命令。
6、基于Atomikos实现XA强一致性分布式事务实战
是哦也能够Atomikos实现XA模式的强一致性，模拟实现跨库的转账实现，在一个项目模块中，使用多数据源（两个数据库）实现不同数据库的数据改变。
原理图如下：
使用Atomikos框架技术的好处在于：项目不需要直接与数据库进行交互进行数据的改变，而是让Atomikos技术对数据库进行封装，由于这个Atomikos技术内部实现了分布式事务的XA模式，所以对于开发人员来说，我不需要像之前那样不但要添加依赖，添加注解，还要与本地事务做出关联，现在就不需要了，只需要简简单单的添加依赖与注解即可，注解也是常用的@Transactional即可，只需要关注业务逻辑实现。
开始一个项目实战案例：
项目需求：在一个项目中，配置多个数据源，用于实现转账功能，一个数据库扣减金额；一个数据库添加金额
第一、创建数据库CREATE DATABASE TX-XA-01 CREATE TABLE `user_account` ( `account_no` varchar(64) COLLATE utf8_bin NOT NULL COMMENT '账户编号', `account_name` varchar(64) COLLATE utf8_bin DEFAULT NULL COMMENT '账户\r\n名称', `account_balance` decimal(10,2) DEFAULT NULL COMMENT '账户余额', PRIMARY KEY (`account_no`) USING BTREE ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin ROW_FORMAT=DYNAMIC;
CREATE DATABASE TX-XA-02 CREATE TABLE `user_account` ( `account_no` varchar(64) COLLATE utf8_bin NOT NULL COMMENT '账户编号', `account_name` varchar(64) COLLATE utf8_bin DEFAULT NULL COMMENT '账户\r\n名称', `account_balance` decimal(10,2) DEFAULT NULL COMMENT '账户余额', PRIMARY KEY (`account_no`) USING BTREE ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin ROW_FORMAT=DYNAMIC;第二、向数据库插入数据
CREATE DATABASE TX-XA-01
INSERT INTO USER_ACCOUNT VALUES("1001","张三",10000);
CREATE DATABASE TX-XA-02
INSERT INTO USER_ACCOUNT VALUES("1002","李四",10000);
第三、创建项目，配置项目环境
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.hui</groupId> <artifactId>atomikos-xa</artifactId> <version>1.0-SNAPSHOT</version> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.6.3</version> </parent> <dependencies> <!-- web --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- durid数据源 --> <dependency> <groupId>com.alibaba</groupId> <artifactId>druid-spring-boot-starter</artifactId> <version>1.1.22</version> </dependency> <!-- mysql-connector --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> </dependency> <!-- mybatis-plus --> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-boot-starter</artifactId> <version>3.3.0</version> </dependency> <!-- atomikos --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-jta-atomikos</artifactId> </dependency> <!-- lombok --> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <!-- 逆向工程数据库 --> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-generator</artifactId> <version>3.5.2</version> </dependency> <dependency> <groupId>org.apache.velocity</groupId> <artifactId>velocity-engine-core</artifactId> <version>2.0</version> </dependency> </dependencies> </project>第四、多数据源配置
DBConfig1
/**
* @Description: 第一个数据库信息的绑定
* @Author: huidou 惠豆
* @CreateTime: 2022/6/15 17:09
*/
@Data
@ConfigurationProperties(prefix = "spring.datasource.dynamic.datasource.master")
public class DBConfig1 {
private String url;
private String username;
private String password;
private String driverClassName;
}
DBConfig2
/**
* @Description: 第二个数据库信息的绑定
* @Author: huidou 惠豆
* @CreateTime: 2022/6/15 17:09
*/
@Data
@ConfigurationProperties(prefix = "spring.datasource.dynamic.datasource.slave")
public class DBConfig2 {
private String url;
private String username;
private String password;
private String driverClassName;
}
MybatisConfig1、MybatisConfig2
/**
* @Description: 配置第一个数据源，将其与Atomikos进行整合
* @Author: huidou 惠豆
* @CreateTime: 2022/6/15 17:09
*/
@Configuration
@MapperScan(basePackages = "com.hui.mapper1",sqlSessionTemplateRef = "masterSqlSessionTemplate")
public class MybatisConfig1 {
/**
* 创建一个DataSource数据源
* @param dbConfig1
* @return
*/
@Bean(name = "masterDataSource")
public DataSource getDataSource(DBConfig1 dbConfig1) {
// 1. 将本地事务注册到 Atomikos全局事务中
AtomikosDataSourceBean sourceBean = new AtomikosDataSourceBean();
// 2. 设置rm供应商
sourceBean.setUniqueResourceName("masterDataSource");
sourceBean.setXaDataSourceClassName(dbConfig1.getDriverClassName());
// 3. 设置testquery
sourceBean.setTestQuery("select 1");
// 4. 设置超时时间
sourceBean.setBorrowConnectionTimeout(3);
// 设置Mysql链接
MysqlXADataSource dataSource = new MysqlXADataSource();
dataSource.setUrl(dbConfig1.getUrl());
dataSource.setUser(dbConfig1.getUsername());
dataSource.setPassword(dbConfig1.getPassword());
sourceBean.setXaDataSource(dataSource);
return sourceBean;
}
/**
* SqlSessionFactory是mybaits 重要的对象
* @return
*/
@Bean(name = "masterSqlSessionFactory")
public SqlSessionFactory getSqlSessionFactory(@Qualifier("masterDataSource") DataSource dataSource) throws Exception {
MybatisSqlSessionFactoryBean sessionFactoryBean = new MybatisSqlSessionFactoryBean();
sessionFactoryBean.setDataSource(dataSource);
return sessionFactoryBean.getObject();
}
/**
* 负责管理mybatis 的sqlsession sql
* SqlSessionTemplate 替换默认的mybaits 实现的defalutsqlsession不能参与spring事务不能注入 线程不安全
* @return
*/
@Bean(name = "masterSqlSessionTemplate")
public SqlSessionTemplate getSqlSessionTemplate(@Qualifier("masterSqlSessionFactory") SqlSessionFactory sqlSessionFactory){
return new SqlSessionTemplate(sqlSessionFactory);
}
}
/**
* @Description: 配置第二个数据源，将其与Atomikos进行整合
* @Author: huidou 惠豆
* @CreateTime: 2022/6/15 17:09
*/
@Configuration
@MapperScan(basePackages = "com.hui.mapper2",sqlSessionTemplateRef = "slaveSqlSessionTemplate")
public class MybatisConfig2 {
/**
* 创建一个DataSource数据源
* @param dbConfig2
* @return
*/
@Bean(name = "slaveDataSource")
public DataSource getDataSource(DBConfig2 dbConfig2) {
// 1. 将本地事务注册到 Atomikos全局事务中
AtomikosDataSourceBean sourceBean = new AtomikosDataSourceBean();
// 2. 设置rm供应商
sourceBean.setUniqueResourceName("slaveDataSource");
sourceBean.setXaDataSourceClassName(dbConfig2.getDriverClassName());
// 3. 设置testquery
sourceBean.setTestQuery("select 1");
// 4. 设置超时时间
sourceBean.setBorrowConnectionTimeout(3);
// 设置Mysql链接
MysqlXADataSource dataSource = new MysqlXADataSource();
dataSource.setUrl(dbConfig2.getUrl());
dataSource.setUser(dbConfig2.getUsername());
dataSource.setPassword(dbConfig2.getPassword());
sourceBean.setXaDataSource(dataSource);
return sourceBean;
}
/**
* SqlSessionFactory是mybaits 重要的对象
* @return
*/
@Bean(name = "slaveSqlSessionFactory")
public SqlSessionFactory getSqlSessionFactory(@Qualifier("slaveDataSource") DataSource dataSource) throws Exception {
MybatisSqlSessionFactoryBean sessionFactoryBean = new MybatisSqlSessionFactoryBean();
sessionFactoryBean.setDataSource(dataSource);
return sessionFactoryBean.getObject();
}
/**
* 负责管理mybatis 的sqlsession sql
* SqlSessionTemplate 替换默认的mybaits 实现的defalutsqlsession不能参与spring事务不能注入 线程不安全
* @return
*/
@Bean(name = "slaveSqlSessionTemplate")
public SqlSessionTemplate getSqlSessionTemplate(@Qualifier("slaveSqlSessionFactory") SqlSessionFactory sqlSessionFactory){
return new SqlSessionTemplate(sqlSessionFactory);
}
}
逻辑层实现
controller
/**
* 实现转账操作
*/
@RestController
@RequestMapping("/userAccount")
public class UserAccountController {
@Autowired
private IUserAccountService iUserAccountService;
/**
* 实现转账操作
* @param sourceAccountNo 源账户
* @param targetAccountNo 目标账户
* @param amount 转账金额
* @return
*/
@RequestMapping("/transfer")
public String transfer(String sourceAccountNo, String targetAccountNo, BigDecimal amount) {
Integer transfer = iUserAccountService.transfer(sourceAccountNo, targetAccountNo, amount);
if (transfer == 1) {
return "转账成功！";
}
return "转账失败！";
}
}
service
@Service
public class UserAccountServiceImpl implements IUserAccountService {
// 原账户所在数据库
@Autowired
private UserAccountMapper1 userAccountMapper1;
// 目标账户所在数据库
@Autowired
private UserAccountMapper2 userAccountMapper2;
/**
* 实现转账操作
* @param sourceAccountNo 源账户
* @param targetAccountNo 目标账户
* @param amount 转账金额
* @return
*/
@Override
public Integer transfer(String sourceAccountNo, String targetAccountNo, BigDecimal amount) {
// 1、查询原账户
UserAccount sourceAccount = userAccountMapper1.selectById(sourceAccountNo);
// 2、查询目标账户
UserAccount targetAccount = userAccountMapper2.selectById(targetAccountNo);
// 3、判断原账户与目标账户是否存在
if (sourceAccount != null && targetAccount != null) {
// 4、判断原账户余额是否不足
if (sourceAccount.getAccountBalance().compareTo(amount) < 0) {
throw new RuntimeException("原账户金额不足，无法实现转账操作");
}
// 5、原账户扣减余额，并更新
sourceAccount.setAccountBalance(sourceAccount.getAccountBalance().subtract(amount));
int updateSourceAccount = userAccountMapper1.updateById(sourceAccount);
// 6、目标账户增加余额，并更新
targetAccount.setAccountBalance(targetAccount.getAccountBalance().add(amount));
int updateTargetAccount = userAccountMapper2.updateById(targetAccount);
if (updateSourceAccount == 1 && updateTargetAccount == 1) {
return 1;
}
}
return 0;
}
}
mapper
@Repository
public interface UserAccountMapper1 extends BaseMapper<UserAccount> {}
@Repository
public interface UserAccountMapper2 extends BaseMapper<UserAccount> {}
application.yml
server:
port: 6003
spring:
autoconfigure:
#停用druid连接池的自动配置
exclude: com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceAutoConfigure
datasource:
#选用druid的XADataSource数据源，因为这个数据源支持分布式事务管理
type: com.alibaba.druid.pool.xa.DruidXADataSource
#以下是自定义字段
dynamic:
primary: master
datasource:
master:
url: jdbc:mysql://localhost:3306/tx-xa-01?useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&autoReconnect=true
username: root
password: root
driver-class-name: com.mysql.cj.jdbc.Driver
slave:
url: jdbc:mysql://localhost:3306/tx-xa-02?useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&autoReconnect=true
username: root
password: root
driver-class-name: com.mysql.cj.jdbc.Driver
validation-query: SELCET 1
logging:
pattern:
console: logging.pattern.console=%d{MM/dd HH:mm:ss.SSS} %clr(%-5level) --- [%-15thread] %cyan(%-50logger{50}):%msg%n
业务测试
发送请求：
http://localhost:6003/userAccount/transfer?sourceAccountNo=1001&targetAccountNo=1002&amount=1000
正常情况下，可以实现跨数据库对数据进行修改，但是如果我恶意做一个操作，代码如下：
/**
* 实现转账操作
* @param sourceAccountNo 源账户
* @param targetAccountNo 目标账户
* @param amount 转账金额
* @return
*/
@Override
public Integer transfer(String sourceAccountNo, String targetAccountNo, BigDecimal amount) {
// 1、查询原账户
UserAccount sourceAccount = userAccountMapper1.selectById(sourceAccountNo);
// 2、查询目标账户
UserAccount targetAccount = userAccountMapper2.selectById(targetAccountNo);
// 3、判断原账户与目标账户是否存在
if (sourceAccount != null && targetAccount != null) {
// 4、判断原账户余额是否不足
if (sourceAccount.getAccountBalance().compareTo(amount) < 0) {
throw new RuntimeException("原账户金额不足，无法实现转账操作");
}
// 5、原账户扣减余额，并更新
sourceAccount.setAccountBalance(sourceAccount.getAccountBalance().subtract(amount));
int updateSourceAccount = userAccountMapper1.updateById(sourceAccount);
System.out.println(10/0);
// 6、目标账户增加余额，并更新
targetAccount.setAccountBalance(targetAccount.getAccountBalance().add(amount));
int updateTargetAccount = userAccountMapper2.updateById(targetAccount);
if (updateSourceAccount == 1 && updateTargetAccount == 1) {
return 1;
}
}
return 0;
}
从代码上可以看出来，我们在对数据库1做完操作之后进行的一个异常，这时就会造成报错，第一个数据库扣减成功，第二个数据库因异常无法执行，则不能添加，形成数据不一致。
如何解决？
使用Atomikos实现分布式事务数据XA模式的强一致性。
代码上如何修改：很简单，由于我们之前把多数据源已经都与Atomikos框架整合在一起了，所以这个时候，只需要在方法上添加一个注解@Transactional，该注解是import org.springframework.transaction.annotation.Transactional;下的，即可，该注解就包含了Atomikos对事务的处理，然后在主启动类上添加一个注解@EnableTransactionManager，即可实现。
结果：
发送请求：
http://localhost:6003/userAccount/transfer?sourceAccountNo=1001&targetAccountNo=1002&amount=1000
实现数据一致性。
查看14道真题和解析