liuyupeng/saas-houduan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

交换机数据采集唯一标识改为clientId。

Browse Source 
优化收益计算
This commit is contained in:
gaoyutao
2026-01-05 18:32:49 +08:00
parent 2593c58741
commit a2659d0c6b
4 changed files with 106 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java
+46 -32
View File
@@ -635,7 +635,8 @@ public class MessageHandler {
"systemSwArchCollect",
"systemSwOsCollect",
"systemUnameCollect",
"systemUptimeCollect"
"systemUptimeCollect",
"procNumCollect"
);
// 如果collectType在特殊列表中，先删除上个时间段的记录
@@ -1132,21 +1133,36 @@ public class MessageHandler {
}
boolean isSingleInterface = networkInfoList.size() == 1;
boolean isMultiInterface = networkInfoList.size() > 1;
// 查询数据库中当前的网卡数量
RmNetworkInterface countQuery = new RmNetworkInterface();
countQuery.setClientId(clientId);
countQuery.setNewFlag(1);
List<RmNetworkInterface> currentInterfaces = rmNetworkInterfaceService.selectRmNetworkInterfaceList(countQuery);
int currentInterfaceCount = currentInterfaces.size();
// 构建数据库中当前网卡MAC地址的集合
Set<String> currentMacSet = currentInterfaces.stream()
.map(RmNetworkInterface::getMacAddress)
.collect(Collectors.toSet());
// 构建新数据中网卡MAC地址的集合
Set<String> newMacSet = networkInfoList.stream()
.map(NetworkInfo::getMac)
.filter(Objects::nonNull)
.collect(Collectors.toSet());
// 找出需要删除的网卡（数据库中有但新数据中没有的）
Set<String> macsToDelete = new HashSet<>(currentMacSet);
macsToDelete.removeAll(newMacSet);
int currentInterfaceCount = currentInterfaces.size();
// 判断网卡数量是否发生变化
boolean interfaceCountChanged = false;
if (!isRegister && currentInterfaceCount != networkInfoList.size()) {
interfaceCountChanged = true;
}
// 执行删除操作
if (!macsToDelete.isEmpty() && !isRegister) {
deleteNetworkInterfaces(clientId, macsToDelete);
}
for (NetworkInfo networkInfo : networkInfoList) {
List<NetworkInfo> childList = networkInfo.getSubInterfaces();
// 查询该网卡信息是否存在
@@ -1179,7 +1195,7 @@ public class MessageHandler {
}
}
// 如果网卡数量有变动，需要更新网卡绑定状态
if (!isRegister && interfaceCountChanged) {
if (!isRegister && interfaceCountChanged && !isSingleInterface) {
RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote();
updateParam.setClientId(clientId);
updateParam.setMultiPublicIpStatus("0");
@@ -1188,7 +1204,7 @@ public class MessageHandler {
} else {
// 更新网卡信息
RmNetworkInterface oldInterfaceMsg = exits.get(0);
cleanChildOldRecords(clientId, networkInfo.getMac());
// cleanChildOldRecords(clientId, networkInfo.getMac());
// 判断是否需要创建新记录
boolean needCreateNew = (!StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())
|| !StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway())) && StringUtils.equals(networkInfo.getMac(), oldInterfaceMsg.getMacAddress());
@@ -1200,35 +1216,33 @@ public class MessageHandler {
rmNetworkInterfaceChildService.insertRmNetworkInterfaceChild(insertChild);
}
}
if(needCreateNew) {
// 清理旧的历史数据
cleanOldRecords(clientId, networkInfo.getMac());
// 将已存在的记录标记为旧数据
RmNetworkInterface oldData = new RmNetworkInterface();
oldData.setNewFlag(0);
oldData.setClientId(clientId);
oldData.setMacAddress(oldInterfaceMsg.getMacAddress());
rmNetworkInterfaceService.updateRmNetworkInterfaceByMac(oldData);
// 创建新记录
RmNetworkInterface insertData = new RmNetworkInterface();
setNetworkInterfaceData(insertData, networkInfo, clientId);
rmNetworkInterfaceService.insertRmNetworkInterface(insertData);
// 如果网卡名称有变化，需要更新网卡绑定状态
if(!isRegister && !StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())){
RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote();
updateParam.setClientId(clientId);
updateParam.setMultiPublicIpStatus("0");
remoteRevenueConfigService.updateStatusByResource(updateParam, SecurityConstants.INNER);
}
} else {
// 更新现有记录
updateNetworkInterface(networkInfo, clientId, oldInterfaceMsg, isSingleInterface, isRegister);
}
// 更新现有记录
updateNetworkInterface(networkInfo, clientId, oldInterfaceMsg, isSingleInterface, isRegister);
}
}
}
/**
* 删除过时的网卡信息
* @param clientId 客户端ID
* @param macsToDelete 需要删除的MAC地址集合
*/
private void deleteNetworkInterfaces(String clientId, Set<String> macsToDelete) {
for (String macAddress : macsToDelete) {
RmNetworkInterface query = new RmNetworkInterface();
query.setClientId(clientId);
query.setMacAddress(macAddress);
query.setNewFlag(1);
List<RmNetworkInterface> oldExits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(query);
if(!oldExits.isEmpty()) {
oldExits.forEach(oldMsg -> {
rmNetworkInterfaceService.deleteRmNetworkInterfaceById(oldMsg.getId());
});
}
}
// 记录删除操作日志
log.info("删除客户端 {} 的过时网卡信息，MAC地址: {}", clientId, macsToDelete);
}
/**
* 设置网卡信息公共字段
tongran-rocketmq/src/main/java/com/tongran/rocketmq/snmp/scheduler/MultiSwitchCollectionScheduler.java
+48 -133
View File
@@ -43,6 +43,7 @@ public class MultiSwitchCollectionScheduler {
private final Map<String, Map<String, ScheduledFuture<?>>> allSwitchTasks = new ConcurrentHashMap<>();
private final Map<String, SwitchOidDto> allSwitchConfigs = new ConcurrentHashMap<>();
private final Map<String, List<CollectVo>> allSwitchCollectConfigs = new ConcurrentHashMap<>();
private final Map<String, ScheduledFuture<?>> heartbeatTasks = new ConcurrentHashMap<>();
// 使用构造器注入而不是@Autowired
public MultiSwitchCollectionScheduler(TaskScheduler taskScheduler,
@@ -88,11 +89,11 @@ public class MultiSwitchCollectionScheduler {
List<CollectVo> collectVos, SwitchOidVo switchOidVo) {
// 确保采集器已初始化
if (!ensureCollectorInitialized()) {
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", switchIp);
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", clientId);
return;
}
// 停止该交换机的现有任务
stopSwitchCollection(switchIp);
stopSwitchCollection(clientId);
// 创建交换机配置
SwitchOidDto switchVo = createSwitchConfig(switchIp, community, port, collectVos, switchOidVo);
@@ -132,12 +133,12 @@ public class MultiSwitchCollectionScheduler {
}
// 存储配置
allSwitchTasks.put(switchIp, switchTasks);
allSwitchConfigs.put(switchIp, switchVo);
allSwitchCollectConfigs.put(switchIp, new ArrayList<>(collectVos));
allSwitchTasks.put(clientId, switchTasks);
allSwitchConfigs.put(clientId, switchVo);
allSwitchCollectConfigs.put(clientId, new ArrayList<>(collectVos));
log.debug("启动交换机 {} 的采集任务，包含 {} 个采集类型{}",
switchIp, switchTasks.size());
clientId, switchTasks.size());
}
/**
@@ -211,7 +212,7 @@ public class MultiSwitchCollectionScheduler {
int securityLevel, List<CollectVo> collectVos,
SwitchOidVo switchOidVo) {
// 停止该交换机的现有任务
stopSwitchCollection(switchIp);
stopSwitchCollection(clientId);
// 创建SNMP v3交换机配置
SwitchOidDto switchVo = createV3SwitchConfig(switchIp, port, securityName,
@@ -232,12 +233,12 @@ public class MultiSwitchCollectionScheduler {
}
// 存储配置
allSwitchTasks.put(switchIp, switchTasks);
allSwitchConfigs.put(switchIp, switchVo);
allSwitchCollectConfigs.put(switchIp, new ArrayList<>(collectVos));
allSwitchTasks.put(clientId, switchTasks);
allSwitchConfigs.put(clientId, switchVo);
allSwitchCollectConfigs.put(clientId, new ArrayList<>(collectVos));
log.debug("启动SNMP v3交换机 {} 的采集任务，包含 {} 个采集类型{}",
switchIp, switchTasks.size());
clientId, switchTasks.size());
}
/**
@@ -262,7 +263,7 @@ public class MultiSwitchCollectionScheduler {
MessageProducer messageProducer = new MessageProducer();
// 确保采集器已初始化
if (!ensureCollectorInitialized()) {
log.error("SNMP采集器未初始化，无法启动交换机 {} 的心跳检测任务", switchIp);
log.error("SNMP采集器未初始化，无法启动交换机 {} 的心跳检测任务", clientId);
return null;
}
Runnable heartbeatTask = () -> {
@@ -270,7 +271,7 @@ public class MultiSwitchCollectionScheduler {
// 获取当前时间作为检测时间戳
long checkTimestamp = System.currentTimeMillis();
log.debug("[心跳检测] 开始检测交换机 {} 的心跳状态, 时间: {}", switchIp, checkTimestamp);
log.debug("[心跳检测] 开始检测交换机 {} 的心跳状态, 时间: {}", clientId, checkTimestamp);
// 使用系统名称OID (1.3.6.1.2.1.1.5.0) 进行心跳检测
// 创建专门用于心跳检测的OID配置
@@ -305,13 +306,13 @@ public class MultiSwitchCollectionScheduler {
"heartbeat",
JSONObject.toJSONString(message)
);
log.debug("[心跳检测] 完成检测交换机 {} 的心跳状态, 响应时间: {}ms", switchIp, responseTime);
log.debug("[心跳检测] 完成检测交换机 {} 的心跳状态, 响应时间: {}ms", clientId, responseTime);
}else{
log.error("[心跳检测] 检测交换机 {} 心跳失败: {}", switchIp, result.getErrorMessage());
log.error("[心跳检测] 检测交换机 {} 心跳失败: {}", clientId, result.getErrorMessage());
}
} catch (Exception e) {
log.error("[心跳检测] 检测交换机 {} 心跳失败: {}", switchIp, e.getMessage());
log.error("[心跳检测] 检测交换机 {} 心跳失败: {}", clientId, e.getMessage());
}
};
@@ -320,8 +321,9 @@ public class MultiSwitchCollectionScheduler {
Duration interval = Duration.ofSeconds(intervalSeconds);
ScheduledFuture<?> future = taskScheduler.scheduleWithFixedDelay(
heartbeatTask, Instant.now().plus(initialDelay), interval);
log.debug("交换机 {} 心跳检测任务启动，检测间隔: 30秒", switchIp);
// 存储心跳任务
heartbeatTasks.put(clientId, future);
log.debug("交换机 {} 心跳检测任务启动，检测间隔: 30秒", clientId);
return future;
}
@@ -338,7 +340,7 @@ public class MultiSwitchCollectionScheduler {
private ScheduledFuture<?> startTrafficCollection(String switchIp, String clientId, SwitchOidDto switchVo, CollectVo collectVo) {
// 确保采集器已初始化
if (!ensureCollectorInitialized()) {
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", switchIp);
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", clientId);
return null;
}
// 创建Cron表达式：每5分钟的第0秒执行（0,5,10,15...）
@@ -349,7 +351,7 @@ public class MultiSwitchCollectionScheduler {
// 获取当前时间作为采集时间戳（对齐到5分钟整点）
long collectionTimestamp = getAlignedTimestamp();
log.debug("[流量采集] 开始采集交换机 {} 的网络数据, 时间: {}", switchIp, collectionTimestamp);
log.debug("[流量采集] 开始采集交换机 {} 的网络数据, 时间: {}", clientId, collectionTimestamp);
CollectionResult result = dynamicOidCollector.getInterfaceInfoByType(switchVo, "net");
@@ -359,17 +361,17 @@ public class MultiSwitchCollectionScheduler {
// 这里可以处理采集结果，比如保存到数据库或发送到消息队列
processCollectionResult(result, clientId);
log.debug("[流量采集] 完成采集交换机 {} 的网络数据, 时间戳: {}", switchIp, collectionTimestamp);
log.debug("[流量采集] 完成采集交换机 {} 的网络数据, 时间戳: {}", clientId, collectionTimestamp);
} catch (Exception e) {
log.error("[流量采集] 采集交换机 {} 数据失败: {}", switchIp, e.getMessage());
log.error("[流量采集] 采集交换机 {} 数据失败: {}", clientId, e.getMessage());
}
};
CronTrigger trigger = new CronTrigger(cronExpression);
ScheduledFuture<?> future = taskScheduler.schedule(trafficTask, trigger);
log.debug("交换机 {} 流量采集任务启动，时间点: 每5分钟整点", switchIp);
log.debug("交换机 {} 流量采集任务启动，时间点: 每5分钟整点", clientId);
return future;
}
/**
@@ -409,7 +411,7 @@ public class MultiSwitchCollectionScheduler {
private ScheduledFuture<?> startRegularCollection(String switchIp, String clientId, SwitchOidDto switchVo, CollectVo collectVo) {
// 确保采集器已初始化
if (!ensureCollectorInitialized()) {
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", switchIp);
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", clientId);
return null;
}
long intervalSeconds = collectVo.getInterval() != null ? collectVo.getInterval() : 300L;
@@ -419,7 +421,7 @@ public class MultiSwitchCollectionScheduler {
Runnable regularTask = () -> {
try {
log.debug("[{}] 开始采集交换机 {} 的数据, 时间: {}",
collectVo.getType(), switchIp, LocalDateTime.now());
collectVo.getType(), clientId, LocalDateTime.now());
// 获取对齐到整分钟的时间戳
long collectionTimestamp = getAlignedMinuteTimestamp();
// 根据类型执行对应的采集
@@ -429,11 +431,11 @@ public class MultiSwitchCollectionScheduler {
// 处理采集结果
processCollectionResult(result, clientId);
log.debug("[{}] 完成采集交换机 {} 的数据", collectVo.getType(), switchIp);
log.debug("[{}] 完成采集交换机 {} 的数据", collectVo.getType(), clientId);
} catch (Exception e) {
log.error("[{}] 采集交换机 {} 数据失败: {}",
collectVo.getType(), switchIp, e.getMessage());
collectVo.getType(), clientId, e.getMessage());
}
};
@@ -441,7 +443,7 @@ public class MultiSwitchCollectionScheduler {
regularTask, startTime, interval);
log.debug("交换机 {} [{}] 采集任务启动，间隔: {}秒, 开始时间: {}",
switchIp, collectVo.getType(), intervalSeconds, startTime);
clientId, collectVo.getType(), intervalSeconds, startTime);
return future;
}
/**
@@ -615,14 +617,19 @@ public class MultiSwitchCollectionScheduler {
/**
* 停止单个交换机的所有采集任务
*/
public void stopSwitchCollection(String switchIp) {
Map<String, ScheduledFuture<?>> tasks = allSwitchTasks.remove(switchIp);
public void stopSwitchCollection(String clientId) {
Map<String, ScheduledFuture<?>> tasks = allSwitchTasks.remove(clientId);
if (tasks != null) {
tasks.values().forEach(future -> future.cancel(false));
log.debug("停止交换机 {} 的所有采集任务{}", switchIp);
log.debug("停止交换机 {} 的所有采集任务{}", clientId);
}
allSwitchConfigs.remove(switchIp);
allSwitchCollectConfigs.remove(switchIp);
ScheduledFuture<?> heartbeatTask = heartbeatTasks.remove(clientId);
if (heartbeatTask != null && !heartbeatTask.isCancelled()) {
heartbeatTask.cancel(false);
log.debug("停止交换机 {} 的心跳检测任务", clientId);
}
allSwitchConfigs.remove(clientId);
allSwitchCollectConfigs.remove(clientId);
}
/**
@@ -632,110 +639,18 @@ public class MultiSwitchCollectionScheduler {
allSwitchTasks.values().forEach(tasks ->
tasks.values().forEach(future -> future.cancel(false)));
allSwitchTasks.clear();
// 停止所有心跳任务
heartbeatTasks.values().forEach(future -> {
if (future != null && !future.isCancelled()) {
future.cancel(false);
}
});
heartbeatTasks.clear();
allSwitchConfigs.clear();
allSwitchCollectConfigs.clear();
System.out.println("停止所有交换机的采集任务");
}
/**
* 获取交换机的任务状态
*/
public Map<String, Object> getSwitchStatus(String switchIp) {
Map<String, Object> status = new HashMap<>();
status.put("switchIp", switchIp);
Map<String, ScheduledFuture<?>> tasks = allSwitchTasks.get(switchIp);
List<CollectVo> collectConfigs = allSwitchCollectConfigs.get(switchIp);
if (tasks != null && collectConfigs != null) {
status.put("isRunning", true);
status.put("taskCount", tasks.size());
// 获取交换机配置信息
SwitchOidDto switchConfig = allSwitchConfigs.get(switchIp);
if (switchConfig != null) {
status.put("snmpVersion", switchConfig.getVersion() == SnmpConstants.version3 ? "v3" : "v2c");
status.put("securityName", switchConfig.getSecurityName());
status.put("securityLevel", getSecurityLevelDescription(switchConfig.getSecurityLevel()));
}
List<Map<String, Object>> taskStatusList = new ArrayList<>();
for (CollectVo collectVo : collectConfigs) {
Map<String, Object> taskStatus = new HashMap<>();
taskStatus.put("type", collectVo.getType());
taskStatus.put("collect", collectVo.isCollect());
taskStatus.put("interval", collectVo.getInterval());
ScheduledFuture<?> task = tasks.get(collectVo.getType());
taskStatus.put("isActive", task != null && !task.isCancelled());
taskStatusList.add(taskStatus);
}
status.put("tasks", taskStatusList);
} else {
status.put("isRunning", false);
}
return status;
}
/**
* 获取安全级别描述
*/
private String getSecurityLevelDescription(int securityLevel) {
switch (securityLevel) {
case SecurityLevel.NOAUTH_NOPRIV:
return "noAuthNoPriv";
case SecurityLevel.AUTH_NOPRIV:
return "authNoPriv";
case SecurityLevel.AUTH_PRIV:
return "authPriv";
default:
return "unknown";
}
}
/**
* 获取所有交换机的状态
*/
public Map<String, Map<String, Object>> getAllSwitchesStatus() {
Map<String, Map<String, Object>> allStatus = new HashMap<>();
for (String switchIp : allSwitchTasks.keySet()) {
allStatus.put(switchIp, getSwitchStatus(switchIp));
}
return allStatus;
}
/**
* 更新交换机的采集配置
*/
public void updateSwitchConfig(String switchIp, String community, int port, String clientId,
List<CollectVo> collectVos, SwitchOidVo switchOidVo) {
// 先停止现有任务
stopSwitchCollection(switchIp);
// 重新启动任务
startSwitchCollection(switchIp, community, port, clientId, collectVos, switchOidVo);
log.info("更新交换机 {} 的采集配置{}", switchIp);
}
/**
* 更新SNMP v3交换机的采集配置
*/
public void updateV3SwitchConfig(String switchIp, int port, String clientId, String securityName,
String authProtocol, String authPassword,
String privProtocol, String privPassword,
int securityLevel, List<CollectVo> collectVos,
SwitchOidVo switchOidVo) {
// 先停止现有任务
stopSwitchCollection(switchIp);
// 重新启动任务
startV3SwitchCollection(switchIp, port, clientId, securityName, authProtocol, authPassword,
privProtocol, privPassword, securityLevel, collectVos, switchOidVo);
log.debug("更新SNMP v3交换机 {} 的采集配置{}", switchIp);
}
/**
* 批量处理设备配置（支持v2c和v3混合）
*/
@@ -769,7 +684,7 @@ public class MultiSwitchCollectionScheduler {
);
}
} catch (Exception e) {
log.error("处理交换机配置失败，switchIp: {}", device.getSwitchIp(), e);
log.error("处理交换机配置失败，clientId: {}", device.getClientId(), e);
}
}
}