Springboot使用Redis发布订阅自动更新缓存数据源

背景

当项目有很多数据源的时候，通常会在启动的时候就把数据源连接加载缓存上，当数据源进行变更后如何自动实时将缓存的数据源进行更新呢？如果是单个项目直接调接口方法就行了，但是涉及到分布式多个系统呢？

解决方案：

使用Redis轻量级消息队列，它可以实现实时通知，实时状态更新等功能,配合AOP实现自动更新数据源状态。

下面结合代码写一个使用示例：

1.首先创建数据源对象

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;
import org.apache.commons.lang3.StringUtils;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 *
 * @author ws
 * @since 2022-08-12
 */
@Getter
@Setter
@ToString
@Accessors(chain = true)
@TableName("ed_datasource_info")
public class DatasourceInfo implements Serializable {

    private static final long serialVersionUID = 1L;

    @TableId(value = "id", type = IdType.AUTO)
    private Integer id;

    /**
     * 数据源编码
     */
    @TableField("datasource_code")
    private String datasourceCode;

    /**
     * 数据源名称
     */
    @TableField("datasource_name")
    private String datasourceName;

    /**
     * 数据源类型
     */
    @TableField("datasource_type")
    private String datasourceType;

    /**
     * 类型 0:数据库 1:Rest-api
     */
    @TableField("type")
    private Integer type;

    /**
     * 创建人
     */
    @TableField("creator")
    private String creator;

    /**
     * 模式
     */
    @TableField("schema_name")
    private String schemaName;


    @TableField("create_time")
    private Date createTime;

    @TableField("update_time")
    private Date updateTime;

    /**
     * 数据源连接信息
     */
    @TableField("link_json")
    private String linkJson;
 
}

2.初始化启动加载数据源

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.sztech.common.constant.DataSourceTypeEnum;
import com.sztech.entity.DatasourceInfo;
import com.sztech.service.DatasourceInfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class DataSourceRecovery implements InitializingBean {

    @Resource
    private DatasourceInfoService datasourceInfoService;

    @Override
    public void afterPropertiesSet() throws Exception {
        refresh();
    }

    private void refresh() throws Exception{
        this.refresh(null);
    }

    public void refresh(String sourceCode){
        QueryWrapper<DatasourceInfo> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("type", DataSourceTypeEnum.DB.getKey());
        if(StringUtils.isNotBlank(sourceCode)){
            queryWrapper.eq("datasource_code",sourceCode);
        }

        List<DatasourceInfo> list = datasourceInfoService.list(queryWrapper);
        if(CollectionUtils.isEmpty(list)){
            return;
        }

        CountDownLatch countDownLatch = new CountDownLatch(list.size());
        for(DatasourceInfo datasourceInfo : list){
            new Thread(new ReadloadThread(datasourceInfo, countDownLatch)).start();
        }

        try {
            countDownLatch.await(1,TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            log.error("数据源加载等待超时",e);
        }
    }

    /**
     * 多线程加载数据源,提高启动速度
     */
    static class ReadloadThread implements Runnable {

        private DatasourceInfo datasourceInfo;
        private CountDownLatch countDownLatch;

        public ReadloadThread() {
        }

        public ReadloadThread(DatasourceInfo datasourceInfo,CountDownLatch countDownLatch) {
            this.datasourceInfo = datasourceInfo;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                DataSourceContext.setClientMap(datasourceInfo);
                DataSourceContext.setConfigMap(datasourceInfo.getDatasourceCode(),datasourceInfo);
            }catch (Exception e){
                log.error("datasource:{},加载失败",datasourceInfo.getDatasourceCode(),e);
            }finally {
                countDownLatch.countDown();
            }
        }
    }
}

3.创建DataSourceContext,用于数据源缓存数据源连接

import com.sztech.core.tool.DBTool;
import com.sztech.entity.DatasourceInfo;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * User: wangsheng
 * Date: 2022-02-11
 * Time: 14:05
 */
public class DataSourceContext {
    /**
     * 客户端缓存
     */
    private final static Map<String, IClient> clientMap = new ConcurrentHashMap<>();

    /**
     * 数据源配置缓存
     */
    private final static Map<String, DatasourceInfo> configMap = new ConcurrentHashMap<>();

    public static void setClientMap(DatasourceInfo datasourceInfo) {
        if(clientMap.containsKey(datasourceInfo.getDatasourceCode())){
            try {
                clientMap.get(datasourceInfo.getDatasourceCode()).close();
            }catch (Exception ignored){
            }
        }
        clientMap.put(datasourceInfo.getDatasourceCode(),
                DBTool.buildClient(datasourceInfo));
    }

