From a2659d0c6b14efeb2bd53cac0801378d6af9bf58 Mon Sep 17 00:00:00 2001 From: gaoyutao Date: Mon, 5 Jan 2026 18:32:49 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BA=A4=E6=8D=A2=E6=9C=BA=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E9=87=87=E9=9B=86=E5=94=AF=E4=B8=80=E6=A0=87=E8=AF=86=E6=94=B9?= =?UTF-8?q?=E4=B8=BAclientId=E3=80=82=20=E4=BC=98=E5=8C=96=E6=94=B6?= =?UTF-8?q?=E7=9B=8A=E8=AE=A1=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EpsInitialTrafficDataServiceImpl.java | 17 +- .../rocketmq/handler/MessageHandler.java | 78 ++++---- .../MultiSwitchCollectionScheduler.java | 181 +++++------------- .../InitialBandwidthTrafficMapper.xml | 2 +- 4 files changed, 106 insertions(+), 172 deletions(-) diff --git a/tongran-modules/tongran-system/src/main/java/com/tongran/system/service/impl/EpsInitialTrafficDataServiceImpl.java b/tongran-modules/tongran-system/src/main/java/com/tongran/system/service/impl/EpsInitialTrafficDataServiceImpl.java index f621537..23db55a 100644 --- a/tongran-modules/tongran-system/src/main/java/com/tongran/system/service/impl/EpsInitialTrafficDataServiceImpl.java +++ b/tongran-modules/tongran-system/src/main/java/com/tongran/system/service/impl/EpsInitialTrafficDataServiceImpl.java @@ -765,13 +765,18 @@ public class EpsInitialTrafficDataServiceImpl implements EpsInitialTrafficDataSe // 获取所有不同的businessCode Set businessCodes = records.stream() .map(EpsMethodChangeRecord::getBusinessCode) + .filter(Objects::nonNull) .collect(Collectors.toSet()); - - // 对每个businessCode计算全天的95带宽 - for (String businessCode : businessCodes) { - // 设置当前要统计的业务代码（需要在queryParam或calculateSegment95中支持businessCode过滤） - queryParam.setBusinessId(businessCode); // 假设queryParam有setBusinessCode方法 - calculateSegment95(queryParam, dailyStartTime, dailyEndTime, calculationMode); + if(businessCodes != null && !businessCodes.isEmpty()){ + // 对每个businessCode计算全天的95带宽 + for (String businessCode : businessCodes) { + // 设置当前要统计的业务代码（需要在queryParam或calculateSegment95中支持businessCode过滤） + queryParam.setBusinessId(businessCode); // 假设queryParam有setBusinessCode方法 + calculateSegment95(queryParam, dailyStartTime, dailyEndTime, calculationMode); + } + }else{ + queryParam.setBusinessId(null); + calculateNormalDeviceBandwidth(queryParam, dailyStartTime, dailyEndTime, calculationMode); } }else{ queryParam.setBusinessId(null); diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java index a16fccf..fda46b3 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java @@ -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 currentInterfaces = rmNetworkInterfaceService.selectRmNetworkInterfaceList(countQuery); - int currentInterfaceCount = currentInterfaces.size(); + // 构建数据库中当前网卡MAC地址的集合 + Set currentMacSet = currentInterfaces.stream() + .map(RmNetworkInterface::getMacAddress) + .collect(Collectors.toSet()); + // 构建新数据中网卡MAC地址的集合 + Set newMacSet = networkInfoList.stream() + .map(NetworkInfo::getMac) + .filter(Objects::nonNull) + .collect(Collectors.toSet()); + + // 找出需要删除的网卡（数据库中有但新数据中没有的） + Set 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 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 macsToDelete) { + for (String macAddress : macsToDelete) { + RmNetworkInterface query = new RmNetworkInterface(); + query.setClientId(clientId); + query.setMacAddress(macAddress); + query.setNewFlag(1); + List oldExits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(query); + if(!oldExits.isEmpty()) { + oldExits.forEach(oldMsg -> { + rmNetworkInterfaceService.deleteRmNetworkInterfaceById(oldMsg.getId()); + }); + } + } + + // 记录删除操作日志 + log.info("删除客户端 {} 的过时网卡信息，MAC地址: {}", clientId, macsToDelete); + } /** * 设置网卡信息公共字段 diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/snmp/scheduler/MultiSwitchCollectionScheduler.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/snmp/scheduler/MultiSwitchCollectionScheduler.java index 1f3e317..c6e4ea9 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/snmp/scheduler/MultiSwitchCollectionScheduler.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/snmp/scheduler/MultiSwitchCollectionScheduler.java @@ -43,6 +43,7 @@ public class MultiSwitchCollectionScheduler { private final Map>> allSwitchTasks = new ConcurrentHashMap<>(); private final Map allSwitchConfigs = new ConcurrentHashMap<>(); private final Map> allSwitchCollectConfigs = new ConcurrentHashMap<>(); + private final Map> heartbeatTasks = new ConcurrentHashMap<>(); // 使用构造器注入而不是@Autowired public MultiSwitchCollectionScheduler(TaskScheduler taskScheduler, @@ -88,11 +89,11 @@ public class MultiSwitchCollectionScheduler { List 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 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> tasks = allSwitchTasks.remove(switchIp); + public void stopSwitchCollection(String clientId) { + Map> 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 getSwitchStatus(String switchIp) { - Map status = new HashMap<>(); - status.put("switchIp", switchIp); - - Map> tasks = allSwitchTasks.get(switchIp); - List 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> taskStatusList = new ArrayList<>(); - for (CollectVo collectVo : collectConfigs) { - Map 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> getAllSwitchesStatus() { - Map> 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 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 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); } } } diff --git a/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml index becfb0e..e6004e8 100644 --- a/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml +++ b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml @@ -177,7 +177,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" from ${tableName} and client_id = #{clientId} - and name = #{name} + and name like concat(#{name},'%') order by create_time desc limit 1