Mybatis操作多數(shù)據(jù)源實(shí)現(xiàn)的方法

今天小編給大家分享的是Mybatis操作多數(shù)據(jù)源實(shí)現(xiàn)的方法，相信很多人都不太了解，為了讓大家更加了解，所以給大家總結(jié)了以下內(nèi)容，一起往下看吧。一定會(huì)有所收獲的哦。

現(xiàn)在有一個(gè)Mysql數(shù)據(jù)源和一個(gè)Postgresql數(shù)據(jù)源，使用Mybatis對(duì)兩個(gè)數(shù)據(jù)源進(jìn)行操作：

1. 注入多數(shù)據(jù)源

可以對(duì)兩個(gè)數(shù)據(jù)源分別實(shí)現(xiàn)其Service層和Mapper層，以及Mybatis的配置類：

@Configuration
// 這里需要配置掃描包路徑，以及sqlSessionTemplateRef
@MapperScan(basePackages = "com.example.mybatisdemo.mapper.mysql", sqlSessionTemplateRef = "mysqlSqlSessionTemplate")
public class MysqlMybatisConfigurer {
    /**
     * 注入Mysql數(shù)據(jù)源
     */
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.mysql")
    public DataSource mysqlDatasource() {
        return new DruidDataSource();
    }
    /**
     * 注入mysqlSqlSessionFactory
     */
    @Bean
    public SqlSessionFactory mysqlSqlSessionFactory(DataSource mysqlDatasource) throws Exception {
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(mysqlDatasource);
        // 設(shè)置對(duì)應(yīng)的mapper文件
        factoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:" +
                "/mappers/MysqlMapper.xml"));
        return factoryBean.getObject();
    }
    /**
     * 注入mysqlSqlSessionTemplate
     */
    @Bean
    public SqlSessionTemplate mysqlSqlSessionTemplate(SqlSessionFactory mysqlSqlSessionFactory) {
        return new SqlSessionTemplate(mysqlSqlSessionFactory);
    }
    /**
     * 注入mysqlTransactionalManager
     */
    @Bean
    public DataSourceTransactionManager mysqlTransactionalManager(DataSource mysqlDatasource) {
        return new DataSourceTransactionManager(mysqlDatasource);
    }
}

@Configuration
// 這里需要配置掃描包路徑，以及sqlSessionTemplateRef
@MapperScan(basePackages = "com.example.mybatisdemo.mapper.postgresql", sqlSessionTemplateRef = "postgresqlSqlSessionTemplate")
public class PostgresqlMybatisConfigurer {
    /**
     * 注入Postgresql數(shù)據(jù)源
     */
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.postgresql")
    public DataSource postgresqlDatasource() {
        return new DruidDataSource();
    }
    /**
     * 注入postgresqlSqlSessionFactory
     */
    @Bean
    public SqlSessionFactory postgresqlSqlSessionFactory(DataSource postgresqlDatasource) throws Exception {
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(postgresqlDatasource);
        // 設(shè)置對(duì)應(yīng)的mapper文件
        factoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:" +
                "/mappers/PostgresqlMapper.xml"));
        return factoryBean.getObject();
    }
    /**
     * 注入postgresqlSqlSessionTemplate
     */
    @Bean
    public SqlSessionTemplate postgresqlSqlSessionTemplate(SqlSessionFactory postgresqlSqlSessionFactory) {
        return new SqlSessionTemplate(postgresqlSqlSessionFactory);
    }
    /**
     * 注入postgresqlTransactionalManager
     */
    @Bean
    public DataSourceTransactionManager postgresqlTransactionalManager(DataSource postgresqlDatasource) {
        return new DataSourceTransactionManager(postgresqlDatasource);
    }
}

在配置類中，分別注入了一個(gè)事務(wù)管理器TransactionManager，這個(gè)和事務(wù)管理是相關(guān)的。在使用@Transactional注解時(shí)，需要配置其value屬性指定對(duì)應(yīng)的事務(wù)管理器。

2. 動(dòng)態(tài)數(shù)據(jù)源

Spring中提供了AbstractRoutingDataSource抽象類，可以用于動(dòng)態(tài)地選擇數(shù)據(jù)源。

public abstract class AbstractRoutingDataSource extends AbstractDataSource implements InitializingBean {
    @Nullable
    private Map<Object, Object> targetDataSources;
    @Nullable
    private Object defaultTargetDataSource;
    private boolean lenientFallback = true;
    private DataSourceLookup dataSourceLookup = new JndiDataSourceLookup();
    @Nullable
    private Map<Object, DataSource> resolvedDataSources;
    @Nullable
    private DataSource resolvedDefaultDataSource;
    // 略
}

