SpringBoot 多数据源及事务解决方案

来自：https://blog.csdn.net/qq381332153

1. 背景

• 如何动态管理多个数据源以及切换？

• 如何保证多数据源场景下的数据一致性(事务)？

2. 数据源切换原理

• targetDataSources&defaultTargetDataSource

• resolvedDataSources&resolvedDefaultDataSource

• determineCurrentLookupKey

3. 配置文件解决方案

• 定义DynamicDataSource类继承AbstractRoutingDataSource，重写determineCurrentLookupKey()方法。

• 配置多个数据源注入targetDataSources和defaultTargetDataSource，通过afterPropertiesSet()方法将数据源写入resolvedDataSources和resolvedDefaultDataSource。

• 调用AbstractRoutingDataSource的getConnection()方法时，determineTargetDataSource()方法返回DataSource执行底层的getConnection()。

3.1 创建数据源

spring.datasource.type=com.alibaba.druid.pool.DruidDataSource  
spring.datasource.driverClassName=com.mysql.jdbc.Driver  
# 主数据源  
spring.datasource.druid.master.url=jdbcUrl  
spring.datasource.druid.master.username=***  
spring.datasource.druid.master.password=***  
# 其他数据源  
spring.datasource.druid.second.url=jdbcUrl  
spring.datasource.druid.second.username=***  
spring.datasource.druid.second.password=***  

@Configuration  
public class DynamicDataSourceConfig {  
    @Bean  
    @ConfigurationProperties("spring.datasource.druid.master")  
    public DataSource firstDataSource(){  
        return DruidDataSourceBuilder.create().build();  
    }  
   
    @Bean  
    @ConfigurationProperties("spring.datasource.druid.second")  
    public DataSource secondDataSource(){  
        return DruidDataSourceBuilder.create().build();  
    }  
   
    @Bean  
    @Primary  
    public DynamicDataSource dataSource(DataSource firstDataSource, DataSource secondDataSource) {  
        Map

3.2 AOP处理

@Documented  
@Retention(RetentionPolicy.RUNTIME)  
@Target({ElementType.METHOD})  
public @interface SwitchDataSource {  
    String value();  
}  

3.3 方案不足

4. 数据库表解决方案

4.1 设计数据源表

4.2 自定义数据源管理

4.2.1 定义管理接口

public interface DataSourceManager {  
    void put(String var1, DataSource var2);  
   
    DataSource get(String var1);  
   
    Boolean hasDataSource(String var1);  
   
    void remove(String var1);  
   
    void closeDataSource(String var1);  
   
    Collection all();  
}  

4.2.2 自定义数据源

4.2.3 切面处理

public DataSource determineTargetDataSource() {  
        String lookupKey = DataSourceContextHolder.getKey();  
        DataSource dataSource = Optional.ofNullable(lookupKey)  
                .map(dataSources::get)  
                .orElse(defaultDataSource);  
        if (dataSource == null) {  
            throw new IllegalStateException("Cannot determine DataSource for lookup key [" + lookupKey + "]");  
        }  
        return dataSource;  
    }  

4.2.4 管理数据源状态

@SwitchDataSource      
@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)  

6. 多库事务处理

6.1 关于事务的理解

• begin(事务开始)： 可以认为存在于数据库的命令中，比如Mysql的start transaction命令，但是在JDBC编程方式中不存在。

• close(事务关闭)：Spring事务的close()方法，是把Connection对象归还给数据库连接池，与事务无关。

• suspend(事务挂起)：Spring中事务挂起的语义是：需要新事务时，将现有的Connection保存起来(还有尚未提交的事务)，然后创建新的Connection2，Connection2提交、回滚、关闭完毕后，再把Connection1取出来继续执行。

• resume(事务恢复)： 嵌套事务执行完毕，返回上层事务重新绑定连接对象到事务管理器的过程。

• 关闭一个数据库连接，这已经不再是事务的方法了。

6.2 自定义管理事务

6.2.1 定义多事务注解

@Target({ElementType.METHOD})  
@Retention(RetentionPolicy.RUNTIME)  
@Documented  
public @interface MultiTransaction {  
    String transactionManager() default "multiTransactionManager";  
    // 默认数据隔离级别，随数据库本身默认值  
    IsolationLevel isolationLevel() default IsolationLevel.DEFAULT;  
    // 默认为主库数据源  
    String datasourceId() default "default";  
    // 只读事务，若有更新操作会抛出异常  
    boolean readOnly() default false;  

6.2.3 包装Connection

public class ConnectionProxy implements Connection {  
   
    private final Connection connection;  
   