    public static void setConfigMap(String key, DatasourceInfo datasourceInfo) {
        configMap.put(key, datasourceInfo);
    }

    public static void removeClientMap(String key) {
        if(clientMap.containsKey(key)){
            try {
                clientMap.get(key).close();
            }catch (Exception ignored){
            }
        }
        clientMap.remove(key);
    }

    public static void removeConfigMap(String key) {
        configMap.remove(key);
    }

    public static IClient getClientMap(String key) {
        IClient client = clientMap.get(key);
        if(null == client){
            throw new RuntimeException(String.format("数据源编码:[%s]不存在或被删除...", key));
        }
        return client;
    }

    public static DatasourceInfo getConfigMap(String key) {
        DatasourceInfo datasourceInfo = configMap.get(key);
        if(null == datasourceInfo){
            throw new RuntimeException(String.format("数据源编码:[%s]不存在或被删除...", key));
        }

        return datasourceInfo;
    }
}

package com.sztech.core.tool;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.odps.Instance;
import com.sztech.common.constant.ResultEnum;
import com.sztech.common.exception.BizException;
import com.sztech.common.utils.ReflectionUtils;
import com.sztech.common.utils.SpringUtils;
import com.sztech.common.utils.ThreadPoolUtil;
import com.sztech.core.datasource.DataSourceContext;
import com.sztech.core.datasource.IClient;
import com.sztech.core.datasource.rdbms.RdbmsConfig;
import com.sztech.entity.*;
import com.sztech.pojo.dto.ColumnDto;
import com.sztech.pojo.dto.QueryTableDto;
import com.sztech.pojo.dto.TableDto;
import com.sztech.pojo.node.PartitionColumn;
import com.sztech.pojo.vo.*;
import com.sztech.service.CreateTableLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Description:
 * User: wangsheng
 * Date: 2022-08-12
 * Time: 16:59
 */
@Slf4j
public class DBTool {

    /**
     * 建立客户端
     */
    public static IClient buildClient(DatasourceInfo datasourceInfo) {
        IClient client = ReflectionUtils.getInstanceFromCache(datasourceInfo.getDatasourceType(), "type", IClient.class);
        return client.open(datasourceInfo);
    }

    /**
     * 测试数据源
     *
     * @return
     */
    public static boolean testSource(DatasourceInfo datasourceInfo) {
        IClient client = ReflectionUtils.getInstanceFromCache(datasourceInfo.getDatasourceType(), "type", IClient.class);
        return client.testSource(datasourceInfo);
    }

