一个是用户数据库，跟用户挂钩，包含一些除了数据库连接信息之外的其他用户数据
public class ExternalDataBaseDO implements Serializable {
//数据库id
@Id
private String id;
//数据源名称
private String connectName;
//数据库名
private String dataBaseName;
//登录名
private String userName;
//密码
private String passWord;
//连接地址
private String host;
//连接地址
private String port;
//数据库类型
private String databaseType;
//数据源提供者
private String provider;
...
}
一个是真实唯一的数据库连接配置，数据来自用户信息，根据host/basename和数据库类型确定唯一
public class DatabaseConfig {
private DatabaseType databaseType;
private String host;
private int port;
private String databaseName;
private String username;
private String password;
public DatabaseConfig(DatabaseType databaseType, String host, int port, String databaseName, String username, String password) {
this.databaseType = databaseType;
this.host = host;
this.port = port;
this.databaseName = databaseName;
this.username = username;
this.password = password;
}
public DatabaseConfig(ExternalDataBaseDO dataBaseDO) {
this.databaseName = dataBaseDO.getDataBaseName();
this.host = dataBaseDO.getHost();
this.port = Integer.parseInt(dataBaseDO.getPort().trim());
this.username = dataBaseDO.getUserName();
this.password = dataBaseDO.getPassWord();
switch (dataBaseDO.getDatabaseType()) {
case "mssql":
this.databaseType = DatabaseType.SQLSERVER;
if (StrUtil.isBlank(dataBaseDO.getDataBaseName())) {
//默认数据库master
this.databaseName = "master";
}
break;
case "postgresql":
this.databaseType = DatabaseType.POSTGRESQL;
if (StrUtil.isBlank(dataBaseDO.getDataBaseName())) {
//默认数据库postgres
this.databaseName = "postgres";
}
break;
default:
throw new MayException(CodeMsg.ERROR_DATABASE_TYPE);
}
}
public String getUrl() {
switch (databaseType) {
case SQLSERVER:
return String.format(DatabaseType.SQLSERVER.getJdbcUrlTemplate(),
host, port, databaseName);
case POSTGRESQL:
return String.format(DatabaseType.POSTGRESQL.getJdbcUrlTemplate(),
host, port, databaseName);
default:
throw new IllegalArgumentException("Unsupported database type: " + databaseType);
}
}
}
作用是根据数据库类型得到驱动和url
public enum DatabaseType {
MYSQL("com.mysql.jdbc.Driver", "jdbc:mysql://{host}/{database}"),
SQLSERVER("com.microsoft.sqlserver.jdbc.SQLServerDriver", "jdbc:sqlserver://%s:%d;DatabaseName=%s;encrypt=false"),
POSTGRESQL("org.postgresql.Driver", "jdbc:postgresql://%s:%d/%s?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=Asia/Shanghai&zeroDateTimeBehavior=CONVERT_TO_NULL&allowMultiQueries=true");
private final String driverClass;
private final String jdbcUrlTemplate;
DatabaseType(String driverClass, String jdbcUrlTemplate) {
this.driverClass = driverClass;
this.jdbcUrlTemplate = jdbcUrlTemplate;
}
public String getDriverClass() {
return driverClass;
}
public String getJdbcUrlTemplate() {
return jdbcUrlTemplate;
}
}
一个数据库配置一个连接池，这里使用的org.apache.commons.dbcp.BasicDataSource，其他连接池也行
public class ConnectionPool {
private final DatabaseConfig databaseConfig;
private final BasicDataSource dataSource;
public ConnectionPool(DatabaseConfig databaseConfig) {
this.databaseConfig = databaseConfig;
this.dataSource = new BasicDataSource();
this.dataSource.setDriverClassName(databaseConfig.getDatabaseType().getDriverClass());
this.dataSource.setUrl(databaseConfig.getUrl());
this.dataSource.setUsername(databaseConfig.getUsername());
this.dataSource.setPassword(databaseConfig.getPassword());
this.dataSource.setInitialSize(1); // 设置核心连接数为1 初始连接数
this.dataSource.setMaxActive(10); // 设置最大连接数为10 最大连接数
this.dataSource.setMaxWait(3000); // 设置最大连接等待时间毫秒 3秒
this.dataSource.setMinEvictableIdleTimeMillis(10*60000); // 设置最小可空闲时间(10分钟)
this.dataSource.setTimeBetweenEvictionRunsMillis(10*60000); // 检测空闲连接的时间间隔毫秒
}
public Connection getConnection() {
try {
System.out.println("已用连接数:" + dataSource.getNumActive());
System.out.println("空闲连接数:" + dataSource.getNumIdle());
return dataSource.getConnection();
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
public DatabaseConfig getDatabaseConfig() {
return databaseConfig;
}
public BasicDataSource getDataSource() {
return dataSource;
}
}
负责维护数据库连接与用户数据库关系
public class DatabaseConnection {
/**
* 数据库唯一标识字符串(类型+ip+端口+用户名)：数据库连接池
*/
private static final Map<String, ConnectionPool> CONNECTION_FACTORIES = new ConcurrentHashMap<>();
/**
* 用户数据库配置id: 数据库唯一对象
*/
private static final Map<String, DatabaseConfig> USERDBMAP = new ConcurrentHashMap<>();
/**
* 用户数据库Id获取数据库连接
*
* @param userDbId
* @return
*/
public static Connection getConnection(String userDbId) {
DatabaseConfig databaseConfig = USERDBMAP.get(userDbId);
if (databaseConfig == null) {
return null;
}
return getConnection(databaseConfig);
}
/**
* 缓存所有用户数据库连接
*
* @param dataBaseDO
* @return
*/
public static boolean addConnection(ExternalDataBaseDO dataBaseDO) {
DatabaseConfig databaseConfig = new DatabaseConfig(dataBaseDO);
String key = generateKey(databaseConfig);
if (CONNECTION_FACTORIES.get(key) == null) {
//创建连接池
try {
ConnectionPool connectionPool = new ConnectionPool(databaseConfig);
Connection connection = connectionPool.getConnection();
log.info("数据库已连接 {}:{}", dataBaseDO.getHost(), dataBaseDO.getDataBaseName());
CONNECTION_FACTORIES.put(key, connectionPool);
connection.close();
} catch (SQLException e) {
log.error("添加数据库时连接失败：{}", databaseConfig.toString());
return false;
}
}
//添加用户数据库映射
USERDBMAP.put(dataBaseDO.getId(), databaseConfig);
return true;
}
/**
* 根据数据库唯一对象获取数据库连接
*
* @param databaseConfig
* @return
*/
public static Connection getConnection(DatabaseConfig databaseConfig) {
String key = generateKey(databaseConfig);
ConnectionPool connectionPool = CONNECTION_FACTORIES.get(key);
//连接池为空则创建连接
if (connectionPool == null) {
synchronized (CONNECTION_FACTORIES) {
connectionPool = CONNECTION_FACTORIES.get(key);
if (connectionPool == null) {
connectionPool = new ConnectionPool(databaseConfig);
CONNECTION_FACTORIES.put(key, connectionPool);
}
}
}
return connectionPool.getConnection();
}
public static String generateKey(DatabaseConfig databaseConfig) {
return databaseConfig.getDatabaseType().name() + "_" + databaseConfig.getHost() + "_" + databaseConfig.getPort() + "_" + databaseConfig.getDatabaseName();
}
/**
* 查询所有连接池列表
*
* @return
*/
public static List<ConnectionPool> getAllConnectionPools() {
return new ArrayList<>(CONNECTION_FACTORIES.values());
}
/**
* 查询用户数据库信息
*
* @return
*/
public static DatabaseConfig getUserDbInfo(String userDbId) {
return USERDBMAP.get(userDbId);
}
// 私有构造函数，避免在外部创建实例
private DatabaseConnection() {
}
}
其中 addConnection 方法是用户添加数据库时，保存用户数据库对象的同时建立数据库连接池
到这一步核心功能就已经实现好了，简单测试一下
public static void main(String[] args) throws SQLException {
// 构建 DatabaseConfig ，切换DatabaseType和数据库检查不同类型数据库是否都能连接正常
DatabaseConfig sqlserverConfig = new DatabaseConfig(
DatabaseType.SQLSERVER,"192.168.10.153",1433,"myits","sa","admin888888"
);
// 获取连接
Connection connection = DatabaseConnection.getConnection(sqlserverConfig);
Connection connection2 = DatabaseConnection.getConnection(sqlserverConfig);
Connection connection3 = DatabaseConnection.getConnection(sqlserverConfig);
// 使用连接执行 SQL 查询操作
PreparedStatement statement = connection.prepareStatement("SELECT * FROM STB WHERE OBJECTID=?");
statement.setInt(1, 1);
ResultSet resultSet = statement.executeQuery();
while (resultSet.next()) {
int id = resultSet.getInt("OBJECTID");
String name = resultSet.getString("XIAN");
String age = resultSet.getString("HXLX");
System.out.println("id: " + id + ", name: " + name + ", age: " + age);
}
// 关闭连接并释放连接到连接池中
statement.close();
connection.close();
Connection connection4 = DatabaseConnection.getConnection(sqlserverConfig);
}
上一步测试没问题就可以根据业务需求封装一个查询管理工具类，以后都通过这个管理类用用户数据库id来执行sql，不用在意数据库类型和连接信息
public class DatabaseQueryManage {
/**
* 查询表数据量
*
* @param databaseId
* @param tableName
* @return
*/
public static long getTotalNum(String databaseId, String tableName) {
String sql = "SELECT count(1) FROM " + tableName;
Connection connection = DatabaseConnection.getConnection(databaseId);
if (connection == null) {
throw new MayException(CodeMsg.Request_ERROR, "数据库未加载到连接池" + databaseId);
}
Statement statement = null;
ResultSet resultSet = null;
try {
statement = connection.createStatement();
resultSet = statement.executeQuery(sql);
long total = 0;
while (resultSet.next()) {
total = resultSet.getLong(1);
}
return total;
} catch (SQLException e) {
log.error("查询表信息失败：{},数据库id:{},异常信息：{}", sql, databaseId, e.getMessage());
throw new MayException(CodeMsg.SERVER_ERROR, "数据库查询表信息失败" + tableName);
} finally {
try {
if (statement != null) {
statement.close();
}
if (resultSet != null) {
resultSet.close();
}
connection.close();
} catch (SQLException e) {
log.error("关闭数据库连接异常：{}", e.getMessage());
}
}
}
/**
* 根据sql查询表数据
*
* @param sql
* @return
* @throws SQLException
*/
public static List<Entity> query(String database, String sql) throws SQLException {
Connection connection = DatabaseConnection.getConnection(database);
if (connection == null) {
throw new MayException(CodeMsg.Request_ERROR, "数据库未加载到连接池" + database);
}
try {
return SqlExecutor.query(connection, sql, new EntityListHandler());
} catch (SQLException e) {
log.error("sql执行失败：{}", sql);
throw new MayException(CodeMsg.Request_ERROR, "sql执行失败");
} finally {
try {
connection.close();
} catch (SQLException e) {
log.error("关闭数据库连接异常：{}", e.getMessage());
}
}
}
/**
* 生成分页查询sql
*
* @param userDbId 用户数据库id
* @param tableName 表名
* @param pageSize 每页行数
* @param pageNumber 当前页数
* @param fields 1查询字段 as xxx,2排序字段，3原字段
* @return
*/
public static String generateSelectSql(String userDbId, String tableName, int pageSize, int pageNumber, String... fields) {
StringBuilder sb = new StringBuilder(fields[0]);
sb.append(" FROM ").append(tableName);
DatabaseConfig databaseConfig = DatabaseConnection.getUserDbInfo(userDbId);
int offset = (pageNumber - 1) * pageSize;
if (databaseConfig.getDatabaseType().equals(DatabaseType.POSTGRESQL)) {
sb.insert(0, "SELECT ");
sb.append(" ORDER BY ").append(fields[1]);
sb.append(" LIMIT ").append(pageSize).append(" OFFSET ").append(offset);
} else if (databaseConfig.getDatabaseType().equals(DatabaseType.SQLSERVER)) {
if (fields.length > 2) {
sb.insert(0, "SELECT " + fields[2] + " FROM (SELECT ROW_NUMBER() OVER (ORDER BY " + fields[1] + ") AS RowNumber,")
.append(") AS t WHERE t.RowNumber BETWEEN " + (offset + 1) + " AND " + (offset + pageSize));
} else {
sb.insert(0, "SELECT * FROM (SELECT ROW_NUMBER() OVER (ORDER BY " + fields[1] + ") AS RowNumber,")
.append(") AS t WHERE t.RowNumber BETWEEN " + (offset + 1) + " AND " + (offset + pageSize));
}
}
return sb.toString();
}
/**
* 查询表字段
*
* @return
*/
public static List<String> getTableField(String userDbId, String table) {
Connection connection = DatabaseConnection.getConnection(userDbId);
if (connection == null) {
log.error("数据库id未加载到连接池：{}", userDbId);
throw new MayException(CodeMsg.SQL_ERROR, "数据库id未加载到连接池:" + userDbId);
}
DatabaseConfig databaseConfig = DatabaseConnection.getUserDbInfo(userDbId);
Statement statement = null;
ResultSet resultSet = null;
try {
String sql = "";
if (databaseConfig.getDatabaseType().equals(DatabaseType.SQLSERVER)) {
sql = CommonConstant.SQL_SQLSERVER_FIELD.replace("${tableName}", table);
} else if (databaseConfig.getDatabaseType().equals(DatabaseType.POSTGRESQL)) {
sql = CommonConstant.SQL_POSTGRESQL_FIELD.replace("${tableName}", table);
}
statement = connection.createStatement();
resultSet = statement.executeQuery(sql);
List<String> rowdata = new ArrayList<>();
while (resultSet.next()) {
rowdata.add(resultSet.getString(1));
}
return rowdata;
} catch (SQLException e) {
log.error("查询表信息失败：{}", userDbId);
throw new MayException(CodeMsg.SERVER_ERROR, "mssql数据库查询表信息失败");
} finally {
try {
statement.close();
resultSet.close();
connection.close();
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
}
/**
* 查询数据库下所有表
*
* @return
*/
public static List<KeyValOfStrVo> getTables(String userDbId) {
DatabaseConfig config = DatabaseConnection.getUserDbInfo(userDbId);
Connection connection = DatabaseConnection.getConnection(config);
List<KeyValOfStrVo> names = new ArrayList<>();
Statement statement = null;
ResultSet resultSet = null;
try {
statement = connection.createStatement();
if (config.getDatabaseType().equals(DatabaseType.SQLSERVER)) {
resultSet = statement.executeQuery(CommonConstant.SQL_SQLSERVER_TABLE);
} else if (config.getDatabaseType().equals(DatabaseType.POSTGRESQL)) {
resultSet = statement.executeQuery(CommonConstant.SQL_POSTGRESQL_TABLE);
}
while (resultSet.next()) {
String tableName = resultSet.getString(2);
if (StrUtil.isBlank(tableName)) {
names.add(new KeyValOfStrVo(resultSet.getString(1), resultSet.getString(1)));
} else {
names.add(new KeyValOfStrVo(resultSet.getString(1), resultSet.getString(2)));
}
}
} catch (SQLException e) {
log.error("数据库连接失败");
throw new MayException(CodeMsg.SERVER_ERROR, "mssql数据库连接失败");
} finally {
try {
statement.close();
resultSet.close();
connection.close();
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
return names;
}
/**
* 查询数据库列表
*
* @return
*/
public static List<String> getDatabaseList(ExternalDataBaseDO dataBase) {
DatabaseConfig databaseConfig = new DatabaseConfig(dataBase);
List<String> names = new ArrayList<>();
Connection connection = DatabaseConnection.getConnection(databaseConfig);
Statement statement = null;
ResultSet resultSet = null;
try {
statement = connection.createStatement();
if (databaseConfig.getDatabaseType().equals(DatabaseType.POSTGRESQL)) {
resultSet = statement.executeQuery(CommonConstant.SQL_POSTGRESQL_DBNAME);
} else {
resultSet = statement.executeQuery(CommonConstant.SQL_SQLSERVER_DBNAME);
}
while (resultSet.next()) {
//从第一列获取库名
names.add(resultSet.getString(1));
}
} catch (SQLException e) {
log.error("数据库连接失败", e);
throw new MayException(CodeMsg.SERVER_ERROR, "数据库连接失败:" + e.getMessage());
}
return names;
}
}
项目启动时从持久化数据库内读取已添加的数据库
@Component
@Log4j2
public class StartUpHook implements ApplicationListener<ContextRefreshedEvent> {
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
log.info("从mongodb读取并初始化外部数据源");
List<ExternalDataBaseDO> dataList = 从业务数据库查询出所有用户数据库表;
if (CollectionUtil.isEmpty(dataList)) {
return;
}
for (ExternalDataBaseDO mydatasource : dataList) {
boolean res = DatabaseConnection.addConnection(mydatasource);
if (!res) {
log.error("数据库连接失败：{}：{}-{}", mydatasource.getHost(), mydatasource.getPort(), mydatasource.getDataBaseName());
}
}
}
}
项目关闭时释放所有连接
@Component
public class ShutdownHook implements ApplicationListener<ContextClosedEvent> {
@Override
public void onApplicationEvent(ContextClosedEvent event) {
System.out.println("已收到停机请求，请耐心等待线程处理完工作");
//循环遍历所有连接池，释放所有连接
for (ConnectionPool connectionPool : DatabaseConnection.getAllConnectionPools()) {
try {
connectionPool.getDataSource().close();
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
}
}