SpringBoot如何实现多数据源事务管理？
- 内容介绍
- 文章标签
- 相关推荐
本文共计1875个文字，预计阅读时间需要8分钟。
背景：文章介绍了Spring Boot下多数据源动态切换的整合方案，并在后续使用过程中发现事务控制中存在多种bug，决定对此问题进行分析与解决。
前提：需要了解多数据源切换的流程和事务控制的原理。
问题分析及解决：
1.问题表现：在事务控制过程中，发现多数据源切换时会出现事务不一致、数据丢失等问题。
2.原因分析：
a. 事务隔离级别设置不当：可能导致脏读、不可重复读、幻读等问题。 b. 数据源切换时机不当：在事务开始前或结束后切换数据源，导致事务无法正常提交。 c. 缺乏统一的事务管理器：不同数据源使用不同的事务管理器，导致事务控制混乱。3.解决方案：
a. 优化事务隔离级别：根据业务需求选择合适的事务隔离级别，如READ COMMITTED、REPEATABLE READ等。 b. 合理控制数据源切换时机：在事务开始前或结束后进行数据源切换，确保事务的正常执行。 c. 使用统一的事务管理器：采用统一的事务管理器，如AbstractRoutingDataSource，简化事务控制过程。 d. 增加数据源切换监控：实时监控数据源切换过程，确保切换的及时性和正确性。 e. 异常处理：完善异常处理机制，确保在事务过程中出现异常时，能够及时捕获并处理。总结：通过以上措施，可以有效解决Spring Boot下多数据源动态切换中的事务控制问题，确保系统稳定运行。
背景之前有文章提供了springboot多数据源动态注册切换的整合方案，在后续使用过程中，发现在事务控制中有多种bug发生，决定对此问题进行分析与解决
前情提要多数据源切换流程结构图如下所示，包含几个组成元素
-
自定义的数据源配置处理，通过DruidDataSource对象动态注册到系统中
-
自定义数据源标识注解与切面
-
数据源切换时的上下文线程变量持有者
-
自定义AbstractRoutingDataSource，实现数据源路由切换
在Controller加入@Transitional注解后，数据源切换会失效，只会操作主库，查询资料后解决方案是将切面的Order设置为-1使之执行顺序在事务控制拦截之前，修改后证实有效，但是后续再次切换别的库或者进行主库操作无效，拿到的connection始终是第一次切换后的库对应的连接
分析代码后发现AbstractRoutingDataSource只负责提供getConnection这一层级，但是后续对connection的操作无法跟踪，项目框架mybatis和jdbcTemplate混合使用，后续操作在spring层面对于事务/数据源/连接这三者的逻辑层面操作是相同的，jdbcTemplate代码较为简单，所以以此为切入点进一步分析
通过断点调试会发现sql语句的执行最终会落到execute方法，方法中开始就是通过DataSourceUtils.getConnection获取连接，这里就是我们需要追踪的地方，点进去发现跳转到doGetConnection方法，这里面就是我们需要分析的具体逻辑
第一行获取的ConnectionHolder就是当前事务对应的线程持有对象，因为我们知道，事务的本质就是方法内部的sql执行时对应的是同一个数据库connection，对于不同的嵌套业务方法，唯一相同的是当前线程ID一致，所以我们将connection与线程绑定就可以实现事务控制
点进getResource方法，发现dataSource是作为一个key去一个Map集合里取出对应的contextHolder
到这里我们好像发现点什么，之前对jdbcTemplatechu实例化设定数据源直接赋值自定义的DynamicDataSource，所以在事物中每次我们获取connection依据就是DynamicDataSource这个对象作为key，所以每次都会一样了！！
@Bean
public JdbcTemplate jdbcTemplate(){
JdbcTemplate jdbcTemplate = null;
try{
jdbcTemplate = new JdbcTemplate(dynamicDataSource());
}catch (Exception e){
e.printStackTrace();
}
return jdbcTemplate;
}
后续针对mybatis查找了相关资料，事务控制默认实现是SpringManagedTransaction，源码查看后发现了熟悉的DataSourceUtils.getConnection，证明我们的分析方向是正确的
解决方案 jdbcTemplate自定义操作类继承jdbcTemplate重写getDataSource，将我们获取的DataSource这个对应的key指定到实际切换库的数据源对象上即可
public class DynamicJdbcTemplate extends JdbcTemplate {
@Override
public DataSource getDataSource() {
DynamicDataSource router = (DynamicDataSource) super.getDataSource();
DataSource acuallyDataSource = router.getAcuallyDataSource();
return acuallyDataSource;
}
public DynamicJdbcTemplate(DataSource dataSource) {
super(dataSource);
}
}
public DataSource getAcuallyDataSource() {
Object lookupKey = determineCurrentLookupKey();
if (null == lookupKey) {
return this;
}
DataSource determineTargetDataSource = this.determineTargetDataSource();
return determineTargetDataSource == null ? this : determineTargetDataSource;
}
mybatis
自定义事务操作类，实现Transaction接口，替换TransitionFactory，这里的实现与网上的解决方案略有不同，网上是定义三个变量，datasource(动态数据源对象)/connection（主连接）/connections（从库连接），但是框架需要mybatis和jdbctemplate进行统一，mybatis是从connection层面控制，jdbctemplate是从datasource层面控制，所以全部使用键值对存储
public class DynamicTransaction implements Transaction {
private final DynamicDataSource dynamicDataSource;
private ConcurrentHashMap<String, DataSource> dataSources;
private ConcurrentHashMap<String, Connection> connections;
private ConcurrentHashMap<String, Boolean> autoCommits;
private ConcurrentHashMap<String, Boolean> isConnectionTransactionals;
public DynamicTransaction(DataSource dataSource) {
this.dynamicDataSource = (DynamicDataSource) dataSource;
dataSources = new ConcurrentHashMap<>();
connections = new ConcurrentHashMap<>();
autoCommits = new ConcurrentHashMap<>();
isConnectionTransactionals = new ConcurrentHashMap<>();
}
public Connection getConnection() throws SQLException {
String dataBaseID = DBContextHolder.getDataSource();
if (!dataSources.containsKey(dataBaseID)) {
DataSource dataSource = dynamicDataSource.getAcuallyDataSource();
dataSources.put(dataBaseID, dataSource);
}
if (!connections.containsKey(dataBaseID)) {
Connection connection = DataSourceUtils.getConnection(dataSources.get(dataBaseID));
connections.put(dataBaseID, connection);
}
if (!autoCommits.containsKey(dataBaseID)) {
boolean autoCommit = connections.get(dataBaseID).getAutoCommit();
autoCommits.put(dataBaseID, autoCommit);
}
if (!isConnectionTransactionals.containsKey(dataBaseID)) {
boolean isConnectionTransactional = DataSourceUtils.isConnectionTransactional(connections.get(dataBaseID), dataSources.get(dataBaseID));
isConnectionTransactionals.put(dataBaseID, isConnectionTransactional);
}
return connections.get(dataBaseID);
}
public void commit() throws SQLException {
for (String dataBaseID : connections.keySet()) {
Connection connection = connections.get(dataBaseID);
boolean isConnectionTransactional = isConnectionTransactionals.get(dataBaseID);
boolean autoCommit = autoCommits.get(dataBaseID);
if (connection != null && !isConnectionTransactional && !autoCommit) {
connection.commit();
}
}
}
public void rollback() throws SQLException {
for (String dataBaseID : connections.keySet()) {
Connection connection = connections.get(dataBaseID);
boolean isConnectionTransactional = isConnectionTransactionals.get(dataBaseID);
boolean autoCommit = autoCommits.get(dataBaseID);
if (connection != null && !isConnectionTransactional && !autoCommit) {
connection.rollback();
}
}
}
public void close() {
for (String dataBaseID : connections.keySet()) {
Connection connection = connections.get(dataBaseID);
DataSource dataSource = dataSources.get(dataBaseID);
DataSourceUtils.releaseConnection(connection, dataSource);
}
}
public Integer getTimeout() {
return null;
}
}
public class DynamicTransactionFactory extends SpringManagedTransactionFactory {
@Override
public Transaction newTransaction(DataSource dataSource, TransactionIsolationLevel level, boolean autoCommit) {
return new DynamicTransaction(dataSource);
}
}
@Bean
public SqlSessionFactory sqlSessionFactory() throws Exception {
//SpringBootExecutableJarVFS.addImplClass(SpringBootVFS.class);
final PackagesSqlSessionFactoryBean sessionFactory = new PackagesSqlSessionFactoryBean();
sessionFactory.setDataSource(dynamicDataSource());
sessionFactory.setTransactionFactory(new DynamicTransactionFactory());
sessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver()
.getResources("classpath*:mybatis/**/*Mapper.xml"));
//关闭驼峰转换，防止带下划线的字段无法映射
sessionFactory.getObject().getConfiguration().setMapUnderscoreToCamelCase(false);
return sessionFactory.getObject();
}
事务管理器
事务中库动态切换的问题解决了，但是只针对了主库事务，如果从库操作也需要事务的特性该如何操作呢，这里就需要在注册数据源时针对每个数据源手动注册一个事务管理器
主库是固定的，可以直接在配置Bean中声明masterTransitionManage并设置为默认
@Bean("masterTransactionManager")
@Primary
public DataSourceTransactionManager MasterTransactionManager() {
return new DataSourceTransactionManager(masterDataSource());
}
从库的事务管理器我们可以拿到dataSource初始化对象，然后向Spring容器注册单例对象
public static void registerSingletonBean(String beanName, Object singletonObject) {
//将applicationContext转换为ConfigurableApplicationContext
ConfigurableApplicationContext configurableApplicationContext = (ConfigurableApplicationContext) context;
//获取BeanFactory
DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) configurableApplicationContext.getAutowireCapableBeanFactory();
if(configurableApplicationContext.containsBean(beanName)) {
defaultListableBeanFactory.destroySingleton(beanName);
}
//动态注册bean.
defaultListableBeanFactory.registerSingleton(beanName, singletonObject);
}
SpringBootBeanUtil.registerSingletonBean(key + "TransactionManager", new DataSourceTransactionManager(druidDataSource));
在使用时只要对@Transitional注解指定transitionFactory名字即可
总结解决这个问题花费了三天的时间，查了很多资料和解决方案，很多都是只有参考性或者特异性的，所以还是需把握问题的核心加上部分源码的追踪，比如本文中需要清晰的认识到Transition-Connection-LocalThread三者的关联关系，才能找对排查的方向
后续实现了集成基于JMS（atomikos）的XA两段式提交的全局事务，使用DruidXADataSrouce出现了druid和atomikos两者线程池交互出现泄露的情况放弃了，给小伙伴们避个坑
本文共计1875个文字，预计阅读时间需要8分钟。
背景：文章介绍了Spring Boot下多数据源动态切换的整合方案，并在后续使用过程中发现事务控制中存在多种bug，决定对此问题进行分析与解决。
前提：需要了解多数据源切换的流程和事务控制的原理。
问题分析及解决：
1.问题表现：在事务控制过程中，发现多数据源切换时会出现事务不一致、数据丢失等问题。
2.原因分析：
a. 事务隔离级别设置不当：可能导致脏读、不可重复读、幻读等问题。 b. 数据源切换时机不当：在事务开始前或结束后切换数据源，导致事务无法正常提交。 c. 缺乏统一的事务管理器：不同数据源使用不同的事务管理器，导致事务控制混乱。3.解决方案：
a. 优化事务隔离级别：根据业务需求选择合适的事务隔离级别，如READ COMMITTED、REPEATABLE READ等。 b. 合理控制数据源切换时机：在事务开始前或结束后进行数据源切换，确保事务的正常执行。 c. 使用统一的事务管理器：采用统一的事务管理器，如AbstractRoutingDataSource，简化事务控制过程。 d. 增加数据源切换监控：实时监控数据源切换过程，确保切换的及时性和正确性。 e. 异常处理：完善异常处理机制，确保在事务过程中出现异常时，能够及时捕获并处理。总结：通过以上措施，可以有效解决Spring Boot下多数据源动态切换中的事务控制问题，确保系统稳定运行。
背景之前有文章提供了springboot多数据源动态注册切换的整合方案，在后续使用过程中，发现在事务控制中有多种bug发生，决定对此问题进行分析与解决
前情提要多数据源切换流程结构图如下所示，包含几个组成元素
-
自定义的数据源配置处理，通过DruidDataSource对象动态注册到系统中
-
自定义数据源标识注解与切面
-
数据源切换时的上下文线程变量持有者
-
自定义AbstractRoutingDataSource，实现数据源路由切换
在Controller加入@Transitional注解后，数据源切换会失效，只会操作主库，查询资料后解决方案是将切面的Order设置为-1使之执行顺序在事务控制拦截之前，修改后证实有效，但是后续再次切换别的库或者进行主库操作无效，拿到的connection始终是第一次切换后的库对应的连接
分析代码后发现AbstractRoutingDataSource只负责提供getConnection这一层级，但是后续对connection的操作无法跟踪，项目框架mybatis和jdbcTemplate混合使用，后续操作在spring层面对于事务/数据源/连接这三者的逻辑层面操作是相同的，jdbcTemplate代码较为简单，所以以此为切入点进一步分析
通过断点调试会发现sql语句的执行最终会落到execute方法，方法中开始就是通过DataSourceUtils.getConnection获取连接，这里就是我们需要追踪的地方，点进去发现跳转到doGetConnection方法，这里面就是我们需要分析的具体逻辑
第一行获取的ConnectionHolder就是当前事务对应的线程持有对象，因为我们知道，事务的本质就是方法内部的sql执行时对应的是同一个数据库connection，对于不同的嵌套业务方法，唯一相同的是当前线程ID一致，所以我们将connection与线程绑定就可以实现事务控制
点进getResource方法，发现dataSource是作为一个key去一个Map集合里取出对应的contextHolder
到这里我们好像发现点什么，之前对jdbcTemplatechu实例化设定数据源直接赋值自定义的DynamicDataSource，所以在事物中每次我们获取connection依据就是DynamicDataSource这个对象作为key，所以每次都会一样了！！
@Bean
public JdbcTemplate jdbcTemplate(){
JdbcTemplate jdbcTemplate = null;
try{
jdbcTemplate = new JdbcTemplate(dynamicDataSource());
}catch (Exception e){
e.printStackTrace();
}
return jdbcTemplate;
}
后续针对mybatis查找了相关资料，事务控制默认实现是SpringManagedTransaction，源码查看后发现了熟悉的DataSourceUtils.getConnection，证明我们的分析方向是正确的
解决方案 jdbcTemplate自定义操作类继承jdbcTemplate重写getDataSource，将我们获取的DataSource这个对应的key指定到实际切换库的数据源对象上即可
public class DynamicJdbcTemplate extends JdbcTemplate {
@Override
public DataSource getDataSource() {
DynamicDataSource router = (DynamicDataSource) super.getDataSource();
DataSource acuallyDataSource = router.getAcuallyDataSource();
return acuallyDataSource;
}
public DynamicJdbcTemplate(DataSource dataSource) {
super(dataSource);
}
}
public DataSource getAcuallyDataSource() {
Object lookupKey = determineCurrentLookupKey();
if (null == lookupKey) {
return this;
}
DataSource determineTargetDataSource = this.determineTargetDataSource();
return determineTargetDataSource == null ? this : determineTargetDataSource;
}
mybatis
自定义事务操作类，实现Transaction接口，替换TransitionFactory，这里的实现与网上的解决方案略有不同，网上是定义三个变量，datasource(动态数据源对象)/connection（主连接）/connections（从库连接），但是框架需要mybatis和jdbctemplate进行统一，mybatis是从connection层面控制，jdbctemplate是从datasource层面控制，所以全部使用键值对存储
public class DynamicTransaction implements Transaction {
private final DynamicDataSource dynamicDataSource;
private ConcurrentHashMap<String, DataSource> dataSources;
private ConcurrentHashMap<String, Connection> connections;
private ConcurrentHashMap<String, Boolean> autoCommits;
private ConcurrentHashMap<String, Boolean> isConnectionTransactionals;
public DynamicTransaction(DataSource dataSource) {
this.dynamicDataSource = (DynamicDataSource) dataSource;
dataSources = new ConcurrentHashMap<>();
connections = new ConcurrentHashMap<>();
autoCommits = new ConcurrentHashMap<>();
isConnectionTransactionals = new ConcurrentHashMap<>();
}
public Connection getConnection() throws SQLException {
String dataBaseID = DBContextHolder.getDataSource();
if (!dataSources.containsKey(dataBaseID)) {
DataSource dataSource = dynamicDataSource.getAcuallyDataSource();
dataSources.put(dataBaseID, dataSource);
}
if (!connections.containsKey(dataBaseID)) {
Connection connection = DataSourceUtils.getConnection(dataSources.get(dataBaseID));
connections.put(dataBaseID, connection);
}
if (!autoCommits.containsKey(dataBaseID)) {
boolean autoCommit = connections.get(dataBaseID).getAutoCommit();
autoCommits.put(dataBaseID, autoCommit);
}
if (!isConnectionTransactionals.containsKey(dataBaseID)) {
boolean isConnectionTransactional = DataSourceUtils.isConnectionTransactional(connections.get(dataBaseID), dataSources.get(dataBaseID));
isConnectionTransactionals.put(dataBaseID, isConnectionTransactional);
}
return connections.get(dataBaseID);
}
public void commit() throws SQLException {
for (String dataBaseID : connections.keySet()) {
Connection connection = connections.get(dataBaseID);
boolean isConnectionTransactional = isConnectionTransactionals.get(dataBaseID);
boolean autoCommit = autoCommits.get(dataBaseID);
if (connection != null && !isConnectionTransactional && !autoCommit) {
connection.commit();
}
}
}
public void rollback() throws SQLException {
for (String dataBaseID : connections.keySet()) {
Connection connection = connections.get(dataBaseID);
boolean isConnectionTransactional = isConnectionTransactionals.get(dataBaseID);
boolean autoCommit = autoCommits.get(dataBaseID);
if (connection != null && !isConnectionTransactional && !autoCommit) {
connection.rollback();
}
}
}
public void close() {
for (String dataBaseID : connections.keySet()) {
Connection connection = connections.get(dataBaseID);
DataSource dataSource = dataSources.get(dataBaseID);
DataSourceUtils.releaseConnection(connection, dataSource);
}
}
public Integer getTimeout() {
return null;
}
}
public class DynamicTransactionFactory extends SpringManagedTransactionFactory {
@Override
public Transaction newTransaction(DataSource dataSource, TransactionIsolationLevel level, boolean autoCommit) {
return new DynamicTransaction(dataSource);
}
}
@Bean
public SqlSessionFactory sqlSessionFactory() throws Exception {
//SpringBootExecutableJarVFS.addImplClass(SpringBootVFS.class);
final PackagesSqlSessionFactoryBean sessionFactory = new PackagesSqlSessionFactoryBean();
sessionFactory.setDataSource(dynamicDataSource());
sessionFactory.setTransactionFactory(new DynamicTransactionFactory());
sessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver()
.getResources("classpath*:mybatis/**/*Mapper.xml"));
//关闭驼峰转换，防止带下划线的字段无法映射
sessionFactory.getObject().getConfiguration().setMapUnderscoreToCamelCase(false);
return sessionFactory.getObject();
}
事务管理器
事务中库动态切换的问题解决了，但是只针对了主库事务，如果从库操作也需要事务的特性该如何操作呢，这里就需要在注册数据源时针对每个数据源手动注册一个事务管理器
主库是固定的，可以直接在配置Bean中声明masterTransitionManage并设置为默认
@Bean("masterTransactionManager")
@Primary
public DataSourceTransactionManager MasterTransactionManager() {
return new DataSourceTransactionManager(masterDataSource());
}
从库的事务管理器我们可以拿到dataSource初始化对象，然后向Spring容器注册单例对象
public static void registerSingletonBean(String beanName, Object singletonObject) {
//将applicationContext转换为ConfigurableApplicationContext
ConfigurableApplicationContext configurableApplicationContext = (ConfigurableApplicationContext) context;
//获取BeanFactory
DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) configurableApplicationContext.getAutowireCapableBeanFactory();
if(configurableApplicationContext.containsBean(beanName)) {
defaultListableBeanFactory.destroySingleton(beanName);
}
//动态注册bean.
defaultListableBeanFactory.registerSingleton(beanName, singletonObject);
}
SpringBootBeanUtil.registerSingletonBean(key + "TransactionManager", new DataSourceTransactionManager(druidDataSource));
在使用时只要对@Transitional注解指定transitionFactory名字即可
总结解决这个问题花费了三天的时间，查了很多资料和解决方案，很多都是只有参考性或者特异性的，所以还是需把握问题的核心加上部分源码的追踪，比如本文中需要清晰的认识到Transition-Connection-LocalThread三者的关联关系，才能找对排查的方向
后续实现了集成基于JMS（atomikos）的XA两段式提交的全局事务，使用DruidXADataSrouce出现了druid和atomikos两者线程池交互出现泄露的情况放弃了，给小伙伴们避个坑