SpringBoot_分布式事务

朴灿烈づ我的快乐病毒、 2021-10-06 03:54 468阅读 0赞

### SpringBoot\_分布式事务 ###

*  方案一：JTA
 *   *  依赖pom.xml
     *  配置文件application.properties
     *  配置源码
     *   *  数据源配置
         *  核心配置
         *  入口配置
         *  使用
 *  方案二：最终一致性
 *   *  流程
     *  技术支持

分布式事物解决方案：  
1、两段提交协议（2PC）- JTA  
2、事务补偿（TCC）  
3、消息队列实现最终一致性

为解决分布式系统的数据一致性问题出现了两阶段提交协议（2 Phase Commitment Protocol），两阶段提交由协调者和参与者组成，共经过两个阶段和三个操作，部分关系数据库如Oracle、MySQL支持两阶段提交协议。

2PC的优点：实现强一致性，部分关系数据库支持（Oracle、MySQL等）。  
缺点：整个事务的执行需要由协调者在多个节点之间去协调，增加了事务的执行时间，性能低下。解决方案有：springboot+Atomikos or Bitronix

# 方案一：JTA #

## 依赖pom.xml ##

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jta-atomikos</artifactId>
    </dependency>

## 配置文件application.properties ##

# Mysql 1
    mysql.datasource.test1.url = jdbc:mysql://localhost:3306/test1?useUnicode=true&characterEncoding=utf-8
    mysql.datasource.test1.username = root
    mysql.datasource.test1.password = 123456
    mysql.datasource.test1.minPoolSize = 3
    mysql.datasource.test1.maxPoolSize = 25
    mysql.datasource.test1.maxLifetime = 20000
    mysql.datasource.test1.borrowConnectionTimeout = 30
    mysql.datasource.test1.loginTimeout = 30
    mysql.datasource.test1.maintenanceInterval = 60
    mysql.datasource.test1.maxIdleTime = 60
     
    # Mysql 2
    mysql.datasource.test2.url =jdbc:mysql://localhost:3306/test2?useUnicode=true&characterEncoding=utf-8
    mysql.datasource.test2.username =root
    mysql.datasource.test2.password =123456
    mysql.datasource.test2.minPoolSize = 3
    mysql.datasource.test2.maxPoolSize = 25
    mysql.datasource.test2.maxLifetime = 20000
    mysql.datasource.test2.borrowConnectionTimeout = 30
    mysql.datasource.test2.loginTimeout = 30
    mysql.datasource.test2.maintenanceInterval = 60
    mysql.datasource.test2.maxIdleTime = 60

## 配置源码 ##

### 数据源配置 ###

@Data
    /**
     * 将application.properties配置文件中配置自动封装到实体类字段中
     * @author Administrator
     */
    @ConfigurationProperties(prefix = "mysql.datasource.test1") // 注意这个前缀要和application.properties文件的前缀一样
    public class DBConfig1 {
     
    	private String url;
    	// 比如这个url在properties中是这样子的mysql.datasource.test1.username = root
    	private String username;
    	private String password;
    	private int minPoolSize;
    	private int maxPoolSize;
    	private int maxLifetime;
    	private int borrowConnectionTimeout;
    	private int loginTimeout;
    	private int maintenanceInterval;
    	private int maxIdleTime;
    	private String testQuery;
    }
    
    @Data
    /**
     * 将application.properties配置文件中配置自动封装到实体类字段中
     * @author Administrator
     */
    @ConfigurationProperties(prefix = "mysql.datasource.test2")// 注意这个前缀要和application.properties文件的前缀一样
    public class DBConfig2 {
     
    	private String url;
    	// 比如这个url在properties中是这样子的mysql.datasource.test1.username = root
    	private String username;
    	private String password;
    	private int minPoolSize;
    	private int maxPoolSize;
    	private int maxLifetime;
    	private int borrowConnectionTimeout;
    	private int loginTimeout;
    	private int maintenanceInterval;
    	private int maxIdleTime;
    	private String testQuery;
    }

### 核心配置 ###