    public ConnectionProxy(Connection connection) {  
        this.connection = connection;  
    }  
   
    @Override  
    public void commit() throws SQLException {  
        // connection.commit();  
    }  
   
    public void realCommit() throws SQLException {  
        connection.commit();  
    }  
   
    @Override  
    public void close() throws SQLException {  
        //connection.close();  
    }  
   
    public void realClose() throws SQLException {  
        if (!connection.getAutoCommit()) {  
            connection.setAutoCommit(true);  
        }  
        connection.close();  
    }  
   
    @Override  
    public void rollback() throws SQLException {  
        if(!connection.isClosed())  
            connection.rollback();  
    }  
    ...  
}  

6.2.4 事务上下文管理

public class TransactionHolder {  
    // 是否开启了一个MultiTransaction  
    private boolean isOpen;  
    // 是否只读事务  
    private boolean readOnly;  
    // 事务隔离级别  
    private IsolationLevel isolationLevel;  
    // 维护当前线程事务ID和连接关系  
    private ConcurrentHashMap connectionMap;  
    // 事务执行栈  
    private Stack executeStack;  
    // 数据源切换栈  
    private Stack datasourceKeyStack;  
    // 主事务ID  
    private String mainTransactionId;  
    // 执行次数  
    private AtomicInteger transCount;  
   
    // 事务和数据源key关系  
    private ConcurrentHashMap executeIdDatasourceKeyMap;  
   
}  

6.2.5 数据源兼容处理

@Override  
public Connection getConnection() throws SQLException {  
    TransactionHolder transactionHolder = MultiTransactionManager.TRANSACTION_HOLDER_THREAD_LOCAL.get();  
    if (Objects.isNull(transactionHolder)) {  
        return determineTargetDataSource().getConnection();  
    }  
    ConnectionProxy ConnectionProxy = transactionHolder.getConnectionMap()  
            .get(transactionHolder.getExecuteStack().peek());  
    if (ConnectionProxy == null) {  
        // 没开跨库事务，直接返回  
        return determineTargetDataSource().getConnection();  
    } else {  
        transactionHolder.addCount();  
        // 开了跨库事务，从当前线程中拿包装过的Connection  
        return ConnectionProxy;  
    }  
}  

6.2.6 切面处理

package com.github.mtxn.transaction.aop;  
@Aspect  
@Component  
@Slf4j  
@Order(99999)  
public class MultiTransactionAop {  
   
    @Pointcut("@annotation(com.github.mtxn.transaction.annotation.MultiTransaction)")  
    public void pointcut() {  
        if (log.isDebugEnabled()) {  
            log.debug("start in transaction pointcut...");  
        }  
    }  
   
   
    @Around("pointcut()")  
    public Object aroundTransaction(ProceedingJoinPoint point) throws Throwable {  
        MethodSignature signature = (MethodSignature) point.getSignature();  
        // 从切面中获取当前方法  
        Method method = signature.getMethod();  
        MultiTransaction multiTransaction = method.getAnnotation(MultiTransaction.class);  
        if (multiTransaction == null) {  
            return point.proceed();  
        }  
        IsolationLevel isolationLevel = multiTransaction.isolationLevel();  
        boolean readOnly = multiTransaction.readOnly();  
        String prevKey = DataSourceContextHolder.getKey();  
        MultiTransactionManager multiTransactionManager = Application.resolve(multiTransaction.transactionManager());  
        // 切数据源，如果失败使用默认库  
        if (multiTransactionManager.switchDataSource(point, signature, multiTransaction)) return point.proceed();  
        // 开启事务栈  
        TransactionHolder transactionHolder = multiTransactionManager.startTransaction(prevKey, isolationLevel, readOnly, multiTransactionManager);  
        Object proceed;  
   
        try {  
            proceed = point.proceed();  
            multiTransactionManager.commit();  
        } catch (Throwable ex) {  
            log.error("execute method:{}#{},err:", method.getDeclaringClass(), method.getName(), ex);  
            multiTransactionManager.rollback();  
            throw ExceptionUtils.api(ex, "系统异常：%s", ex.getMessage());  
        } finally {  
            // 当前事务结束出栈  
            String transId = multiTransactionManager.getTrans().getExecuteStack().pop();  
            transactionHolder.getDatasourceKeyStack().pop();  
            // 恢复上一层事务  
            DataSourceContextHolder.setKey(transactionHolder.getDatasourceKeyStack().peek());  
            // 最后回到主事务，关闭此次事务  
            multiTransactionManager.close(transId);  
        }  
        return proceed;  
   
    }  
   
   
}  

7.总结