一、背景

当我们一个数据库存储量已经不满足需求时，需要进行多数据源存储。在不同的数据源中，就会引发的多数据源查询的一系列问题

二、sharding jdbc 简介

Sharding-JDBC是谷歌的一个开源的框架，提供标准化的数据分片、分布式事务和数据库治理功能，定位为轻量级Java框架，在Java的JDBC层提供的额外服务。 它使用客户端直连数据库，以jar包形式提供服务，无需额外部署和依赖，可理解为增强版的JDBC驱动，完全兼容JDBC和各种ORM框架。
适用于任何基于Java的ORM框架，如：JPA, Hibernate, Mybatis, Spring JDBC Template或直接使用JDBC。
基于任何第三方的数据库连接池，如：DBCP, C3P0, BoneCP, Druid, HikariCP等。
支持任意实现JDBC规范的数据库。目前支持MySQL，Oracle，SQLServer和PostgreSQL。

三、项目融合sharding jdbc

项目现有的是动态切换对应的数据源，key是部门的id，并不满足我们对一条sql语句就可以查询所有的部门数据的问题，所以要对现有有的逻辑进行改造
首先我们要获取到所有的数据源的信息，初始化一份数据，如下图，这个是现有的存数据源信息的表dic_database_info

image (1).png

然后我们要通过这张表获取所有的部门的数据源，并且初始化到线程池中，代码如下：

/**
     * @Description: 通过jdbc查询出数据连接信息　　
     * @return
     * @date 2020-09-11 12:25
     * @throws
     */
    private Map<String, DataSource> getDepartmentDataSources (DataSource rdsDataSource){
        String sql = "select id, department_id as departmentId, access_url as accessUrl,\n" +
                "        access_key as accessKey, access_value as accessValue\n" +
                "        from dic_database_info where status=1 order by id ";
        Map<String, DataSource> dataSourceMap = new HashMap<>();
        try {
            PreparedStatement preparedStatement = rdsDataSource.getConnection().prepareStatement(sql);
            ResultSet resultSet = preparedStatement.executeQuery();

            while (resultSet.next()){
                String departmentId = String.valueOf(resultSet.getLong("departmentId"));
                String accessUrl = resultSet.getString("accessUrl");
                String accessKey = resultSet.getString("accessKey");
                String accessValue = resultSet.getString("accessValue");
                accessValue = JasyptUtils.decrypt(accessValue,salt);
                DataSource dataSource = DataSourceBuilder.create().driverClassName(driverClassName)
                        .url(accessUrl)
                        .username(accessKey)
                        .password(accessValue)

                        .build();
                String serialNum = shardingSphereDataSourceSerialRecord.addDataSourceAlias(departmentId);
                dataSourceMap.put(serialNum,dataSource);
            }
        }catch (SQLException e){
            LOGGER.error("通过JDBC查询各局数据库信息出错:{}",e.getMessage());
        }
        return dataSourceMap;
    }

然后在maven引入包

<dependency>
            <groupId>org.apache.shardingsphere</groupId>
            <artifactId>sharding-jdbc-core</artifactId>
            <version>4.1.1</version>
</dependency>

由于办件服务只需要查询某一些表，所以我们把查询的表的表名配置在配置文件中

spring.shardingsphere.sharding.ext.tables=agent_materials,agents,applicant_materials,applicants,application_result_info,current_node,fee_unit_info,group_data_search,mark_results,node_result,paid_info,result_receive_info,service_application,service_attachment,service_correction,service_info_apply_nullify,service_info_nullify,tab_accept_affair_request,tab_apply_invalid_service_application,tab_apply_revoke_service_application,tab_get_node_info,tab_service_application_accepted,tab_service_application_result,tab_service_application_result_info

然后在配置文件中获取

@ConfigurationProperties(prefix = "spring.shardingsphere.sharding.ext")
    public static class ShardingJdbcProperties {

        private List<String> tables = Collections.emptyList();

        public List<String> getTables() {
            return tables;
        }

        public void setTables(List<String> tables) {
            this.tables = tables;
        }
    }