@Configuration
    // basePackages 最好分开配置 如果放在同一个文件夹可能会报错
    @MapperScan(basePackages = "czs.mapper1", sqlSessionTemplateRef = "testSqlSessionTemplate")
    public class MyBatisConfig1 {
     
    	// 配置数据源
    	@Bean(name = "testDataSource")
    	public DataSource testDataSource(DBConfig1 testConfig) throws SQLException {
    		MysqlXADataSource mysqlXaDataSource = new MysqlXADataSource();
    		mysqlXaDataSource.setUrl(testConfig.getUrl());
    		mysqlXaDataSource.setPinGlobalTxToPhysicalConnection(true);
    		mysqlXaDataSource.setPassword(testConfig.getPassword());
    		mysqlXaDataSource.setUser(testConfig.getUsername());
    		mysqlXaDataSource.setPinGlobalTxToPhysicalConnection(true);
     
    		// 将本地事务注册到创 Atomikos全局事务
    		AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
    		xaDataSource.setXaDataSource(mysqlXaDataSource);
    		xaDataSource.setUniqueResourceName("testDataSource");
     
    		xaDataSource.setMinPoolSize(testConfig.getMinPoolSize());
    		xaDataSource.setMaxPoolSize(testConfig.getMaxPoolSize());
    		xaDataSource.setMaxLifetime(testConfig.getMaxLifetime());
    		xaDataSource.setBorrowConnectionTimeout(testConfig.getBorrowConnectionTimeout());
    		xaDataSource.setLoginTimeout(testConfig.getLoginTimeout());
    		xaDataSource.setMaintenanceInterval(testConfig.getMaintenanceInterval());
    		xaDataSource.setMaxIdleTime(testConfig.getMaxIdleTime());
    		xaDataSource.setTestQuery(testConfig.getTestQuery());
    		return xaDataSource;
    	}
     
    	@Bean(name = "testSqlSessionFactory")
    	public SqlSessionFactory testSqlSessionFactory(@Qualifier("testDataSource") DataSource dataSource)
    			throws Exception {
    		SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
    		bean.setDataSource(dataSource);
    		return bean.getObject();
    	}
     
    	@Bean(name = "testSqlSessionTemplate")
    	public SqlSessionTemplate testSqlSessionTemplate(
    			@Qualifier("testSqlSessionFactory") SqlSessionFactory sqlSessionFactory) throws Exception {
    		return new SqlSessionTemplate(sqlSessionFactory);
    	}
    }
    
    @Configuration
    @MapperScan(basePackages = "czs.mapper2", sqlSessionTemplateRef = "test2SqlSessionTemplate")
    public class MyBatisConfig2 {
     
    	// 配置数据源
    	@Bean(name = "test2DataSource")
    	public DataSource testDataSource(DBConfig2 testConfig) throws SQLException {
    		MysqlXADataSource mysqlXaDataSource = new MysqlXADataSource();
    		mysqlXaDataSource.setUrl(testConfig.getUrl());
    		mysqlXaDataSource.setPinGlobalTxToPhysicalConnection(true);
    		mysqlXaDataSource.setPassword(testConfig.getPassword());
    		mysqlXaDataSource.setUser(testConfig.getUsername());
    		mysqlXaDataSource.setPinGlobalTxToPhysicalConnection(true);
     
    		AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
    		xaDataSource.setXaDataSource(mysqlXaDataSource);
    		xaDataSource.setUniqueResourceName("test2DataSource");
     
    		xaDataSource.setMinPoolSize(testConfig.getMinPoolSize());
    		xaDataSource.setMaxPoolSize(testConfig.getMaxPoolSize());
    		xaDataSource.setMaxLifetime(testConfig.getMaxLifetime());
    		xaDataSource.setBorrowConnectionTimeout(testConfig.getBorrowConnectionTimeout());
    		xaDataSource.setLoginTimeout(testConfig.getLoginTimeout());
    		xaDataSource.setMaintenanceInterval(testConfig.getMaintenanceInterval());
    		xaDataSource.setMaxIdleTime(testConfig.getMaxIdleTime());
    		xaDataSource.setTestQuery(testConfig.getTestQuery());
    		return xaDataSource;
    	}
     
    	@Bean(name = "test2SqlSessionFactory")
    	public SqlSessionFactory testSqlSessionFactory(@Qualifier("test2DataSource") DataSource dataSource)
    			throws Exception {
    		SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
    		bean.setDataSource(dataSource);
    		return bean.getObject();
    	}
     
    	@Bean(name = "test2SqlSessionTemplate")
    	public SqlSessionTemplate testSqlSessionTemplate(
    			@Qualifier("test2SqlSessionFactory") SqlSessionFactory sqlSessionFactory) throws Exception {
    		return new SqlSessionTemplate(sqlSessionFactory);
    	}
    }

### 入口配置 ###

添加配置注解扫描  
1、Mybatis

@EnableConfigurationProperties(value={DBConfig1.class,DBConfig2.class})

2、JPA

@EnableJpaRepositories(basePackages ={"cn.mywork.dao","cn.mywork.user1.dao","cn.mywork.user2.dao"})
    @EntityScan("cn.mywork.entity")
    @EnableConfigurationProperties(value={DBConfig1.class,DBConfig2.class})

### 使用 ###

@Service
    public class ManyService1 {
     
    	@Autowired
    	private UserMapper1 userMapper1;
     
    	@Autowired
    	private UserMapper2 userMapper2;
     
    	// 开启事务，由于使用jta+atomikos解决分布式事务，所以此处不必再指定事务
    	@Transactional
    	public int insert(String name, Integer age) {
    		int insert = userMapper1.insert(name, age);
    		int i = 1 / age;// 赋值age为0故意引发事务
    		return insert;
    	}
     
    	// 开启事务，由于使用jta+atomikos解决分布式事务，所以此处不必再指定事务
    	@Transactional
    	public int insertDb1AndDb2(String name, Integer age) {
    		int insert = userMapper1.insert(name, age);
    		int insert2 = userMapper2.insert(name, age);
    		int i = 1 / age;// 赋值age为0故意引发事务
    		return insert + insert2;
    	}
     
    }

# 方案二：最终一致性 #

## 流程 ##

1、订单服务和库存服务完成检查和预留资源。  
2、订单服务在本地事务中完成添加订单表记录和添加“减少库存任务消息”。  
3、由定时任务根据消息表的记录发送给MQ通知库存服务执行减库存操作。  
4、库存服务执行减少库存，并且记录执行消息状态（为避免重复执行消息，在执行减库存之前查询是否执行过此消息）。  
5、库存服务向MQ发送完成减少库存的消息。  
6、订单服务接收到完成库存减少的消息后删除原来添加的“减少库存任务消息”。

实现最终事务一致要求：预留资源成功理论上要求正式执行成功，如果执行失败会进行重试，要求业务执行方法实现幂等。

优点 ：  
由MQ按异步的方式协调完成事务，性能较高。  
不用实现try/confirm/cancel接口，开发成本比TCC低。  
缺点：  
此方式基于关系数据库本地事务来实现，会出现频繁读写数据库记录，浪费数据库资源，另外对于高并发操作不是最佳方案。

## 技术支持 ##

消息队列、定时任务