    public static List<String> getSchemas(DatasourceInfo datasourceInfo) {
        List<String> schemas = new ArrayList<>();
        Connection conn = null;
        try {
            IClient client = ReflectionUtils.getInstanceFromCache(datasourceInfo.getDatasourceType(), "type", IClient.class);
            Class.forName(client.driverName());
            String linkJson = datasourceInfo.getLinkJson();
            RdbmsConfig rdbmsConfig = JSONObject.parseObject(linkJson).toJavaObject(RdbmsConfig.class);
            conn = DriverManager.getConnection(rdbmsConfig.getJdbcUrl(), rdbmsConfig.getUsername(), rdbmsConfig.getDecodePassword());
            DatabaseMetaData metadata = conn.getMetaData();
            try (ResultSet resultSet = metadata.getSchemas()) {
                while (resultSet.next()) {
                    String schemaName = resultSet.getString("TABLE_SCHEM");
                    schemas.add(schemaName);
                }
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException ex) {
                    ex.printStackTrace();
                }
            }
        }
        return schemas;
    }

    /**
     * 获取驱动名称
     */
    public static String getDriverName(String datasourceType) {
        IClient client = ReflectionUtils.getInstanceFromCache(datasourceType, "type", IClient.class);
        return client.driverName();
    }

    /**
     * 获取表中列信息
     */
    public static List<ColumnDto> getColumns(String datasourceCode, String tableName) {
        return DataSourceContext.getClientMap(datasourceCode).getColumns(tableName);
    }

    /**
     * 获取表中分区列信息
     */
    public static List<String> getPartitionColumns(String datasourceCode, String tableName) {
        return DataSourceContext.getClientMap(datasourceCode).getPartitionColumns(tableName);
    }

    /**
     * 获取表信息
     */
    public static List<String> getTableNames(String datasourceCode, String tableNameLike) {
        return DataSourceContext.getClientMap(datasourceCode).getTableNames(tableNameLike);
    }

    /**
     * 获取表信息
     */
    public static List<TableDto> getTables(String datasourceCode) {
        return DataSourceContext.getClientMap(datasourceCode).getTables();
    }

    /**
     * 获取单个表信息
     */
    public static TableDto getTableByName(String datasourceCode, String tableName) {
        return DataSourceContext.getClientMap(datasourceCode).getTableByName(tableName);
    }

    /**
     * 获取单个表信息(创建时间，字段数)
     */
    public static TableDto getTableField(String datasourceCode, String tableName) {
        return DataSourceContext.getClientMap(datasourceCode).getTableField(tableName);
    }


    /**
     * 获取表信息（获取创建时间）
     *
     * @param dto
     * @return
     */
    public static TableInfoVo getTableData(QueryTableDto dto) {
        IClient client = DataSourceContext.getClientMap(dto.getDataSourceCode());
        return client.getTableInfo(dto.getTableName());
    }


    /**
     * 根据字段type建表
     */
    public static void createTableByColumns(List<ColumnDto> columnDtos, String tableName, String datasourceCode) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        List<String> sqls = client.buildTableSql(columnDtos, tableName, true);
        log.info("执行建表语句为：" + JSON.toJSONString(sqls));
        sqls.forEach(s -> client.executeCommandSyn(s, new HashMap<>()));
    }

    /**
     * 根据字段type建表
     */
    public static void createTableByNotTransformedColumns(List<ColumnDto> columnDtos, String tableName, String datasourceCode) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        List<String> sqls = client.buildTableSql(columnDtos, tableName, false);
        log.info("执行建表语句为：" + JSON.toJSONString(sqls));
        sqls.forEach(s -> client.executeCommandSyn(s, new HashMap<>()));
    }

    /**
     * 创建索引
     * 注: oracle 索引名在整个库里必须唯一 否则建立失败
     *
     * @param datasourceCode 数据源编码
     * @param tableName      表名
     * @param filedNames     filed1,filed2...
     * @param unique         唯一
     */
    public static void createIndex(String datasourceCode, String tableName, String filedNames, Boolean unique) {
        DataSourceContext.getClientMap(datasourceCode).createIndex(tableName, filedNames, unique);
    }

    /**
     * sql校验
     *
     * @param datasourceCode
     * @param sql
     * @param sourceType
     * @return
     */
    public static Map<String, Object> checkSql(String datasourceCode, String sql, String sourceType) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        return client.checkSql(sql, sourceType);
    }

    /**
     * 根据sql创建表
     *
     * @param datasourceCode
     * @param sql
     */
    public static void createTableWithSql(String datasourceCode, String sql) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        log.info("执行建表语句为：" + JSON.toJSONString(sql));
        client.executeCommandSyn(sql, new HashMap<>());