通過源碼可以看到，該抽象類實(shí)現(xiàn)了InitializingBean接口，并在其afterPropertiesSet方法中將數(shù)據(jù)源以<lookupkey, dataSource>的形式放入一個(gè)Map中。

public void afterPropertiesSet() {
    if (this.targetDataSources == null) {
        throw new IllegalArgumentException("Property 'targetDataSources' is required");
    } else {
        this.resolvedDataSources = CollectionUtils.newHashMap(this.targetDataSources.size());
        this.targetDataSources.forEach((key, value) -> {
            Object lookupKey = this.resolveSpecifiedLookupKey(key);
            DataSource dataSource = this.resolveSpecifiedDataSource(value);
            // 將數(shù)據(jù)源以<lookupkey, dataSource>的形式放入Map中
            this.resolvedDataSources.put(lookupKey, dataSource);
        });
        if (this.defaultTargetDataSource != null) {
            this.resolvedDefaultDataSource = this.resolveSpecifiedDataSource(this.defaultTargetDataSource);
        }
    }
}

該類中還有一個(gè)determineTargetDataSource方法，是根據(jù)lookupkey從Map中獲取對(duì)應(yīng)的數(shù)據(jù)源，如果沒有獲取到，則使用默認(rèn)的數(shù)據(jù)源。

protected DataSource determineTargetDataSource() {
    Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
    Object lookupKey = this.determineCurrentLookupKey();
    // 根據(jù)lookupkey從Map中獲取對(duì)應(yīng)的數(shù)據(jù)源
    DataSource dataSource = (DataSource)this.resolvedDataSources.get(lookupKey);
    if (dataSource == null && (this.lenientFallback || lookupKey == null)) {
        dataSource = this.resolvedDefaultDataSource;
    }
    if (dataSource == null) {
        throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");
    } else {
        return dataSource;
    }
}

lookupkey是通過determineTargetDataSource方法獲取到的，而它是一個(gè)抽象方法，我們要做的就是通過實(shí)現(xiàn)這個(gè)方法，來(lái)控制獲取到的數(shù)據(jù)源。

@Nullable
protected abstract Object determineCurrentLookupKey();

(1) 創(chuàng)建并注入動(dòng)態(tài)數(shù)據(jù)源

創(chuàng)建AbstractRoutingDataSource的子類，實(shí)現(xiàn)determineCurrentLookupKey方法

public class RoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return DataSourceContextHolder.get();
    }
}

這里的DataSourceContextHolder是一個(gè)操作ThreadLocal對(duì)象的工具類

public class DataSourceContextHolder {
    /**
     * 數(shù)據(jù)源上下文
     */
    private static final ThreadLocal<DataSourceType> contextHolder = new ThreadLocal<>();
    /**
     * 設(shè)置數(shù)據(jù)源類型
     */
    public static void set(DataSourceType type) {
        contextHolder.set(type);
    }
    /**
     * 獲取數(shù)據(jù)源類型
     *
     * @return DataSourceType
     */
    public static DataSourceType get() {
        return contextHolder.get();
    }
    /**
     * 使用MYSQL數(shù)據(jù)源
     */
    public static void mysql() {
        set(DataSourceType.MYSQL);
    }
    /**
     * 使用Postgresql數(shù)據(jù)源
     */
    public static void postgresql() {
        set(DataSourceType.POSTGRESQL);
    }
    public static void remove() {
        contextHolder.remove();
    }
}

通過調(diào)用DataSourceContextHolder.mysql()或者DataSourceContextHolder.postgresql()就能修改contextHolder的值，從而在動(dòng)態(tài)數(shù)據(jù)源的determineTargetDataSource方法中就能獲取到對(duì)應(yīng)的數(shù)據(jù)源。

在數(shù)據(jù)源配置類中，將mysql和postgresql的數(shù)據(jù)源設(shè)置到動(dòng)態(tài)數(shù)據(jù)源的Map中，并注入容器。

@Configuration
public class DataSourceConfigurer {
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.mysql")
    public DataSource mysqlDatasource() {
        return new DruidDataSource();
    }
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.postgresql")
    public DataSource postgresqlDatasource() {
        return new DruidDataSource();
    }
    @Bean
    public RoutingDataSource routingDataSource(DataSource mysqlDatasource, DataSource postgresqlDatasource) {
        Map<Object, Object> dataSources = new HashMap<>();
        dataSources.put(DataSourceType.MYSQL, mysqlDatasource);
        dataSources.put(DataSourceType.POSTGRESQL, postgresqlDatasource);
        RoutingDataSource routingDataSource = new RoutingDataSource();
        routingDataSource.setDefaultTargetDataSource(mysqlDatasource);
        // 設(shè)置數(shù)據(jù)源
        routingDataSource.setTargetDataSources(dataSources);
        return routingDataSource;
    }
}

