Springboot 多数据源事务

起因

在一个service方法上使用的事务,其中有方法是调用的多数据源orderDB

但是多数据源没有生效,而是使用的primaryDB

原因

spring 事务实现的方式

以 @Transactional 注解为例 (也可以看 TransactionTemplate， 这个流程更简单一点)。
入口：ProxyTransactionManagementConfiguration
（从 config 类入手，需要哪些bean一目了然，然后直接顺着看下去就可以了）
主要有以下3个bean

• TransactionAttri``buteSource：实现是 AnnotationTransactionAttributeSource， 提供从（存在 @Transactional 注解的）方法上读取事务的属性（注解的属性）的功能
• TransactionInterceptor：事务方法拦截器的bean，在执行事务方法时，转到 (TransactionAspectSupport#invokeWithinTransaction) 方法，即spring事务处理的主要逻辑。
• BeanFactoryTransactionAttributeSourceAdvisor：一个advisor（包含一个 Pointcut 切点和一个 Advice 通知），advice就是上面的事务拦截器，Pointcut 切点匹配能通过 TransactionAttributeSource 获取到事务信息的方法。

拦截器逻辑大概如下:

解决方案

每个数据源手动配置SqlSessionFactory

这种方式是通过手动声明创建orm框架对应的bean来实现多数据源的操作，即每个数据源都自己手动创建一套对用的bean。
不支持多个数据源事务，手动配置较繁琐
(如果使用的spring而不是springboot的话，就不会有这种多数据源的疑问，因为本来就要自己声明bean)

动态数据源(本次使用)

只需要把@Transactional(rollbackFor = Exception.class) 换为@DSTransactional即可
并且抛出异常事务也会回滚

动态数据源实现原理

同样看一下 DynamicDataSourceAutoConfiguration 这个配置相关的类就大概了解了。

• DynamicRoutingDataSource: 动态数据源，内部使用 Map 保存了多个数据源。获取 connection 时，根据 ThreadLocal 中的 dsKey 获取对应的数据源
  • 另：对于多数据源事务 （TransactionContext.getXID() isNotEmpty），会返回一个 ConnectionProxy 并暂存到 ConnectionFactory 中， 该 ConnectionProxy 不会执行 commit、rollback、close 操作事务相关的方法。

public Connection getConnection() throws SQLException {
    String xid = TransactionContext.getXID();
    if (StringUtils.isEmpty(xid)) {
        // 非多数据源事务直接获取对应 connection
        return determineDataSource().getConnection();
    } else {
        String ds = DynamicDataSourceContextHolder.peek();
        ds = StringUtils.isEmpty(ds) ? "default" : ds;
        // 多数据源事务，使用代理的 connection (屏蔽了 commit 等操作)
        ConnectionProxy connection = ConnectionFactory.getConnection(ds);
        return connection == null ? getConnectionProxy(ds, determineDataSource().getConnection()) : connection;
    }
}

// 获取 代理的 connection， 并将其存入 ConnectionFactory， 内部维护一个 ThreadLocal<Map>， 同时会 setAutoCommit(false) 开启事务
private Connection getConnectionProxy(String ds, Connection connection) {
    ConnectionProxy connectionProxy = new ConnectionProxy(connection, ds);
    ConnectionFactory.putConnection(ds, connectionProxy);
    return connectionProxy;
}

// DynamicRoutingDataSource
// 从 ThreadLocal 获取当前 dsKey 然后获取对应 datasource
public DataSource determineDataSource() {
    String dsKey = DynamicDataSourceContextHolder.peek();
    return getDataSource(dsKey);
}

• DynamicDataSourceAnnotationInterceptor: 处理 @DS 注解的拦截器，获取 @DS 指定的 datasource 并存入 ThreadLocal 中， 供 DynamicRoutingDataSource 使用
• dynamicTransactionAdvisor: 处理 @DSTransactional 多数据源事务注解的拦截器，在执行目标方法前，标记为多数据源事务 (TransactionContext.bind(xid))， 执行完后， 通知 ConnectionFactory 中的 connectionProxy 进行事务的 commit 或 rollback。

// DynamicLocalTransactionAdvisor
public Object invoke(MethodInvocation methodInvocation) throws Throwable {
    if (!StringUtils.isEmpty(TransactionContext.getXID())) {
        return methodInvocation.proceed();
    }
    // 事务是否成功
    boolean state = true;
    Object o;
    String xid = UUID.randomUUID().toString();
    // 标记当前为 多数据源事务
    TransactionContext.bind(xid);
    try {
        o = methodInvocation.proceed();
    } catch (Exception e) {
        state = false;
        throw e;
    } finally {
        // 通知 connectionProxy 进行 commit 或 rollback
        ConnectionFactory.notify(state);
        TransactionContext.remove();
    }
    return o;
}