//        DataSourceContext.getClientMap(datasourceCode).createTableWithSql(sql);
    }

    /**
     * 删除表
     *
     * @param datasourceCode
     * @param tableName
     */
    public static void dropTable(String datasourceCode, String tableName) {
        DataSourceContext.getClientMap(datasourceCode).dropTable(tableName);
    }

    /**
     * 单表查询数据
     */
    public static List<Map<String, Object>> selectDataFromTable(String datasourceCode, List<DataTableColumn> columns, String tableName, String search, Integer limit) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql = client.getSelectSql(columns, tableName, search, limit);
        log.info("执行语句：" + selectSql);
        return client.selectDataFromTable(selectSql, null);
    }

    /**
     * 单表查询数据
     */
    public static List<Map<String, Object>> selectFromTable(String datasourceCode, List<FormColumn> columns, List<FormColumn> searchColumns, String tableName, String search, Integer pageNum, Integer pageSize, MapSqlParameterSource params) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql = client.getFormSelectSql(columns, searchColumns, tableName, search, pageNum, pageSize, params);
        log.info("执行语句：" + selectSql);
        return client.selectDataFromTable(selectSql, params);
    }

    /**
     * 单表查询数据
     */
    public static List<Map<String, Object>> selectFromForBackUp(String datasourceCode, List<FormColumn> columns, List<FormColumn> searchColumns, String tableName, String search, Integer pageNum, Integer pageSize, MapSqlParameterSource params) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql = client.selectFromForBackUp(columns, searchColumns, tableName, search, pageNum, pageSize, params);
        log.info("执行语句：" + selectSql);
        return client.selectDataFromTable(selectSql, params);
    }

    /**
     * 单表查询数据
     */
    public static List<Map<String, Object>> selectFromFile(String datasourceCode, List<FormColumn> columns, List<FormColumn> searchColumns, String tableName, String search, Integer pageNum, Integer pageSize, MapSqlParameterSource params) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql = client.getFormSelectSqlForFile(columns, searchColumns, tableName, search, pageNum, pageSize, params);
        log.info("执行语句：" + selectSql);
        return client.selectDataFromTable(selectSql, params);
    }
    /**
     * 查询单表是否存在文件名
     */
    public static List<Map<String, Object>> getExistOldName(String datasourceCode, String tableName, String search) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql = client.getExistOldName( tableName, search);
        log.info("执行语句：" + selectSql);
        return client.selectDataFromTable(selectSql, null);
    }
    /**
     * 单表查询数据(查询归集表专门使用)
     */
    public static List<Map<String, Object>> selectCollectTable(CollectConditionVo vo) {
        IClient client = DataSourceContext.getClientMap(vo.getDatasourceCode());
        // 获取查询语句
        String selectSql = client.getCollectTable(vo);
        log.info("执行语句：" + selectSql);
        return client.selectDataFromTable(selectSql, vo.getParams());
    }

    /**
     * 单表查询数据量
     */
    public static Map<String, Object> getFormCount(String datasourceCode, List<FormColumn> columns, List<FormColumn> searchColumns, String tableName, String search, MapSqlParameterSource params) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql = client.getCountSql(columns, searchColumns, tableName, search, params);
        log.info("执行语句：" + selectSql);
        return client.getCount(selectSql, params);
    }

    /**
     * 查询区县库表的数据量
     */
    public static Map<String, Object> getCountryCount(String datasourceCode, String tableName,  MapSqlParameterSource params) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql ="select count(1) as count from "+tableName;
        log.info("执行语句：" + selectSql);
        return client.getCount(selectSql, params);
    }

    public static Map<String, Object> getFormCountForFile(String datasourceCode, List<FormColumn> columns, List<FormColumn> searchColumns, String tableName, String search, MapSqlParameterSource params) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        // 获取查询语句
        String selectSql = client.getCountSqlForFile(columns, searchColumns, tableName, search, params);
        log.info("执行语句：" + selectSql);
        return client.getCount(selectSql, params);
    }

    /**
     * 查询表数据量
     */
    public static Long getTableRows(String datasourceCode, String tableName) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        return client.getTableRows(tableName);
    }

    /**
     * 查询表对应分区数据量
     */
    public static Long getTablePartitionRows(String datasourceCode, String tableName, List<PartitionColumn> partitionColumns) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        return client.getTablePartitionRows(tableName, partitionColumns);
    }

    /**
     * 查询表数据量
     */
    public static Integer getTablePhysicalSize(String datasourceCode, String tableName) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        return client.getPhysicalSize(tableName);
    }

    /**
     * 获取表最大值
     *
     * @param datasourceCode 数据源编码
     * @param tableName      表名
     * @param incColumnName  自增列名
     * @return {@link Integer}
     */
    public static Object getMaxValue(String datasourceCode, String tableName, String incColumnName, String condition) {
        return DataSourceContext.getClientMap(datasourceCode).getMaxValue(tableName, incColumnName, condition);
    }

    public static Object getMaxValue(String datasourceCode, String schema, String tableName, String incColumnName, String condition) {
        return DataSourceContext.getClientMap(datasourceCode).getMaxValue(schema, tableName, incColumnName, condition);
    }


    public static Object getMaxTime(String datasourceCode, String schema, String tableName, String incColumnName, String tongId,String condition) {
        return DataSourceContext.getClientMap(datasourceCode).getMaxTime(schema, tableName, incColumnName,tongId, condition);
    }


    /**
     * 字段存在
     *
     * @param datasourceCode 数据源编码
     * @param tableName      表名
     * @param fieldName      字段名
     * @return {@link Boolean}
     */
    public static Boolean fieldExist(String datasourceCode, String tableName, String fieldName) {
        List<ColumnDto> columns = getColumns(datasourceCode, tableName);
        return columns.stream().anyMatch(s -> s.getName().equalsIgnoreCase(fieldName));
    }

    /**
     * 数据预览 获取前十条
     *
     * @return
     */
    public static String dataView(String datasourceCode, String tableName, String condition) {
        return DataSourceContext.getClientMap(datasourceCode).dataView(tableName, condition);
    }

    /**
     * 创建分区临时表
     * odps适用
     */
    public static void createPartitionedTableByColumns(List<ColumnDto> columnDtos, String tableName, String tableComment, String partitionedField, String datasourceCode) {
        DataSourceContext.getClientMap(datasourceCode).createPartitionedTableByColumns(columnDtos, tableName, tableComment, partitionedField);
    }

    /**
     * 同步执行命令
     */
    public static void executeCommandSyn(String datasourceCode, String command, Map<String, Object> params) {
        DataSourceContext.getClientMap(datasourceCode).executeCommandSyn(command, params);
    }

    /**
     * 异步执行命令
     * odps适用
     */
    public static Instance executeCommandASyn(String datasourceCode, String command, Map<String, Object> params) {
        return DataSourceContext.getClientMap(datasourceCode).executeCommandASyn(command, params);
    }

    /**
     * 是否有导出权限
     * odps适用
     *
     * @param datasourceCode 数据源编码
     * @param tableName      表名
     * @return {@link Boolean}
     */
    public static Boolean exportEnable(String datasourceCode, String tableName) {
        return DataSourceContext.getClientMap(datasourceCode).exportEnable(tableName);
    }

    /**
     * 插入单条数据
     *
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer insert(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).insert(vo);
    }

    /**
     * 批量插入数据
     *
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer[] betchInsert(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).betchInsert(vo);
    }

    /**
     * 批量插入数据
     *
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer[] betchInsertByConnection(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).betchInsertByConnection(vo);
    }

    /**
     * 这个方法不需要分装参数，直接传字段名称list就好了
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer[] betchInsertForCommom(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).betchInsertForCommom(vo);
    }

    /**
     * 删除数据
     *
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer delete(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).delete(vo);
    }

    /**
     * 这个删除方法可以自定义条件服号
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer deleteForCommon(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).deleteForCommon(vo);
    }


    public static Integer deleteForFile(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).deleteForFile(vo);
    }

    public static String deleteForPre(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).deleteForPre(vo);
    }

    /**
     * 修改数据
     *
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer update(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).update(vo);
    }



    /**
     * 修改数据
     *
     * @param datasourceCode
     * @param vo
     * @return
     */
    public static Integer updateForFile(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).updateForFile(vo);
    }

    /**
     * 获取表单基本信息
     *
     * @param vo
     * @return
     */
    public static TableMetaDataVo getTableBasicInfo(String datasourceCode, FormTableVo vo) {
        return DataSourceContext.getClientMap(datasourceCode).getTableBasicInfo(vo);
    }

    /**
     * 根据字段type建表
     */
    public static void createCollectTable(List<CatalogColumnInfo> columnDtos, String tableName, String datasourceCode, String tableComment, Boolean ifPartition) {
        IClient client = DataSourceContext.getClientMap(datasourceCode);
        List<String> sqls = client.buildTableSqlForCollect(columnDtos, tableName, tableComment, ifPartition);
        log.info("执行建表语句为：" + JSON.toJSONString(sqls));

        try {
            sqls.forEach(s -> client.executeCommandSyn(s, new HashMap<>()));
        } catch (Exception e) {
            e.printStackTrace();
            String message = e.getMessage();
            if (e instanceof BizException) {
                BizException exception = (BizException) e;
                message = exception.getMsg();
            }

            log.error("建表错误=======================>{}:", message);
            ThreadPoolExecutor instance = ThreadPoolUtil.instance();
            String finalMessage = message;
            instance.submit(() -> {
                CreateTableLog createTableLog = new CreateTableLog();
                createTableLog.setErrorLog(finalMessage);
                createTableLog.setParams(JSON.toJSONString(sqls));
                createTableLog.setCode(tableName);
                CreateTableLogService createTableLogService = SpringUtils.getBean(CreateTableLogService.class);
                createTableLogService.save(createTableLog);
            });
            throw new BizException(ResultEnum.ERROR.getCode(), "建表失败请联系管理员");
        }

    }

    /**
     * 根据字段type建表
     */
    public static void updateCollectTable(CreateCollectVo vo) {
        IClient client = DataSourceContext.getClientMap(vo.getDatasourceCode());
        List<String> sqls = client.buildTableSqlForUpdate(vo);
        log.info("执行更新表语句为：" + JSON.toJSONString(sqls));

        try {
            sqls.forEach(s -> client.executeCommandSyn(s, new HashMap<>()));
        } catch (Exception e) {
            e.printStackTrace();
            String message = e.getMessage();
            if (e instanceof BizException) {
                BizException exception = (BizException) e;
                message = exception.getMsg();
            }
            log.error("建表错误=======================>{}:", message);
            ThreadPoolExecutor instance = ThreadPoolUtil.instance();
            String finalMessage = message;
            instance.submit(() -> {
                CreateTableLog createTableLog = new CreateTableLog();
                createTableLog.setErrorLog(finalMessage);
                createTableLog.setParams(JSON.toJSONString(sqls));
                createTableLog.setCode(vo.getTableName());
                CreateTableLogService createTableLogService = SpringUtils.getBean(CreateTableLogService.class);
                createTableLogService.save(createTableLog);
            });
            log.info("建表失败了开始准备抛出了-------------------------------------->");
            throw new BizException(ResultEnum.ERROR.getCode(), "建表失败请联系管理员");
        }
    }

    /**
     * 获取数据源下所有表信息（包括表名，表字段数，表创建时间）
     *
     * @param datasourceCode
     * @param tableNameLike
     * @return
     */
    public static List<TableDto> getTablesDetail(String datasourceCode, String tableNameLike, Integer start, Integer pageSize, String specifyTableName) {
        return DataSourceContext.getClientMap(datasourceCode).getTablesDetail(tableNameLike, start, pageSize, specifyTableName);
    }

    /**
     * 获取表数量
     * @param datasourceCode
     * @param tableName
     * @return
     */
    public static Long getTableCountSchema(String datasourceCode, String tableName) {
        return DataSourceContext.getClientMap(datasourceCode).getTableCountSchema(tableName);
    }

    public static Integer getTableColumnCount(String dataSourceCode, String tableName) {
        return DataSourceContext.getClientMap(dataSourceCode).getTableColumnCount(tableName);
    }

    public static Integer getPreTableColumnCount(String dataSourceCode, String tableName) {
        return DataSourceContext.getClientMap(dataSourceCode).getPreTableColumnCount(tableName);
    }

    /**
     * 获取符号
     * @return
     */
    public static String getSymbol(String datasourceCode) {
        return DataSourceContext.getClientMap(datasourceCode).getSymbol();
    }

}