初始化配置文件

 @Override
    public void afterPropertiesSet() throws Exception {
        ShardingJdbcProperties shardingJdbcProperties = beanFactory.getBean(ShardingJdbcProperties.class);
        if (shardingJdbcProperties != null){
            logicTables = new HashSet<>(shardingJdbcProperties.getTables());
            initialized = true;
            lock.lock();
            try{
                condition.signalAll();
            }finally {
                lock.unlock();
            }

        }
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }

最后初始化sharding jdbc，并且保存数据源，表信息

@Bean(name = "apiDataSource")
    @ConfigurationProperties(prefix = "spring.datasource.api")
    public DataSource rdsDataSource() {
        return DataSourceBuilder.create().build();
    }

    @Bean
    @Primary
    public DataSource shardingSphereDataSource(ShardingJdbcProperties shardingJdbcProperties, @Qualifier("apiDataSource") DataSource dataSource) throws SQLException {

        final String MASTER_DATASOURCE_ALIAS = "master";

        ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
        shardingRuleConfig.setDefaultDatabaseShardingStrategyConfig(new HintShardingStrategyConfiguration(
                new DepartmentDataSourceHintShardingAlgorithm()));
        shardingRuleConfig.setDefaultTableShardingStrategyConfig(new HintShardingStrategyConfiguration(
                new DepartmentDataSourceHintShardingAlgorithm.DepartmentTableHintShardingAlgorithm()));

        Map<String, DataSource> fullDataSourceMap = new HashMap<>();
        fullDataSourceMap.put(MASTER_DATASOURCE_ALIAS, dataSource);
        Map<String, DataSource> departmentDatasourceMap = getDepartmentDataSources(dataSource);
        if (departmentDatasourceMap != null && departmentDatasourceMap.size() > 0 ){
            fullDataSourceMap.putAll(departmentDatasourceMap);
        }
        shardingRuleConfig.setDefaultDataSourceName(MASTER_DATASOURCE_ALIAS);

        for (String table : shardingJdbcProperties.getTables()){
            TableRuleConfiguration tableRuleConfiguration = getDepartmentTableRuleConfiguration(table);
            if (tableRuleConfiguration != null){
                shardingRuleConfig.getTableRuleConfigs().add(tableRuleConfiguration);
                shardingRuleConfig.getBindingTableGroups().add(table);
            }
        }

        Properties properties = new Properties();
        properties.put("sql.show", sqlShow);
        return ShardingDataSourceFactory.createDataSource(fullDataSourceMap, shardingRuleConfig, properties);
    }



    private TableRuleConfiguration getDepartmentTableRuleConfiguration (String table) {
        final String DS_TABLE_EXPRESSION = "%s${0..%d}.%s";
        if (StringUtils.isNotEmpty(table)){
            String expression = String.format(DS_TABLE_EXPRESSION, shardingSphereDataSourceSerialRecord.getDataSourcePrefix(),
                    shardingSphereDataSourceSerialRecord.getMaxSerialNum(), table);
            return  new TableRuleConfiguration(table, expression);
        }
        return null;
    }


    @Bean(name = "apiSqlSessionFactory")
    @Primary
    public SqlSessionFactory rdsSqlSessionFactory(@Qualifier("shardingSphereDataSource")DataSource dynamicDataSource) throws Exception {
        SqlSessionFactoryBean sqlSessionFactory = new SqlSessionFactoryBean();
        sqlSessionFactory.setDataSource(dynamicDataSource);
        sqlSessionFactory.setConfigLocation(new ClassPathResource("mybatis-config.xml"));
        sqlSessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:mapper/**/*.xml"));
        return sqlSessionFactory.getObject();
    }

    @Bean(name = "apiTransactionManager")
    @Primary
    public DataSourceTransactionManager rdsTransactionManager(@Qualifier("shardingSphereDataSource") DataSource dynamicDataSource){
        return new DataSourceTransactionManager(dynamicDataSource);
    }

    @Bean(name = "apiSqlSessionTemplate")
    @Primary
    public SqlSessionTemplate rdsSqlSessionTemplate(@Qualifier("apiSqlSessionFactory") SqlSessionFactory rdsSqlSessionFactory){
        return new SqlSessionTemplate(rdsSqlSessionFactory);
    }

    @Bean(name = "apiNamedParameterJdbcTemplate")
    @Primary
    public NamedParameterJdbcTemplate rdsNamedParameterJdbcTemplate(@Qualifier("shardingSphereDataSource") DataSource dynamicDataSource){
        return new NamedParameterJdbcTemplate(dynamicDataSource);
    }


    public static void checkInitialize() throws InterruptedException {
        if (!initialized) {
            lock.lock();
            try{
                condition.await(5000, TimeUnit.MILLISECONDS);
            }finally {
                lock.unlock();
            }

        }
    }

    public static boolean containLogicTable(String tableName){
        try {
            checkInitialize();
        } catch (InterruptedException e) {
        }
        return logicTables == null ? false : logicTables.contains(tableName);
    }

    public static Set<String> getLogicTables () {
        try {
            checkInitialize();
        } catch (InterruptedException e) {
        }
        return logicTables == null ? Collections.emptySet() : logicTables;
    }