(2) Mybatis配置類

由于使用了動(dòng)態(tài)數(shù)據(jù)源，所以只需要編寫一個(gè)配置類即可。

@Configuration
@MapperScan(basePackages = "com.example.mybatisdemo.mapper", sqlSessionTemplateRef = "sqlSessionTemplate")
public class MybatisConfigurer {
    // 注入動(dòng)態(tài)數(shù)據(jù)源
    @Resource
    private RoutingDataSource routingDataSource;
    @Bean
    public SqlSessionFactory sqlSessionFactory() throws Exception {
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(routingDataSource);
        // 這里可以直接設(shè)置所有的mapper.xml文件
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath" +
                ":mappers/*.xml"));
        return sqlSessionFactoryBean.getObject();
    }
    @Bean
    public SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {
        return new SqlSessionTemplate(sqlSessionFactory);
    }
    @Bean
    public DataSourceTransactionManager transactionalManager(DataSource mysqlDatasource) {
        return new DataSourceTransactionManager(mysqlDatasource);
    }
}

(3) 使用注解簡(jiǎn)化數(shù)據(jù)源切換

我們雖然可以使用DataSourceContextHolder類中的方法進(jìn)行動(dòng)態(tài)數(shù)據(jù)源切換，但是這種方式有些繁瑣，不夠優(yōu)雅?？梢钥紤]使用注解的形式簡(jiǎn)化數(shù)據(jù)源切換。
我們先定義兩個(gè)注解，表示使用Mysql數(shù)據(jù)源或Postgresql數(shù)據(jù)源：

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface Mysql {
}

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface Postgresql {
}

再定義一個(gè)切面，當(dāng)使用了注解時(shí)，會(huì)先調(diào)用切換數(shù)據(jù)源的方法，再執(zhí)行后續(xù)邏輯。

@Component
@Aspect
public class DataSourceAspect {
    @Pointcut("@within(com.example.mybatisdemo.aop.Mysql) || @annotation(com.example.mybatisdemo.aop.Mysql)")
    public void mysqlPointcut() {
    }
    @Pointcut("@within(com.example.mybatisdemo.aop.Postgresql) || @annotation(com.example.mybatisdemo.aop.Postgresql)")
    public void postgresqlPointcut() {
    }
    @Before("mysqlPointcut()")
    public void mysql() {
        DataSourceContextHolder.mysql();
    }
    @Before("postgresqlPointcut()")
    public void postgresql() {
        DataSourceContextHolder.postgresql();
    }
}

在使用動(dòng)態(tài)數(shù)據(jù)源的事務(wù)操作時(shí)有兩個(gè)需要注意的問題：

問題一    同一個(gè)事務(wù)操作兩個(gè)數(shù)據(jù)源

Mybatis使用Executor執(zhí)行SQL時(shí)需要獲取連接，BaseExecutor類中的getConnection方法調(diào)用了SpringManagedTransaction中的getConnection方法，這里優(yōu)先從connection字段獲取連接，如果connection為空，才會(huì)調(diào)用openConnection方法，并把連接賦給connection字段。

也就是說(shuō)，如果你使用的是同一個(gè)事務(wù)來(lái)操作兩個(gè)數(shù)據(jù)源，那拿到的都是同一個(gè)連接，會(huì)導(dǎo)致數(shù)據(jù)源切換失敗。

protected Connection getConnection(Log statementLog) throws SQLException {
    Connection connection = this.transaction.getConnection();
    return statementLog.isDebugEnabled() ? ConnectionLogger.newInstance(connection, statementLog, this.queryStack) : connection;
}

public Connection getConnection() throws SQLException {
    if (this.connection == null) {
        this.openConnection();
    }
    return this.connection;
}

private void openConnection() throws SQLException {
    this.connection = DataSourceUtils.getConnection(this.dataSource);
    this.autoCommit = this.connection.getAutoCommit();
    this.isConnectionTransactional = DataSourceUtils.isConnectionTransactional(this.connection, this.dataSource);
    LOGGER.debug(() -> {
        return "JDBC Connection [" + this.connection + "] will" + (this.isConnectionTransactional ? " " : " not ") + "be managed by Spring";
    });
}

問題二      兩個(gè)獨(dú)立事務(wù)分別操作兩個(gè)數(shù)據(jù)源

(1) 在開啟事務(wù)的時(shí)候，DataSourceTransactionManager中的doBegin方法會(huì)先獲取Connection，并保存到ConnectionHolder中，將數(shù)據(jù)源和ConnectionHolder的對(duì)應(yīng)關(guān)系綁定到TransactionSynchronizationManager中。