import lombok.extern.slf4j.Slf4j;
import org.reflections.Reflections;
import java.lang.reflect.Modifier;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class ReflectionUtils {

    private static final Map<String, Set<?>> clazzMap = new ConcurrentHashMap<>();
    private static final ReentrantLock clazzLock = new ReentrantLock();

    /**
     * 通过反射获取接口/抽象类的所有实现类
     * 通过缓存类信息减少查找时间
     * 接口与抽象类必须放在实现类的同级目录或者父目录
     */
    @SuppressWarnings("unchecked")
    public static <T> Set<Class<? extends T>> getReflections(Class<T> clazz) {
        if (clazzMap.containsKey(clazz.getName())) {
            return (Set<Class<? extends T>>) clazzMap.get(clazz.getName());
        }

        try {
            clazzLock.lock();
            if (clazzMap.containsKey(clazz.getName())) {
                return (Set<Class<? extends T>>) clazzMap.get(clazz.getName());
            }

            Reflections reflections = new Reflections(clazz.getPackage().getName());
            Set<Class<? extends T>> subTypesOf = reflections.getSubTypesOf(clazz);
            clazzMap.put(clazz.getName(), subTypesOf);
            return subTypesOf;
        } catch (Exception e) {
            log.error("getReflections error", e);
        } finally {
            clazzLock.unlock();
        }

        return new HashSet<>();
    }


    /**
     * 通过反射获取新对象
     * @param type type
     * @param methodName methodName
     * @param clazz clazz
     * @return <T>
     */
    public static <T> T getInstance(String type, String methodName, Class<T> clazz) {
        Set<Class<? extends T>> set = getReflections(clazz);
        for (Class<? extends T> t : set) {
            try {
                //排除抽象类
                if (Modifier.isAbstract(t.getModifiers())) {
                    continue;
                }
                Object obj = t.getMethod(methodName).invoke(t.newInstance());
                if (type.equalsIgnoreCase(obj.toString())) {
                    return t.newInstance();
                }
            } catch (Exception e) {
                log.error("getInstance error", e);
            }
        }

        throw new RuntimeException("implement class not exist");
    }

    /**
     * 通过反射获取新对象
     * @param type type
     * @param methodName methodName
     * @param clazz clazz
     * @return <T>
     */
    public static <T> T getInstanceFromCache(String type, String methodName, Class<T> clazz) {
        return getInstance(type, methodName, clazz);
    }

}