然后，封装一个类，对外提供切换数据源的接口

package com.iwhalecloud.citybrain.common.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.shardingsphere.api.hint.HintManager;

import java.util.*;

@Slf4j
public class DynamicDataSourceContextHolder {

    private static final ThreadLocal<String> CONTEXT_HOLDER = new ThreadLocal<String>() {
        /**
         * 将 master 数据源的 key作为默认数据源的 key
         */
        @Override
        protected String initialValue() {
            return "master";
        }
    };

    /**
     * 数据源的 key集合，用于切换时判断数据源是否存在
     */
    public static List<Object> dataSourceKeys = new ArrayList<>();

    /**
     * 切换数据源
     * @param keys
     */
    public static void setDataSourceKey(String ... keys) {

        if (ArrayUtils.isNotEmpty(keys)){
            log.info("调用数据源部门id为："+ Arrays.toString(keys));
            Set<String> logicTables = ShardingJdbcConfig.getLogicTables();
            HintManager hintManager = HintManager.getInstance();
            ShardingSphereDataSourceSerialRecord shardingSphereDataSourceSerialRecord = ShardingSphereDataSourceSerialRecord.getInstance();
            if (CollectionUtils.isNotEmpty(logicTables)){
                for (String logicTable : logicTables){
                    for (String key : keys){
                        if (shardingSphereDataSourceSerialRecord.containDataSourceAliase(key)){
                            hintManager.addDatabaseShardingValue(logicTable, key);
                        }
                    }
                }
            }
            CONTEXT_HOLDER.set(keys[0]);
        }else{
            log.info("调用master数据源");
        }
    }

    public static void setDataSourceKeyByDepartmentIds (String tableName, String ... departmentIds){
        if (ArrayUtils.isEmpty(departmentIds) || StringUtils.isEmpty(tableName)){
            return;
        }
        HintManager hintManager = HintManager.getInstance();
        ShardingSphereDataSourceSerialRecord shardingSphereDataSourceSerialRecord = ShardingSphereDataSourceSerialRecord.getInstance();
        for (String departmentId : departmentIds){
            if (!shardingSphereDataSourceSerialRecord.containDataSourceAliase(departmentId)){
                continue;
            }
            String serialNum = shardingSphereDataSourceSerialRecord.getDataSourceSerialNum(departmentId);
            hintManager.addDatabaseShardingValue(tableName, serialNum);
        }
    }


    /**
     * 获取数据源
     * @return
     */
    public static String getDataSourceKey() {
        return CONTEXT_HOLDER.get();
    }

    /**
     * 重置数据源
     */
    public static void clearDataSourceKey() {
        HintManager.clear();
        CONTEXT_HOLDER.remove();
    }

    /**
     * 判断是否包含数据源
     * @param key 数据源key
     * @return
     */
    public static boolean containDataSourceKey(String key) {
        return dataSourceKeys.contains(key);
    }

    /**
     * 添加数据源keys
     * @param keys
     * @return
     */
    public static boolean addDataSourceKeys(Collection<? extends Object> keys) {
        return dataSourceKeys.addAll(keys);
    }

}

然后就可以通过DynamicDataSourceContextHolder.setDataSourceKey()切换数据源了，不同的数据源相同的表结果一条sql就可以搞定了