同一个事务操作多个数据库

codeImport 2020-03-26

4504

使用过Spring管理事务的同学都知道在同一个事务中只能对同一个数据库进行操作，本文要解决的是同一个事务中操作多个数据库的问题。

前言

有些业务由于一些特殊原因必须在同一个事务中对两个数据库进行修改操作，且必须保证强一致性。但是默认情况下，使用spring进行事务管理，同一个事务只支持操作一个数据库，如果操作了多个数据库，会抛出某些表不存在的异常。

为什么会抛出这个异常？

spring在开启事务时，就会从连接池中获取到一条连接，放入事务上下文中，每次执行sql时，都会使用该连接调用数据库。因为一条连接只能对应一个数据库，所以当操作第二个数据库的表时，用第一个数据库的连接当然是找不到对应的表的。

事务开启时，获取数据库连接并置入事务上下文中的代码在DataSourceTransactionManager类中，这也是spring默认的事务管理器。

    @Override
    protected void doBegin(Object transaction, TransactionDefinition definition) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
        Connection con = null;


        try {
            if (!txObject.hasConnectionHolder() ||
                    txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
                // 此处从连接池获取连接，先放入ConnectionHolder中
                Connection newCon = obtainDataSource().getConnection();
                txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
            }


            prepareTransactionalConnection(con, definition);
            txObject.getConnectionHolder().setTransactionActive(true);


            // 此处将ConnectionHolder置入事务上下文中
            // Bind the connection holder to the thread.
            if (txObject.isNewConnectionHolder()) {
                TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
            }
        }


    }

每次执行sql获取数据库连接是在SpringManagedTransaction中

    public Connection getConnection() throws SQLException {
        if (this.connection == null) {
            // 事务中第一次执行sql的时候会进行这里获取连接
            this.openConnection();
        }


        return this.connection;
    }


    private void openConnection() throws SQLException {
        // 这里的DataSourceUtils就是从事务上下文中取出原先存在那里的连接
        this.connection = DataSourceUtils.getConnection(this.dataSource);
        this.autoCommit = this.connection.getAutoCommit();
        this.isConnectionTransactional = DataSourceUtils.isConnectionTransactional(this.connection, this.dataSource);


    }

    public static Connection doGetConnection(DataSource dataSource) throws SQLException {
        Assert.notNull(dataSource, "No DataSource specified");
        // 此处就是从事务上下文中的ConnectionHolder获取到了开启事务时存放的连接
        ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource);
        if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
            conHolder.requested();
            if (!conHolder.hasConnection()) {
                conHolder.setConnection(fetchConnection(dataSource));
            }
            return conHolder.getConnection();
        }


        return con;
    }

如何解决？

从上节分析可以看出，默认情况下，每个事务只能共用一个连接，而每个连接只能连接一个数据库，所以每个事务就只能操作一个数据库。但是有一个细节需要注意，spring使用了一个容器ConnectionHolder存放连接，最终也是从该容器中获取了连接。所以笔者的做法就是重写ConnectionHolder，使得该容器可以持有多条连接，然后使用一个切面拦截所有sql调用，判断该调用是要连接到哪个数据库，再从ConnectionHolder中获取到对应数据库的连接。事务执行完毕后，对该ConnectionHolder中的所有连接执行commit操作。

持有多条连接的ConnectionHolder如下：

public class DynamicConnectionHolder extends ConnectionHolder {
    private Map<Object, Connection> targetConnectionMap = new HashMap<>();
    @Override
    public Connection getConnection() {
        // 此处的key为切面拦截sql调用时根据调用所在的包获取到的数据库枚举
        Object key = DynamicDataSourceContextHolder.getDataSourceKey();
        if (key == null) {
            DynamicDataSourceContextHolder.useMaster();
            key = DynamicDataSourceContextHolder.getDataSourceKey();
        }
        Connection connection = targetConnectionMap.get(key);
        if (connection == null) {
            try {
                connection = dataSource.getConnection();
                // 数据库连接autoCommit默认为true,此处设置自动提交为false
                connection.setAutoCommit(false);
                targetConnectionMap.put(key, connection);
            } catch (SQLException ex) {
                logger.error(ex.getMessage());
            }
        }
        return connection;
    }
}

使用的切面如下，此处的关键是操作不同数据库的Dao要写在不同的包中

    @Pointcut("execution(* com.xxx.xxx.xxx.dao.db1..*(..)))
    public void daoAspect() {


    }


    @Before("daoAspect()")
    public void switchDataSource(JoinPoint point) {
        // 调用Dao方法前设置将要使用的数据库
        DynamicDataSourceContextHolder.useXXX();
    }

因为事务管理器现在必须使用我们自定义的ConnectionHolder，所以还得重写事务管理器DataSourceTransactionManager

public class DynamicTransactionManager extends DataSourceTransactionManager {
    @Override
    protected void doBegin(Object transaction, TransactionDefinition definition) {
        if (transaction instanceof JdbcTransactionObjectSupport) {
            JdbcTransactionObjectSupport support = (JdbcTransactionObjectSupport) transaction;
            try {
                // 设置connectionHolder为我们自定义的DynamicConnectionHolder
                support.setConnectionHolder(new DynamicConnectionHolder(this.getDataSource()));


                Field declaredField = transaction.getClass().getDeclaredField("newConnectionHolder");
                declaredField.setAccessible(true);
                declaredField.set(transaction, true);
            } catch (Exception ex) {
                logger.error(ex.getMessage());
            }
        }
        super.doBegin(transaction, definition);
    }


    @Override
    protected void doCommit(DefaultTransactionStatus status) {
        JdbcTransactionObjectSupport txObject = (JdbcTransactionObjectSupport) status.getTransaction();
        ConnectionHolder conHolder = txObject.getConnectionHolder();
        Set<Object> keys = ((DynamicConnectionHolder) conHolder).getTargetConnectionMap().keySet();
        // 事务提交时，要提交该事务下的所有连接
        try {
            for (Object key : keys) {
                Connection connection = ((DynamicConnectionHolder) conHolder).getTargetConnectionMap().get(key);
                connection.commit();
            }
        } catch (Exception ex) {
            logger.error("事务提交失败", ex);
        }
    }
}

为了让事务每次执行sql时都从ConnectionHolder获取连接，现在就不能使用SpringManagedTransaction获取连接了，需要重写Transaction

public class DynamicManagedTransaction implements Transaction {
        @Override
    public Connection getConnection() throws SQLException {
        if (this.connection == null) {
            openConnection();
        }
        // 每次执行sql，都从DynamicConnectionHolder获取连接
        if (conHolder != null) {
            return conHolder.getConnection();
        } else {
            return this.connection;
        }
    }
}

总结：

要想在同一个事务中操作多个数据库，关键就是要解决同一个事务中获取多个数据库的连接的问题。这套代码已经在生产环境稳定运行了半年，这也是笔者解决这个问题的思路。

近期热文：

聊一聊熟悉又陌生的autoCommit

文章转载自codeImport，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

同一个事务操作多个数据库

前言

为什么会抛出这个异常？

如何解决？

总结：

评论