client客户接口端适配多种数据源


import com.ws.websocket.entity.DatasourceInfo;


/**
 * Description:
 * User: wangsheng
 * Date: 2022-12-30
 * Time: 10:31
 */
public interface IClient {
    /**
     * 连接数据源
     *
     * @param dataSourceInfo 数据源信息
     * @return {@link IClient}
     */
    IClient open(DatasourceInfo dataSourceInfo);
    /**
     * 关闭数据源
     */
    void close();
    /**
     * 驱动类型
     *
     * @return
     */
    String driverName();
    /**
     * 数据源类型
     *
     * @return {@link String}
     */
    String type();
    /**
     * 测试数据源
     *
     * @param datasourceInfo
     * @return
     */
    boolean testSource(DatasourceInfo datasourceInfo);

}


import com.ws.websocket.entity.DatasourceInfo;
//公共查询
public abstract class AbsClient implements IClient  {
    protected DatasourceInfo datasourceInfo;
}

package com.ws.websocket.util;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.ws.websocket.entity.DatasourceInfo;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Properties;

@Slf4j
public abstract class AbsRdbmsClient extends AbsClient {

    protected DruidDataSource druidDataSource;
    @Override
    public IClient open(DatasourceInfo datasourceInfo) {
        RdbmsConfig rdbmsConfig = JSONObject.parseObject(datasourceInfo.getLinkJson()).toJavaObject(RdbmsConfig.class);
        DruidDataSource druidDataSource = new DruidDataSource();
        druidDataSource.setInitialSize(5);
        druidDataSource.setMinIdle(30);
        druidDataSource.setMaxActive(300);
        druidDataSource.setMaxWait(10000);

        druidDataSource.setBreakAfterAcquireFailure(true);// 跳出重试循环
        druidDataSource.setConnectionErrorRetryAttempts(3);// 重试三次
        druidDataSource.setTimeBetweenConnectErrorMillis(3000);
        druidDataSource.setLoginTimeout(3);

        druidDataSource.setUrl(rdbmsConfig.getJdbcUrl());
        druidDataSource.setDriverClassName(driverName());
        druidDataSource.setUsername(rdbmsConfig.getUsername());
        //解密
        //  druidDataSource.setPassword(RsaUtils.decode(rdbmsConfig.getPassword()));
        druidDataSource.setPassword(rdbmsConfig.getPassword());
        // 设置 MetaUtil 工具类所需参数
        Properties properties = new Properties();

        properties.put("remarks", "true");
        properties.put("useInformationSchema", "true");
        druidDataSource.setConnectProperties(properties);

        this.druidDataSource = druidDataSource;
        this.datasourceInfo = datasourceInfo;
        return this;
    }

