fangqiao/qiaoba-boot
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

add

Browse Source
This commit is contained in:
ailanyin
2023-06-26 17:07:45 +08:00
parent caeae9127f
commit ae12b7f2ff
22 changed files with 656 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
qiaoba-commons/qiaoba-common-datasource/src/main/java/com/qiaoba/common/database/config/DynamicDataSourceConfig.java
View File

@ -9,6 +9,7 @@ import com.qiaoba.common.base.constants.TenantConstant;
import com.qiaoba.common.base.context.BaseContext;
import com.qiaoba.common.base.enums.BaseEnum;
import com.qiaoba.common.database.entity.DynamicDataSource;
import com.qiaoba.common.database.monitor.NotOnlineDatasourceMonitor;
import com.qiaoba.common.database.properties.DataSourceProperties;
import com.qiaoba.common.database.service.DynamicDatasourceService;
import com.qiaoba.common.database.utils.JdbcUtil;
@ -89,13 +90,28 @@ public class DynamicDataSourceConfig {
private void initDefault() {
List<DynamicDataSource> dataSources = dataSourceProperties.getDataSources();
// 0索引作为主数据源
addPrimaryMap(TenantConstant.DEFAULT_TENANT_ID, buildDataSource(TenantConstant.DEFAULT_TENANT_ID, dataSources.get(0)));
dataSources.remove(0);
// 非0索引的备用
for (int i = 0; i < dataSources.size(); i++) {
// 0索引作为主数据源
Object dataSource = buildDataSource(TenantConstant.DEFAULT_TENANT_ID, dataSources.get(i));
if (Objects.isNull(dataSource)) {
// 默认的主数据源挂了
// 加入到错误数据源Map 等待重试
NotOnlineDatasourceMonitor.addErrorDatasource(TenantConstant.DEFAULT_TENANT_ID, dataSources.get(i));
dataSources.remove(i);
} else {
addPrimaryMap(TenantConstant.DEFAULT_TENANT_ID, dataSource);
dataSources.remove(i);
break;
}
}
if (CollUtil.isEmpty(PRIMARY_DATASOURCE_MAP)) {
log.error("主系统配置数据源全部无效, 请检查 yml 中相关配置");
}
// 其他数据源备用
addBackupMap(TenantConstant.DEFAULT_TENANT_ID, dataSources);
// 刷新数据源
dataSource.freshDataSource(PRIMARY_DATASOURCE_MAP);
this.dataSource.freshDataSource(PRIMARY_DATASOURCE_MAP);
}
private void initTenant() {
@ -106,10 +122,19 @@ public class DynamicDataSourceConfig {
for (int i = 0; i < dataSources.size(); i++) {
DynamicDataSource dynamicDataSource = dataSources.get(i);
if (BaseEnum.YES.getCode().equals(dynamicDataSource.getIsPrimary())) {
addPrimaryMap(tenantId, buildDataSource(dataSources.get(i).getTenantId(), dataSources.get(i)));
// 去除主要数据源，剩下皆为备用数据源
dataSources.remove(i);
break;
Object dataSource = buildDataSource(dataSources.get(i).getTenantId(), dataSources.get(i));
if (Objects.isNull(dataSource)) {
// 默认的主数据源挂了
// 加入到错误数据源Map 等待重试
NotOnlineDatasourceMonitor.addErrorDatasource(tenantId, dataSources.get(i));
// 在数据源集合中删除, 防止将错误的数据源加载到备用数据源中
dataSources.remove(i);
} else {
addPrimaryMap(tenantId, dataSource);
// 去除主要数据源，剩下皆为备用数据源
dataSources.remove(i);
break;
}
}
}
// 备用数据源
@ -129,6 +154,17 @@ public class DynamicDataSourceConfig {
public void changePrimaryDatasource(String tenantId, Object datasource) {
PRIMARY_DATASOURCE_MAP.put(tenantId, datasource);
// 将数据源的类型保存
DruidPooledConnection connection = null;
try {
connection = ((DruidDataSource) datasource).getConnection();
TENANT_DATASOURCE_TYPE_MAP.put(tenantId, connection.getMetaData().getDatabaseProductName());
} catch (SQLException e) {
e.printStackTrace();
} finally {
// 归还 connection
IoUtil.close(connection);
}
dataSource.freshDataSource(PRIMARY_DATASOURCE_MAP);
}

117
qiaoba-commons/qiaoba-common-datasource/src/main/java/com/qiaoba/common/database/monitor/NotOnlineDatasourceMonitor.java Normal file
View File

@ -0,0 +1,117 @@
package com.qiaoba.common.database.monitor;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ListUtil;
import com.baomidou.lock.LockInfo;
import com.baomidou.lock.LockTemplate;
import com.qiaoba.common.database.config.DynamicDataSourceConfig;
import com.qiaoba.common.database.entity.DynamicDataSource;
import com.qiaoba.common.database.utils.JdbcUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
/**
* 不在线的数据源监控
* 尝试连接,如果连接成功,加入到备用数据源
*
* @author ailanyin
* @version 1.0
* @since 2023/6/26 10:46
*/
@Component
@Slf4j
public class NotOnlineDatasourceMonitor {
/**
* 错误数据源(连接中断)
* <p>
* key: 租户ID
* value: 错误数据源 list
*/
public static Map<String, List<DynamicDataSource>> ERROR_DATASOURCE_MAP = new ConcurrentHashMap<>();
private static final String LOCK_KEY = "lock4j:notOnlineDatasourceMonitor";
@Resource
private LockTemplate lockTemplate;
@PostConstruct
public void init() {
// 10s 运行一次
new Timer().schedule(new TimerTask() {
@Override
public void run() {
// 项目启动时加载数据源还未完成
if (!DynamicDataSourceConfig.COMPLETE_LOAD_DATASOURCE) {
return;
}
// 没有暂时失联的数据源, 直接结束
if (CollUtil.isEmpty(ERROR_DATASOURCE_MAP)) {
return;
}
// expire = -1 锁自动续期, 防止数据源过多或异常等待, 超过默认锁 30s
final LockInfo lockInfo = lockTemplate.lock(LOCK_KEY, -1, 1000);
//申请锁失败 说明集群中其他设备正在执行监控
if (null == lockInfo) {
return;
}
//申请锁成功
try {
log.trace("开始-[错误数据源重试]-线程, 时间: {}", new Date());
tryConnect();
log.trace("结束-[错误数据源重试]-线程, 时间: {}", new Date());
} finally {
// 释放锁
lockTemplate.releaseLock(lockInfo);
}
}
}, 0, 10 * 1000);
}
private void tryConnect() {
Set<String> tenantIds = ERROR_DATASOURCE_MAP.keySet();
for (String tenantId : tenantIds) {
List<DynamicDataSource> errorDatasourceList = ERROR_DATASOURCE_MAP.get(tenantId);
for (int i = 0; i < errorDatasourceList.size(); i++) {
DynamicDataSource errorDatasource = errorDatasourceList.get(i);
// 说明连接成功
boolean check = JdbcUtil.checkConnect(errorDatasource.getDriver(), errorDatasource.getUrl(), errorDatasource.getUsername(), errorDatasource.getPassword());
if (check) {
log.info("数据源重连成功, Url: {}", errorDatasource.getUrl());
// 从errorMap中删除
errorDatasourceList.remove(errorDatasource);
if (CollUtil.isEmpty(errorDatasourceList)) {
ERROR_DATASOURCE_MAP.remove(tenantId);
}
// 加入到备用Map中
addBackupMap(tenantId, errorDatasource);
}
}
}
}
private void addBackupMap(String tenantId, DynamicDataSource dataSource) {
List<DynamicDataSource> dataSourceList = DynamicDataSourceConfig.BACKUP_DATASOURCE_MAP.get(tenantId);
if (CollUtil.isEmpty(dataSourceList)) {
DynamicDataSourceConfig.BACKUP_DATASOURCE_MAP.put(tenantId, ListUtil.toList(dataSource));
} else {
dataSourceList.add(dataSource);
}
}
public static void addErrorDatasource(String tenantId, DynamicDataSource dataSource) {
List<DynamicDataSource> errorDataSourceList = NotOnlineDatasourceMonitor.ERROR_DATASOURCE_MAP.get(tenantId);
if (CollUtil.isEmpty(errorDataSourceList)) {
NotOnlineDatasourceMonitor.ERROR_DATASOURCE_MAP.put(tenantId, ListUtil.toList(dataSource));
} else {
errorDataSourceList.add(dataSource);
}
}
}

170
qiaoba-commons/qiaoba-common-datasource/src/main/java/com/qiaoba/common/database/monitor/DatasourceConnectionMonitor.java → qiaoba-commons/qiaoba-common-datasource/src/main/java/com/qiaoba/common/database/monitor/OnlineDatasourceMonitor.java
View File

@ -1,5 +1,6 @@
package com.qiaoba.common.database.monitor;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.io.IoUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.baomidou.lock.LockInfo;
@ -9,39 +10,40 @@ import com.qiaoba.common.base.enums.DataBaseEnum;
import com.qiaoba.common.database.config.DynamicDataSourceConfig;
import com.qiaoba.common.database.entity.DynamicDataSource;
import com.qiaoba.common.database.service.DynamicDatasourceService;
import com.qiaoba.common.database.utils.JdbcUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
/**
* 监控-数据源连接监控
* 在线的数据源监控
* 尝试连接, 如果连接不成功, 替换可用数据源, 将失败数据源加入到错误数据源, 等待重试
*
* @author ailanyin
* @version 1.0
* @since 2023/6/14 14:48
* @since 2023/6/26 10:46
*/
@Component
@Slf4j
public class DatasourceConnectionMonitor {
public class OnlineDatasourceMonitor {
private static final String LOCK_KEY = "lock4j:onlineDatasourceMonitor";
private static Map<String, String> WAIT_UPDATE_DATASOURCE_STATUS = new ConcurrentHashMap<>();
private static Map<String, List<String>> WAIT_ADD_ERROR_MAP = new ConcurrentHashMap<>();
@Resource
private LockTemplate lockTemplate;
@Resource
private DynamicDataSourceConfig dynamicDataSourceConfig;
@Resource
private DynamicDatasourceService dynamicDatasourceService;
@Resource
private LockTemplate lockTemplate;
private static Map<String, String> WAIT_UPDATE_DATASOURCE_STATUS = new ConcurrentHashMap<>();
private static final String LOCK_KEY = "lock4j:datasourceConnectionMonitor";
@PostConstruct
public void init() {
@ -49,7 +51,7 @@ public class DatasourceConnectionMonitor {
new Timer().schedule(new TimerTask() {
@Override
public void run() {
// 项目加载数据源还未完成
// 项目启动时加载数据源还未完成
if (!DynamicDataSourceConfig.COMPLETE_LOAD_DATASOURCE) {
return;
}
@ -62,8 +64,9 @@ public class DatasourceConnectionMonitor {
}
//申请锁成功
try {
// 执行监控
datasourceConnectionMonitor();
log.trace("开始运行数据源监控线程, 时间: {}", new Date());
tryConnect();
log.trace("结束运行数据源监控线程, 时间: {}", new Date());
} finally {
// 释放锁
lockTemplate.releaseLock(lockInfo);
@ -73,83 +76,62 @@ public class DatasourceConnectionMonitor {
}, 0, 1000);
}
/**
* 核心监控内容
*/
private void datasourceConnectionMonitor() {
log.trace("开始运行数据源监控线程, 时间: {}", new Date());
private void tryConnect() {
for (String tenantId : DynamicDataSourceConfig.TENANT_IDS) {
Object primary = DynamicDataSourceConfig.PRIMARY_DATASOURCE_MAP.get(tenantId);
DynamicDataSourceConfig.BACKUP_DATASOURCE_MAP.get(tenantId);
if (Objects.isNull(primary)) {
// 说明初始化主要数据源的时候出错
log.error("租户[{}]-目前主数据源异常, 开始切换备用数据源", tenantId);
// 切换备用数据源
changePrimary(tenantId);
} else {
DruidDataSource dataSource = (DruidDataSource) primary;
Connection connection = null;
try {
connection = dataSource.getConnection();
if (check(connection, tenantId)) {
// 说明数据源正常
log.trace("租户[{}]-目前主数据源正常, 无需切换数据源", tenantId);
// 主数据 处理任务
if (TenantConstant.DEFAULT_TENANT_ID.equals(tenantId)) {
handleJob();
}
continue;
}
// 切换备用数据源为主数据源
backToPrimary(tenantId);
continue;
}
log.error("租户[{}]-目前主数据源异常, 开始切换备用数据源", tenantId);
// 主数据源异常 切换备用数据源
if (changePrimary(tenantId)) {
// 备用切换成功, 关闭原有异常数据源
IoUtil.close(dataSource);
DruidDataSource dataSource = (DruidDataSource) primary;
Connection connection = null;
try {
connection = dataSource.getConnection();
if (JdbcUtil.checkConnect(connection)) {
// 说明数据源正常
log.trace("租户[{}]-目前主数据源正常, 无需切换数据源", tenantId);
// 系统默认主数据源 处理任务
if (TenantConstant.DEFAULT_TENANT_ID.equals(tenantId)) {
handleUpdateDatasourceStatus();
handleErrorDatasource();
}
} catch (SQLException e) {
} finally {
IoUtil.close(connection);
continue;
}
log.error("租户[{}]-目前主数据源异常, 开始切换备用数据源", tenantId);
// 主数据源异常 切换备用数据源
if (!backToPrimary(tenantId)) {
// 备用切换失败
// 关闭原有异常数据源
IoUtil.close(dataSource);
// 在数据源Map中删除
DynamicDataSourceConfig.PRIMARY_DATASOURCE_MAP.remove(tenantId);
}
// 将原有异常数据源加入到错误数据源Map, 等待重试
addErrorDatasource(tenantId, dataSource);
} catch (SQLException e) {
} finally {
IoUtil.close(connection);
}
}
log.trace("结束运行数据源监控线程, 时间: {}", new Date());
}
/**
* 检查数据源连接可用性
*
* @param conn 数据源
* @param tenantId 租户ID
* @return 结果
*/
private Boolean check(Connection conn, String tenantId) {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
// 允许 2s 延时
stmt.setQueryTimeout(2);
rs = stmt.executeQuery(DataBaseEnum.getCheckSql(DynamicDataSourceConfig.TENANT_DATASOURCE_TYPE_MAP.get(tenantId)));
return true;
} catch (Exception e) {
return false;
} finally {
IoUtil.close(rs);
IoUtil.close(stmt);
}
}
/**
* 切换主数据源
* 切换备用数据源为主数据源
*
* @param tenantId 租户ID
*/
private Boolean changePrimary(String tenantId) {
private Boolean backToPrimary(String tenantId) {
// 备用数据源
List<DynamicDataSource> dataSources = DynamicDataSourceConfig.BACKUP_DATASOURCE_MAP.get(tenantId);
if (Objects.isNull(dataSources)) {
if (CollUtil.isEmpty(dataSources)) {
log.error("租户:[{}]切换备用数据源失败, 原因: 没有备用数据源", tenantId);
return false;
}
@ -184,7 +166,7 @@ public class DatasourceConnectionMonitor {
}
}
private void handleJob() {
private void handleUpdateDatasourceStatus() {
Set<String> keys = WAIT_UPDATE_DATASOURCE_STATUS.keySet();
for (String key : keys) {
try {
@ -195,8 +177,46 @@ public class DatasourceConnectionMonitor {
WAIT_UPDATE_DATASOURCE_STATUS.remove(key);
log.info("更新数据库中租户数据源状态完成, 租户ID: {}", key);
} catch (Exception e) {
log.error("更新数据库中租户数据源状态完成, 租户ID: {}, 失败原因: {}", key, e.getMessage());
log.error("更新数据库中租户数据源状态未完成, 租户ID: {}, 失败原因: {}", key, e.getMessage());
}
}
}
private void handleErrorDatasource() {
Set<String> tenantIds = WAIT_ADD_ERROR_MAP.keySet();
for (String tenantId : tenantIds) {
List<String> ipList = WAIT_ADD_ERROR_MAP.get(tenantId);
for (String ip : ipList) {
NotOnlineDatasourceMonitor.addErrorDatasource(tenantId, dynamicDatasourceService.selectByIp(tenantId, ip));
}
WAIT_ADD_ERROR_MAP.remove(tenantId);
}
}
private void addErrorDatasource(String tenantId, DruidDataSource dataSource) {
// 主系统
if (TenantConstant.DEFAULT_TENANT_ID.equals(tenantId)) {
DynamicDataSource dynamicDataSource = new DynamicDataSource();
dynamicDataSource.setPassword(dataSource.getPassword());
dynamicDataSource.setUsername(dataSource.getUsername());
dynamicDataSource.setUrl(dataSource.getUrl());
dynamicDataSource.setDriver(dataSource.getDriverClassName());
dynamicDataSource.setTenantId(tenantId);
dynamicDataSource.setInitialSize(dataSource.getInitialSize());
dynamicDataSource.setMaxActive(dataSource.getMaxActive());
dynamicDataSource.setMinIdle(dataSource.getMinIdle());
NotOnlineDatasourceMonitor.addErrorDatasource(tenantId, dynamicDataSource);
return;
}
// 普通租户
String ip = DataBaseEnum.getIp(dataSource.getUrl(), dataSource.getDriverClassName());
List<String> errorIpList = WAIT_ADD_ERROR_MAP.get(tenantId);
if (CollUtil.isEmpty(errorIpList)) {
WAIT_ADD_ERROR_MAP.put(tenantId, CollUtil.toList(ip));
} else {
errorIpList.add(ip);
}
}
}

9
qiaoba-commons/qiaoba-common-datasource/src/main/java/com/qiaoba/common/database/service/DynamicDatasourceService.java
View File

@ -28,4 +28,13 @@ public interface DynamicDatasourceService {
* @param datasourceId 数据源ID
*/
void changePrimaryDatasource(String tenantId, String datasourceId);
/**
* 通过IP查询
*
* @param tenantId tenantId
* @param ip ip
* @return obj
*/
DynamicDataSource selectByIp(String tenantId, String ip);
}

27
qiaoba-commons/qiaoba-common-datasource/src/main/java/com/qiaoba/common/database/utils/JdbcUtil.java
View File

@ -2,11 +2,15 @@ package com.qiaoba.common.database.utils;
import cn.hutool.core.io.IoUtil;
import cn.hutool.core.util.StrUtil;
import com.qiaoba.common.base.enums.DataBaseEnum;
import com.qiaoba.common.base.exceptions.ServiceException;
import com.qiaoba.common.database.config.DynamicDataSourceConfig;
import lombok.extern.slf4j.Slf4j;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
/**
* JdbcUtil
@ -42,6 +46,29 @@ public class JdbcUtil {
}
}
/**
* 检查数据源连接可用性
*
* @param conn 数据源
* @return 结果
*/
public static boolean checkConnect(Connection conn) {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
// 允许 2s 延时
stmt.setQueryTimeout(2);
rs = stmt.executeQuery(DataBaseEnum.getCheckSql(conn.getMetaData().getDatabaseProductName()));
return true;
} catch (Exception e) {
return false;
} finally {
IoUtil.close(rs);
IoUtil.close(stmt);
}
}
public static Connection getConnection(String driver, String url, String username, String password) {
Connection conn = null;
try {

3
qiaoba-commons/qiaoba-common-datasource/src/main/resources/META-INF/spring.factories
View File

@ -1,6 +1,7 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.qiaoba.common.database.factories.DynamicDataSourceFactory,\
com.qiaoba.common.database.monitor.DatasourceConnectionMonitor,\
com.qiaoba.common.database.monitor.OnlineDatasourceMonitor,\
com.qiaoba.common.database.monitor.NotOnlineDatasourceMonitor,\
com.qiaoba.common.database.handlers.schema.SchemaHandlerFactory,\
com.qiaoba.common.database.handlers.schema.MysqlSchemaHandler,\
com.qiaoba.common.database.config.MybatisPlusConfig