一、背景
当我们一个数据库存储量已经不满足需求时，需要进行多数据源存储。在不同的数据源中，就会引发的多数据源查询的一系列问题
二、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
然后我们要通过这张表获取所有的部门的数据源，并且初始化到线程池中，代码如下：
/**
* @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就可以搞定了