    @Override
    public void close() {
        druidDataSource.close();
    }


    @Override
    public boolean testSource(DatasourceInfo datasourceInfo) {
        Connection connection = null;
        try {
            Class.forName(driverName());
            String linkJson = datasourceInfo.getLinkJson();
            RdbmsConfig rdbmsConfig = JSONObject.parseObject(linkJson).toJavaObject(RdbmsConfig.class);
            connection = DriverManager.getConnection(rdbmsConfig.getJdbcUrl(), rdbmsConfig.getUsername(), rdbmsConfig.getPassword());
            // 有效
            if (connection.isValid(3)) {
                return true;
            } else {
                return false;
            }
        } catch (SQLException e) {
            log.error("数据源测试失败", e);
            return false;
        } catch (ClassNotFoundException e) {
            log.error("未找到驱动信息：{}", driverName());
            return false;
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
    @Data
    class RdbmsConfig  {

        private String jdbcUrl;

        private String username;

        private String password;



        public void setSSL() {
            String lowerCase = this.jdbcUrl.toLowerCase();
            if (!lowerCase.contains("usessl")) {
                if (this.jdbcUrl.contains("?")) {
                    this.jdbcUrl = this.jdbcUrl + "&useSSL=false";
                } else {
                    this.jdbcUrl = this.jdbcUrl + "?useSSL=false";
                }
            }
        }
    }
}

import com.alibaba.fastjson.JSONObject;
import com.ws.websocket.entity.DatasourceInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

@Slf4j
public class DmClient extends AbsRdbmsClient {
    private String schema;
    @Override
    public String type() {
        return "DMDB";
    }

    @Override
    public String driverName() {
        return "dm.jdbc.driver.DmDriver";
    }
    @Override
    public IClient open(DatasourceInfo datasourceInfo) {
        RdbmsConfig commonLinkParams = JSONObject.parseObject(datasourceInfo.getLinkJson()).toJavaObject(RdbmsConfig.class);
        this.schema = StringUtils.isNotBlank(datasourceInfo.getSchemaName()) ? datasourceInfo.getSchemaName() : commonLinkParams.getUsername().toUpperCase();
        datasourceInfo.setSchemaName(schema);
        return super.open(datasourceInfo);
    }
    @Override
    public void close() {

    }

    @Override
    public boolean testSource(DatasourceInfo datasourceInfo) {
        return false;
    }
}

4.创建redis订阅数据源操作频道配置

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

/**
 * @Author: wangsheng
 * @Data: 2022/8/16 16:40
 */
@Slf4j
@Configuration
public class RedisListenerConfig {
    /**
     * 订阅数据源操作频道
     *
     * @param connectionFactory connectionFactory
     * @param dataSourceMonitor 数据源监视器
     * @return RedisMessageListenerContainer
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            DataSourceMonitor dataSourceMonitor){
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(dataSourceMonitor, new PatternTopic("DATASOURCE_CHANNEL"));
        log.info(dataSourceMonitor.getClass().getName() + " 订阅频道 {}", "DATASOURCE_CHANNEL");
        return container;
    }
}

5.redis监听数据源操作


import com.alibaba.fastjson.JSONObject;
import com.ws.websocket.entity.DatasourceInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;

/**
 * Description: redis监听数据源操作
 * User: wangsheng
 * Date: 2022-08-12
 * Time: 17:07
 */
@Slf4j
@Component
public class DataSourceMonitor implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] bytes) {
        JSONObject box = JSONObject.parseObject(new String(message.getBody(), StandardCharsets.UTF_8));
        String operation = box.getString("key");
        if ("SAVE_OR_UPDATE".equals(operation)) {
            // 更新 DataSourceContext
            DatasourceInfo datasourceInfo = box.getObject("value", DatasourceInfo.class);
            if (datasourceInfo.getType().equals(0)) {
                String datasourceCode = datasourceInfo.getDatasourceCode();
                DataSourceContext.setConfigMap(datasourceCode, datasourceInfo);
                DataSourceContext.setClientMap(datasourceInfo);
                log.info("redis 监听到数据源 {} 新增或更新，更新 DataSourceContext 完成", datasourceCode);
            }

        } else {
            String datasourceCode = box.getString("value");
            // 更新 DataSourceContext
            DataSourceContext.removeConfigMap(datasourceCode);
            DataSourceContext.removeClientMap(datasourceCode);
            log.info("redis 监听到数据源 {} 删除，更新 DataSourceContext 完成", datasourceCode);
        }

    }

}