protected void doBegin(Object transaction, TransactionDefinition definition) {
    DataSourceTransactionManager.DataSourceTransactionObject txObject = (DataSourceTransactionManager.DataSourceTransactionObject)transaction;
    Connection con = null;
    try {
        if (!txObject.hasConnectionHolder() || txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
            // 獲取連接
            Connection newCon = this.obtainDataSource().getConnection();
            if (this.logger.isDebugEnabled()) {
                this.logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
            }
            // 保存到ConnectionHolder中
            txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
        }
        txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
        // 從ConnectionHolder獲取連接
        con = txObject.getConnectionHolder().getConnection();
        // 略
        // 將數(shù)據(jù)源和ConnectionHolder的關(guān)系綁定到TransactionSynchronizationManager中
        if (txObject.isNewConnectionHolder()) {
            TransactionSynchronizationManager.bindResource(this.obtainDataSource(), txObject.getConnectionHolder());
        }
         // 略
}

(2) TransactionSynchronizationManager的bindResource方法將數(shù)據(jù)源和ConnectionHolder的對(duì)應(yīng)關(guān)系存入線程變量resources中。

public abstract class TransactionSynchronizationManager {
    // 線程變量
    private static final ThreadLocal<Map<Object, Object>> resources =
         new NamedThreadLocal<>("Transactional resources");
    // 略
    // 綁定數(shù)據(jù)源和ConnectionHolder的對(duì)應(yīng)關(guān)系
    public static void bindResource(Object key, Object value) throws IllegalStateException {
       Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
       Assert.notNull(value, "Value must not be null");
       Map<Object, Object> map = resources.get();
       // set ThreadLocal Map if none found
       if (map == null) {
          map = new HashMap<>();
          resources.set(map);
       }
       Object oldValue = map.put(actualKey, value);
       // Transparently suppress a ResourceHolder that was marked as void...
       if (oldValue instanceof ResourceHolder && ((ResourceHolder) oldValue).isVoid()) {
          oldValue = null;
       }
       if (oldValue != null) {
          throw new IllegalStateException(
                "Already value [" + oldValue + "] for key [" + actualKey + "] bound to thread");
       }
    }
    // 略
}

(3) 上邊提到的openConnection方法，其實(shí)最終也是從TransactionSynchronizationManager的resources中獲取連接的

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()) {
        logger.debug("Fetching JDBC Connection from DataSource");
        Connection con = fetchConnection(dataSource);
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            try {
                ConnectionHolder holderToUse = conHolder;
                if (conHolder == null) {
                    holderToUse = new ConnectionHolder(con);
                } else {
                    conHolder.setConnection(con);
                }
                holderToUse.requested();
                TransactionSynchronizationManager.registerSynchronization(new DataSourceUtils.ConnectionSynchronization(holderToUse, dataSource));
                holderToUse.setSynchronizedWithTransaction(true);
                if (holderToUse != conHolder) {
                    TransactionSynchronizationManager.bindResource(dataSource, holderToUse);
                }
            } catch (RuntimeException var4) {
                releaseConnection(con, dataSource);
                throw var4;
            }
        }
        return con;
    } else {
        conHolder.requested();
        if (!conHolder.hasConnection()) {
            logger.debug("Fetching resumed JDBC Connection from DataSource");
            conHolder.setConnection(fetchConnection(dataSource));
        }
        // 從ConnectionHolder中獲取連接
        return conHolder.getConnection();
    }
}

也就是說(shuō)，如果修改了數(shù)據(jù)源，那么resources中就找不到對(duì)應(yīng)的連接，就可以重新獲取連接，從而達(dá)到切換數(shù)據(jù)源的目的。然而我們數(shù)據(jù)源的只有一個(gè)，就是動(dòng)態(tài)數(shù)據(jù)源，因此即使使用兩個(gè)獨(dú)立事務(wù)，也不能成功切換數(shù)據(jù)源。

3. 結(jié)語(yǔ)

如果想要使用動(dòng)態(tài)數(shù)據(jù)源的事務(wù)處理，可能需要考慮使用多線程分布式的事務(wù)處理機(jī)制；
如果使用直接注入多個(gè)數(shù)據(jù)源的方式實(shí)現(xiàn)事務(wù)處理，實(shí)現(xiàn)簡(jiǎn)單，但是各數(shù)據(jù)源事務(wù)是獨(dú)立的；
應(yīng)該根據(jù)具體情況進(jìn)行選擇。

關(guān)于Mybatis操作多數(shù)據(jù)源實(shí)現(xiàn)的方法就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的參考價(jià)值，可以學(xué)以致用。如果喜歡本篇文章，不妨把它分享出去讓更多的人看到。