6.创建AOP自动监听数据源变化


import com.alibaba.fastjson.JSONObject;
import com.ws.websocket.entity.DatasourceInfo;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;


/**
 * @Author: wangsheng
 * @Data: 2022/8/15 16:37
 */
@Slf4j
@Aspect
@Component
public class DatasourceAspect {
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    /**
     * 新增或编辑数据源时发布 Redis 消息
     */
    @AfterReturning(value = "execution(* com.ws.service.DatasourceInfoService.saveOrUpdateDatasourceInfo(..))", returning = "datasourceInfo")
    public void saveOrUpdate(JoinPoint joinPoint, DatasourceInfo datasourceInfo) {
        HashMap<String, Object> box = new HashMap<>(4);
        box.put("key", "SAVE_OR_UPDATE");
        box.put("value", datasourceInfo);
        // 发布 Redis 消息
        stringRedisTemplate.convertAndSend("DATASOURCE_CHANNEL",JSONObject.toJSONString(box));
        log.info("新增或更新数据源 {} 方法切面发布 Redis 消息完成", datasourceInfo.getDatasourceCode());
    }

    /**
     * 删除数据源时发布 Redis 消息
     */
    @AfterReturning(value = "execution(* com.ws.service.DatasourceInfoService.deleteDatasourceInfo(..))", returning = "datasourceCode")
    public void delete(JoinPoint joinPoint, String datasourceCode) {
        Map<String, Object> box = new HashMap<>(4);
        box.put("key", "DELETE");
        box.put("value", datasourceCode);
        // 发布 Redis 消息
        stringRedisTemplate.convertAndSend("DATASOURCE_CHANNEL", JSONObject.toJSONString(box));
        log.info("删除数据源 {} 方法切面发布Redis消息完成", datasourceCode);
    }
}

这样就解决了数据源连接信息自动加载更新同步的问题，但还是有个问题，当数据源重启后，缓存的连接信息会失效，且AOP无法监听到数据源重启变动，这个时候还需要一个定时任务对数据源进行连接测试，如果失效则重新连接缓存上。

7.创建定时任务

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ws.websocket.entity.DatasourceInfo;
import com.ws.websocket.service.DatasourceInfoService;
import com.ws.websocket.util.DBTool;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;

@Component
@RequiredArgsConstructor
@Slf4j
public class DataSourceRetryConnectSchedule {
    @Resource
    private DatasourceInfoService datasourceInfoService;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    //每2小时执行一次
    @Scheduled(cron = "0 0 */2 * * ?")
    public void RetryConnect() {
        log.info("开始监测数据源连接");
        QueryWrapper<DatasourceInfo> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("type", 0);

        List<DatasourceInfo> list = datasourceInfoService.list(queryWrapper);
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        for (DatasourceInfo datasourceInfo : list) {
            Boolean bb = DBTool.testSource(datasourceInfo);
            if (!bb) {
                log.info("数据源重连{}"+datasourceInfo.getDatasourceName());
                HashMap<String, Object> box = new HashMap<>(4);
                box.put("key", "SAVE_OR_UPDATE");
                box.put("value", datasourceInfo);
                // 发布 Redis 消息
                stringRedisTemplate.convertAndSend("DATASOURCE_CHANNEL", JSONObject.toJSONString(box));
            }
        }
    }
}