package com.tongran.rocketmq.handler; import com.alibaba.fastjson.JSONObject; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.tongran.common.core.constant.SecurityConstants; import com.tongran.common.core.domain.R; import com.tongran.common.core.enums.MsgEnum; import com.tongran.common.core.utils.DateUtils; import com.tongran.common.core.utils.StringUtils; import com.tongran.common.security.utils.SecurityUtils; import com.tongran.rocketmq.domain.*; import com.tongran.rocketmq.domain.vo.*; import com.tongran.rocketmq.enums.AlarmTypeEnum; import com.tongran.rocketmq.enums.ConditionItemEnum; import com.tongran.rocketmq.model.ProducerMode; import com.tongran.rocketmq.producer.MessageProducer; import com.tongran.rocketmq.service.*; import com.tongran.rocketmq.utils.DataProcessUtil; import com.tongran.rocketmq.utils.JsonDataParser; import com.tongran.rocketmq.utils.SendAlarmPushUtil; import com.tongran.system.api.RemoteRevenueConfigService; import com.tongran.system.api.domain.*; import lombok.extern.slf4j.Slf4j; import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpGet; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; import org.apache.http.util.EntityUtils; import org.redisson.api.RedissonClient; import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.core.RedisOperations; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.SessionCallback; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.math.BigDecimal; import java.math.RoundingMode; import java.util.*; import java.util.concurrent.*; import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; /** * 设备消息处理器 */ @Slf4j @Component @EnableScheduling public class MessageHandler { // 全局HTTP连接池 private static final PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(); private static final CloseableHttpClient httpClient; static { connectionManager.setMaxTotal(100); // 最大连接数 connectionManager.setDefaultMaxPerRoute(50); // 每个路由最大连接数 httpClient = HttpClients.custom() .setConnectionManager(connectionManager) .build(); } // 使用自定义线程池限制并发数 private static final ExecutorService DEVICE_PROCESS_POOL = Executors.newFixedThreadPool(40); // 限制40个并发 private static final ExecutorService IP_QUERY_POOL = Executors.newFixedThreadPool(100); private final Map> messageHandlers = new HashMap<>(); // 硬盘上报状态 String DISK_COUNT_PREFIX = "disk:count:"; // 心跳状态 private static final String HEARTBEAT_STATUS_PREFIX = "heartbeat:status:"; // 心跳时间 private static final String HEARTBEAT_TIME_PREFIX = "heartbeat:time:"; // 心跳告警 private static final String HEARTBEAT_ALERT_PREFIX = "heartbeat:alert:"; String HEARTBEAT_RECOVERY_COUNT_PREFIX = "heartbeat:recovery:count:"; String HEARTBEAT_COUNT_PREFIX = "heartbeat:count:"; // Redis key定义 private static final String USED_PORTS_KEY = "frpc:used_ports"; private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时 private final ConcurrentHashMap> messageBuffer = new ConcurrentHashMap<>(); @Autowired private RedisTemplate redisTemplate; @Autowired private IInitialBandwidthTrafficService initialBandwidthTrafficService; @Autowired private RemoteRevenueConfigService remoteRevenueConfigService; @Autowired private IInitialDockerInfoService initialDockerInfoService; @Autowired private IInitialCpuInfoService initialCpuInfoService; @Autowired private IInitialDiskInfoService initialDiskInfoService; @Autowired private IInitialMemoryInfoService initialMemoryInfoService; @Autowired private IInitialMountPointInfoService initialMountPointInfoService; @Autowired private IInitialHeartbeatListenLogService initialHeartbeatListenLog; @Autowired private IInitialSystemOtherCollectDataService iInitialSystemOtherCollectDataService; @Autowired private IRmResourceRemoteService rmResourceRemoteService; @Autowired private IRmAgentManagementService rmAgentManagementService; @Autowired private DataProcessUtil dataProcessUtil; @Autowired private IRmNetworkInterfaceService rmNetworkInterfaceService; @Autowired private IRmNetworkInterfaceChildService rmNetworkInterfaceChildService; @Autowired private IRmMonitorPolicyService rmMonitorPolicyService; @Autowired private IRmDeploymentPolicyService rmDeploymentPolicyService; @Autowired private IInitialBandwidthTrafficTempService initialBandwidthTrafficTempService; @Autowired private IInitialNetBusinessTrafficTempService initialNetBusinessTrafficTempService; @Autowired private IInitialNetBusinessTrafficService iInitialNetBusinessTrafficService; @Autowired private IRmAlarmLogService rmAlarmLogService; @Autowired private IRmFrpcConfigManageService rmFrpcConfigManageService; @Autowired private IRmPppoeConfigSubService rmPppoeConfigSubService; @Autowired private IAllDiskNameService allDiskNameService; @Autowired private ProducerMode producerMode; @Autowired private RedissonClient redissonClient; @Autowired private IRmOutboundTrafficStatisticsService rmOutboundTrafficStatisticsService; @Autowired private IRmTcpdumpConfigService rmTcpdumpConfigService; @Autowired private SendAlarmPushUtil sendAlarmPushUtil; @Autowired private IRmAlarmThresholdService rmAlarmThresholdService; @Autowired private IInitialMemoryDetailsInfoService initialMemoryDetailsInfoService; /** * 初始化处理器映射 */ @PostConstruct public void init() { registerHandler(MsgEnum.执行脚本策略应答.getValue(), this::handleScriptRspMessage); registerHandler(MsgEnum.Agent版本更新应答.getValue(), this::handleAgentUpdateRspMessage); // 其他类型消息可以单独注册处理器 registerHandler(MsgEnum.注册.getValue(), this::handleRegisterMessage); registerHandler(MsgEnum.获取最新策略.getValue(), this::handleNewPolicyMessage); registerHandler(MsgEnum.CPU上报.getValue(), this::handleCpuMessage); registerHandler(MsgEnum.磁盘上报.getValue(), this::handleDiskMessage); registerHandler(MsgEnum.容器上报.getValue(), this::handleDockerMessage); registerHandler(MsgEnum.内存上报.getValue(), this::handleMemoryMessage); registerHandler(MsgEnum.网络上报.getValue(), this::handleNetMessage); registerHandler(MsgEnum.网络上报重试.getValue(), this::handleNetRecoverMessage); registerHandler(MsgEnum.业务网络上报.getValue(), this::handleBusinessNetMessage); registerHandler(MsgEnum.挂载上报.getValue(), this::handleMountPointMessage); registerHandler(MsgEnum.系统其他上报.getValue(), this::handleOtherSystemMessage); registerHandler(MsgEnum.心跳上报.getValue(), this::handleHeartbeatMessage); registerHandler(MsgEnum.多公网IP探测.getValue(), this::handleNetWorkDelectMessage); registerHandler(MsgEnum.修改frp配置文件应答.getValue(), this::handleUpdateFrpMessage); registerHandler(MsgEnum.macvlan状态上报.getValue(), this::handleMacvlanMessage); registerHandler(MsgEnum.iops结果上报.getValue(), this::handleIopsResultMessage); registerHandler(MsgEnum.tcpdump结果上报.getValue(), this::handleTcpdumpResultMessage); registerHandler(MsgEnum.内存详情上报.getValue(), this::handleMemoryDetailsMessage); } private void handleBusinessNetMessage(DeviceMessage message) { List interfaces = JsonDataParser.parseJsonData(message.getData(), InitialNetBusinessTraffic.class); if(!interfaces.isEmpty()){ String clientId = message.getClientId(); // 时间戳转换 long timestamp = interfaces.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime); // 创建比timestamp少5分钟的时间 long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数 Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒 // 查询临时表信息，计算实际流量值 InitialNetBusinessTrafficTemp temp = new InitialNetBusinessTrafficTemp(); temp.setCreateTime(fiveMinutesEarlierDate); temp.setClientId(clientId); List tempList = initialNetBusinessTrafficTempService.selectInitialNetBusinessTrafficTempList(temp); if(!tempList.isEmpty()){ // 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键 Map tempMap = tempList.stream() .collect(Collectors.toMap( tempItem -> tempItem.getProcessName(), Function.identity(), (existing, replacement) -> existing )); // 2. 预计算除数（避免重复创建对象） BigDecimal divisor = new BigDecimal(300); interfaces.forEach(iface -> { iface.setClientId(clientId); iface.setCreateTime(createTime); // 设置总流量（转换为比特） iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); // 首次采集，速率设为null iface.setInSpeed(null); iface.setOutSpeed(null); iface.setIpv4InSpeed(null); iface.setIpv4OutSpeed(null); iface.setIpv6InSpeed(null); iface.setIpv6OutSpeed(null); // 使用MAC地址+网卡名称作为查找键 String key = iface.getProcessName(); InitialNetBusinessTrafficTemp tempInfo = tempMap.get(key); if (tempInfo != null) { // 计算总流入速率 if (iface.getTotalInSpeed() != null && tempInfo.getTotalInSpeed() != null) { BigDecimal nowInSpeed = new BigDecimal(iface.getTotalInSpeed()); BigDecimal tempInSpeed = new BigDecimal(tempInfo.getTotalInSpeed()); BigDecimal inDiff = nowInSpeed.subtract(tempInSpeed); if (inDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setInSpeed(inDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算总流出速率 if (iface.getTotalOutSpeed() != null && tempInfo.getTotalOutSpeed() != null) { BigDecimal nowOutSpeed = new BigDecimal(iface.getTotalOutSpeed()); BigDecimal tempOutSpeed = new BigDecimal(tempInfo.getTotalOutSpeed()); BigDecimal outDiff = nowOutSpeed.subtract(tempOutSpeed); if (outDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setOutSpeed(outDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv4流入速率 if (iface.getTotalIpv4InSpeed() != null && tempInfo.getTotalIpv4InSpeed() != null) { BigDecimal nowIpv4In = new BigDecimal(iface.getTotalIpv4InSpeed()); BigDecimal tempIpv4In = new BigDecimal(tempInfo.getTotalIpv4InSpeed()); BigDecimal ipv4InDiff = nowIpv4In.subtract(tempIpv4In); if (ipv4InDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv4InSpeed(ipv4InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv4流出速率 if (iface.getTotalIpv4OutSpeed() != null && tempInfo.getTotalIpv4OutSpeed() != null) { BigDecimal nowIpv4Out = new BigDecimal(iface.getTotalIpv4OutSpeed()); BigDecimal tempIpv4Out = new BigDecimal(tempInfo.getTotalIpv4OutSpeed()); BigDecimal ipv4OutDiff = nowIpv4Out.subtract(tempIpv4Out); if (ipv4OutDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv4OutSpeed(ipv4OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv6流入速率 if (iface.getTotalIpv6InSpeed() != null && tempInfo.getTotalIpv6InSpeed() != null) { BigDecimal nowIpv6In = new BigDecimal(iface.getTotalIpv6InSpeed()); BigDecimal tempIpv6In = new BigDecimal(tempInfo.getTotalIpv6InSpeed()); BigDecimal ipv6InDiff = nowIpv6In.subtract(tempIpv6In); if (ipv6InDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv6InSpeed(ipv6InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv6流出速率 if (iface.getTotalIpv6OutSpeed() != null && tempInfo.getTotalIpv6OutSpeed() != null) { BigDecimal nowIpv6Out = new BigDecimal(iface.getTotalIpv6OutSpeed()); BigDecimal tempIpv6Out = new BigDecimal(tempInfo.getTotalIpv6OutSpeed()); BigDecimal ipv6OutDiff = nowIpv6Out.subtract(tempIpv6Out); if (ipv6OutDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv6OutSpeed(ipv6OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } } }); // 清空临时表对应server信息 initialNetBusinessTrafficTempService.deleteBusinessTempMsgByClientId(clientId); }else{ interfaces.forEach(iface -> { iface.setClientId(clientId); iface.setCreateTime(createTime); // 设置总流量（转换为比特） iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); // 首次采集，速率设为null iface.setInSpeed(null); iface.setOutSpeed(null); iface.setIpv4InSpeed(null); iface.setIpv4OutSpeed(null); iface.setIpv6InSpeed(null); iface.setIpv6OutSpeed(null); }); } InitialNetBusinessTraffic data = new InitialNetBusinessTraffic(); // 批量入库集合 data.setList(interfaces); // 临时表 用来计算流量速率 initialNetBusinessTrafficTempService.batchInsertBusinessTemp(interfaces); // 初始流量数据入库 iInitialNetBusinessTrafficService.batchInsertBusinessTraffic(data); }else{ throw new RuntimeException("业务NET流量data数据为空"); } } private void handleMemoryDetailsMessage(DeviceMessage message) { List memoryVoList = JsonDataParser.parseJsonData(message.getData(), MemoryVo.class); String clientId = message.getClientId(); if(memoryVoList != null && !memoryVoList.isEmpty()){ MemoryVo memoryVo = memoryVoList.get(0); // 时间戳转换 long timestamp = memoryVo.getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 List detailsVoList = memoryVo.getDetails(); if(detailsVoList != null && !detailsVoList.isEmpty()){ // 关键：每个clientId有自己独立的key String memoryCountKey = "memory:count:" + clientId; // 1. 给这个客户端的所有内存次数+1 Map memoryCountMap = redisTemplate.opsForHash().entries(memoryCountKey); for (Map.Entry entry : memoryCountMap.entrySet()) { String memoryName = (String) entry.getKey(); String countStr = (String) entry.getValue(); try { int count = Integer.parseInt(countStr) + 1; redisTemplate.opsForHash().put(memoryCountKey, memoryName, String.valueOf(count)); } catch (NumberFormatException e) { redisTemplate.opsForHash().put(memoryCountKey, memoryName, "1"); } } // 2. 处理本次上报的内存 Set reportedMemories = new HashSet<>(); for (InitialMemoryDetailsInfo detailsInfo : detailsVoList) { String memoryName = detailsInfo.getName(); // 假设内存有name字段 reportedMemories.add(memoryName); // 本次上报的内存，次数重置为0 redisTemplate.opsForHash().put(memoryCountKey, memoryName, "0"); detailsInfo.setTotal(memoryVo.getTotal()); detailsInfo.setClientId(clientId); detailsInfo.setCreateTime(createTime); // 存在 detailsInfo.setRemark("1"); initialMemoryDetailsInfoService.insertInitialMemoryDetailsInfo(detailsInfo); } // 3. 检查次数≥3的内存 memoryCountMap = redisTemplate.opsForHash().entries(memoryCountKey); List memoriesToRemove = new ArrayList<>(); for (Map.Entry entry : memoryCountMap.entrySet()) { String memoryName = (String) entry.getKey(); String countStr = (String) entry.getValue(); try { int count = Integer.parseInt(countStr); // 如果次数≥3且本次没上报 if (count >= 3 && !reportedMemories.contains(memoryName)) { // 创建内存缺失记录（根据你的实际业务对象调整） InitialMemoryDetailsInfo missingRecord = new InitialMemoryDetailsInfo(); // 丢失 missingRecord.setRemark("0"); missingRecord.setClientId(clientId); missingRecord.setName(memoryName); initialMemoryDetailsInfoService.insertInitialMemoryDetailsInfo(missingRecord); memoriesToRemove.add(memoryName); } } catch (NumberFormatException e) { memoriesToRemove.add(memoryName); } } // 4. 删除已处理的内存记录 if (!memoriesToRemove.isEmpty()) { redisTemplate.opsForHash().delete(memoryCountKey, memoriesToRemove.toArray()); } } } } private void handleTcpdumpResultMessage(DeviceMessage message) { // 立即提交到专属线程池 DEVICE_PROCESS_POOL.execute(() -> processMessage(message)); } private void processMessage(DeviceMessage message){ List tcpdumpVoList = JsonDataParser.parseJsonData(message.getData(), TcpdumpVo.class); if(tcpdumpVoList != null && !tcpdumpVoList.isEmpty()){ String localProvince = ""; String localIsp = ""; boolean localHasFlag = false; // 根据clientId查询省份信息 RmNetworkInterface networkInterfaceQuery = new RmNetworkInterface(); networkInterfaceQuery.setClientId(message.getClientId()); networkInterfaceQuery.setNewFlag(1); List networkInfoList = rmNetworkInterfaceService.selectRmNetworkInterfaceList(networkInterfaceQuery); if(networkInfoList != null && !networkInfoList.isEmpty()){ for (RmNetworkInterface rmNetworkInterface : networkInfoList) { if(rmNetworkInterface.getProvince() != null && rmNetworkInterface.getIsp() != null && ("1".equals(rmNetworkInterface.getBindIp()) || "3".equals(rmNetworkInterface.getBindIp()))){ localProvince = rmNetworkInterface.getProvince(); localIsp = rmNetworkInterface.getIsp(); localHasFlag = true; break; } } } if(!localHasFlag){ // 查询子网卡 RmNetworkInterfaceChild childQuery = new RmNetworkInterfaceChild(); childQuery.setClientId(message.getClientId()); childQuery.setIpv4Flag(true); List existingChildren = rmNetworkInterfaceChildService.selectRmNetworkInterfaceChildList(childQuery); if(existingChildren != null && !existingChildren.isEmpty()){ RmNetworkInterfaceChild networkInfoChild = existingChildren.get(0); localProvince = networkInfoChild.getProvince(); localIsp = networkInfoChild.getIsp(); } } // 在计算百分比之前，先并行查询所有IP归属地 Map> ipLocationCache = new ConcurrentHashMap<>(); List> futures = new ArrayList<>(); for (TcpdumpVo vo : tcpdumpVoList) { CompletableFuture future = CompletableFuture.runAsync(() -> { ipLocationCache.computeIfAbsent(vo.getIp(), ip -> queryIpLocation(ip)); }, IP_QUERY_POOL); futures.add(future); } // 等待所有查询完成 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); // 计算总数量 Double totalCount = 0.0; // 分别统计IPv4和IPv6 Map> ipTypeMap = new HashMap<>(); ipTypeMap.put("IPv4", new ArrayList<>()); ipTypeMap.put("IPv6", new ArrayList<>()); // 按类型分类 for (TcpdumpVo tcpdumpVo : tcpdumpVoList) { totalCount += tcpdumpVo.getCount(); String ip = tcpdumpVo.getIp(); String type = checkIPVersion(ip); if ("IPv4".equals(type)) { ipTypeMap.get("IPv4").add(tcpdumpVo); } else if ("IPv6".equals(type)) { ipTypeMap.get("IPv6").add(tcpdumpVo); } } // 构建content内容 StringBuilder content = new StringBuilder(); BigDecimal v4TotalRate = BigDecimal.ZERO; BigDecimal v6TotalRate = BigDecimal.ZERO; BigDecimal totalRate = BigDecimal.ZERO; // 处理IPv4统计 if (!ipTypeMap.get("IPv4").isEmpty()) { content.append("##### V4 统计 ##### ##\n"); // 按省份运营商分组统计（过滤本地省份） Map v4StatMap = new HashMap<>(); for (TcpdumpVo tcpdumpVo : ipTypeMap.get("IPv4")) { Map locationInfo = ipLocationCache.get(tcpdumpVo.getIp()); if (locationInfo != null) { String operator = locationInfo.get("operator"); String province = locationInfo.get("province"); // 过滤本地省份和空值 if (province != null && operator != null) { // 标准化运营商名称 String stdLocalIsp = localIsp != null ? localIsp.replace("中国", "") : ""; String stdOperator = operator.replace("中国", ""); // 省不一样或者省一样但运营商不一样 if (!province.equals(localProvince) || !stdLocalIsp.equals(stdOperator)) { String key = province + operator; // 根据图片格式，省份和运营商之间没有空格 double currentCount = v4StatMap.getOrDefault(key, 0.0); v4StatMap.put(key, currentCount + tcpdumpVo.getCount()); } } } } // 计算百分比并排序 List> sortedV4List = new ArrayList<>(v4StatMap.entrySet()); sortedV4List.sort((a, b) -> b.getValue().compareTo(a.getValue())); // 降序排序 // 先计算所有百分比并存储 for (Map.Entry entry : sortedV4List) { BigDecimal percentage = new BigDecimal(entry.getValue()) .divide(new BigDecimal(totalCount), 10, RoundingMode.HALF_UP) .multiply(new BigDecimal(100)) .setScale(2, RoundingMode.HALF_UP); content.append(entry.getKey()).append(": ").append(percentage).append("%\n"); v4TotalRate = v4TotalRate.add(percentage); } // 在记录的位置插入总计信息 totalRate = totalRate.add(v4TotalRate); } // 处理IPv6统计 if (!ipTypeMap.get("IPv6").isEmpty()) { content.append("##### V6 统计 ##### ##\n"); // 按省份运营商分组统计（过滤本地省份） Map v6StatMap = new HashMap<>(); for (TcpdumpVo tcpdumpVo : ipTypeMap.get("IPv6")) { Map locationInfo = ipLocationCache.get(tcpdumpVo.getIp()); if (locationInfo != null) { String operator = locationInfo.get("operator"); String province = locationInfo.get("province"); // 过滤本地省份和空值 if (province != null && operator != null) { // 标准化运营商名称 String stdLocalIsp = localIsp != null ? localIsp.replace("中国", "") : ""; String stdOperator = operator.replace("中国", ""); // 省不一样或者省一样但运营商不一样 if (!province.equals(localProvince) || !stdLocalIsp.equals(stdOperator)) { String key = province + operator; // 根据图片格式，省份和运营商之间没有空格 double currentCount = v6StatMap.getOrDefault(key, 0.0); v6StatMap.put(key, currentCount + tcpdumpVo.getCount()); } } } } // 计算百分比并排序 List> sortedV6List = new ArrayList<>(v6StatMap.entrySet()); sortedV6List.sort((a, b) -> b.getValue().compareTo(a.getValue())); // 降序排序 for (Map.Entry entry : sortedV6List) { BigDecimal percentage = new BigDecimal(entry.getValue()) .divide(new BigDecimal(totalCount), 10, RoundingMode.HALF_UP) .multiply(new BigDecimal(100)) .setScale(2, RoundingMode.HALF_UP); content.append(entry.getKey()).append(": ").append(percentage).append("%\n"); v6TotalRate = v6TotalRate.add(percentage); } totalRate = totalRate.add(v6TotalRate); } // 添加总占比 RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote(); updateData.setClientId(message.getClientId()); updateData.setOutboundRate(totalRate); remoteRevenueConfigService.innerUpdateRegist(updateData, SecurityConstants.INNER); RmOutboundTrafficStatistics insertData = new RmOutboundTrafficStatistics(); insertData.setClientId(message.getClientId()); insertData.setTotalRate(totalRate); insertData.setIpv4Rate(v4TotalRate); insertData.setIpv6Rate(v6TotalRate); insertData.setDescription(content.toString()); rmOutboundTrafficStatisticsService.insertRmOutboundTrafficStatistics(insertData); // 查询告警阈值 RmAlarmThreshold thresholdQuery = new RmAlarmThreshold(); thresholdQuery.setAlarmType(AlarmTypeEnum.出省流量占比过高.getCode()); thresholdQuery.setConditionItem(ConditionItemEnum.出省流量占比过高.getCode()); List rmAlarmThresholdList = rmAlarmThresholdService.selectRmAlarmThresholdList(thresholdQuery); if(rmAlarmThresholdList != null && !rmAlarmThresholdList.isEmpty()){ String operator = rmAlarmThresholdList.get(0).getCompareOperator(); BigDecimal threshold = rmAlarmThresholdList.get(0).getThresholdValue(); if(operator != null){ // 根据运算符1大于，2小于，3等于 判断是否进行告警 boolean shouldAlarm = false; // 根据运算符判断是否进行告警 switch (operator) { case "1": // 大于 shouldAlarm = totalRate.compareTo(threshold) > 0; break; case "2": // 小于 shouldAlarm = totalRate.compareTo(threshold) < 0; break; case "3": // 等于 shouldAlarm = totalRate.compareTo(threshold) == 0; break; default: log.warn("未知的运算符: {}", operator); } if(shouldAlarm){ RmAlarmLog rmAlarmLog = new RmAlarmLog(); rmAlarmLog.setAlarmType(AlarmTypeEnum.出省流量占比过高.getCode()); rmAlarmLog.setClientId(message.getClientId()); rmAlarmLog.setAlarmTime(DateUtils.getNowDate()); rmAlarmLog.setAlarmContent("服务器" + message.getClientId() + "出省流量占比过高"); rmAlarmLogService.insertRmAlarmLog(rmAlarmLog); // 推送消息 sendAlarmPushUtil.sendAlarmPush(rmAlarmLog, AlarmTypeEnum.出省流量占比过高.getMsg()); } } } } } private void handleIopsResultMessage(DeviceMessage message) { List diskIopsResultVoList = JsonDataParser.parseJsonData(message.getData(), DiskIopsResultVo.class); if(diskIopsResultVoList != null && !diskIopsResultVoList.isEmpty()){ DiskIopsResultVo diskIopsResultVo = diskIopsResultVoList.get(0); // 时间戳转换 long timestamp = diskIopsResultVo.getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 AllDiskName allDiskName = new AllDiskName(); allDiskName.setClientId(message.getClientId()); allDiskName.setName(diskIopsResultVo.getName()); allDiskName.setReadIops(diskIopsResultVo.getReadIops()); allDiskName.setWriteIops(diskIopsResultVo.getWriteIops()); allDiskNameService.updateAllDiskName(allDiskName); } } private void handleMacvlanMessage(DeviceMessage message) { List macVlanRspVoList = JsonDataParser.parseJsonData(message.getData(), MacVlanRspVo.class); if(!macVlanRspVoList.isEmpty()){ MacVlanRspVo macVlanRspVo = macVlanRspVoList.get(0); // 时间戳转换 long timestamp = macVlanRspVo.getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 List list = macVlanRspVo.getMacVlans(); for (MacVlanRspVo.MacVlanVo macVlanVo : list) { RmPppoeConfigSub rmPppoeConfigSub = new RmPppoeConfigSub(); rmPppoeConfigSub.setVlanId(Long.valueOf(macVlanVo.getVlanId())); rmPppoeConfigSub.setSerialNumber(Long.valueOf(macVlanVo.getMid())); rmPppoeConfigSub.setClientId(message.getClientId()); rmPppoeConfigSub.setStatus(macVlanVo.getStatus()); rmPppoeConfigSub.setUpdateTime(createTime); rmPppoeConfigSubService.updateRmPppoeConfigSubByVlan(rmPppoeConfigSub); } } } private void handleUpdateFrpMessage(DeviceMessage message) { List rspVoList = JsonDataParser.parseJsonData(message.getData(), FrpRspVo.class); if (!rspVoList.isEmpty()) { FrpRspVo rsp = rspVoList.get(0); // 时间戳转换 long timestamp = rsp.getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 if(rsp.getResCode() == 1) { // 运行成功 修改相关信息 RmFrpcConfigManage frpcData = new RmFrpcConfigManage(); frpcData.setClientId(message.getClientId()); frpcData.setFrpcName(rsp.getName()); frpcData.setFrpcStatus(rsp.getIsRunning()); frpcData.setFrpcLocalPort(rsp.getLocalPort()); frpcData.setFrpcLocalIp(rsp.getLocalIP()); frpcData.setFrpcCreateTime(createTime); frpcData.setCreateTime(DateUtils.getNowDate()); frpcData.setFrpcConnectionStatus(2L); frpcData.setFrpcRemotePort(rsp.getRemotePort()); frpcData.setFrpcType(rsp.getType()); frpcData.setFrpcServerAddr(rsp.getServerAddr()); frpcData.setFrpcServerPort(rsp.getServerPort()); rmFrpcConfigManageService.insertRmFrpcConfigManage(frpcData); } else if(rsp.getResCode() == 2){ RmFrpcConfigManage configManage = new RmFrpcConfigManage(); configManage.setFrpcRemotePort(rsp.getRemotePort()); rmFrpcConfigManageService.addFrpConnect(configManage); } else if(rsp.getResCode() == 3){ // frpc状态上报 RmFrpcConfigManage frpcData = new RmFrpcConfigManage(); frpcData.setClientId(message.getClientId()); frpcData.setFrpcName(rsp.getName()); frpcData.setFrpcType(rsp.getType()); frpcData.setFrpcStatus(rsp.getIsRunning()); frpcData.setFrpcLocalPort(rsp.getLocalPort()); frpcData.setFrpcLocalIp(rsp.getLocalIP()); frpcData.setFrpcRemotePort(rsp.getRemotePort()); frpcData.setFrpcServerAddr(rsp.getServerAddr()); frpcData.setFrpcServerPort(rsp.getServerPort()); frpcData.setUpdateTime(DateUtils.getNowDate()); rmFrpcConfigManageService.insertRmFrpcConfigManage(frpcData); } else{ if(rsp.getRemotePort() != null){ // 生成失败 ，释放端口资源 removeUsedPort(Integer.parseInt(rsp.getRemotePort())); } log.error("生成连接失败：{}",rsp.getResMsg()); } } } /** * 添加端口到已用集合 */ public boolean addUsedPort(int port) { Long result = redisTemplate.opsForSet().add(USED_PORTS_KEY, String.valueOf(port)); return result != null && result > 0; } /** * 从已用集合中移除端口 */ public boolean removeUsedPort(int port) { Long result = redisTemplate.opsForSet().remove(USED_PORTS_KEY, String.valueOf(port)); return result != null && result > 0; } /** * 保存网卡信息 * @param message */ private void handleNetWorkDelectMessage(DeviceMessage message) { List interfaces = JsonDataParser.parseJsonData(message.getData(), RegisterMsgVo.class); if(!interfaces.isEmpty()) { String clientId = message.getClientId(); RegisterMsgVo registerMsg = interfaces.get(0); // 处理网卡信息 processNetworkInterfaces(registerMsg, clientId, false); } } /** * 获取最新策略 * @param deviceMessage */ private void handleNewPolicyMessage(DeviceMessage deviceMessage) { List interfaces = JsonDataParser.parseJsonData(deviceMessage.getData(), RegisterMsgVo.class); if(!interfaces.isEmpty()) { RegisterMsgVo registerMsgVo = interfaces.get(0); String clientId = registerMsgVo.getClientId(); // 如果未下发监控策略，下发 rmMonitorPolicyService.issuePolicyMsgByClientId(clientId); // 如果未下发服务器脚本策略，下发 rmDeploymentPolicyService.issueDeployPolicyMsgByClientId(clientId); // 如果路由有变化，更新路由信息 rmNetworkInterfaceService.updateRouteMsg(clientId); // 如果业务网卡名称有变化，更新防火墙策略 rmNetworkInterfaceService.issueNetName(clientId); } } /** * 服务器注册 * @param message */ private void handleRegisterMessage(DeviceMessage message) { List interfaces = JsonDataParser.parseJsonData(message.getData(), RegisterMsgVo.class); if(!interfaces.isEmpty()) { String clientId = message.getClientId(); RegisterMsgVo registerMsg = interfaces.get(0); // 自动注册服务器信息 RmRegisterMsgRemote rmRegisterMsgRemote = new RmRegisterMsgRemote(); BeanUtils.copyProperties(registerMsg, rmRegisterMsgRemote); int rows = remoteRevenueConfigService.innerAddRegist(rmRegisterMsgRemote, SecurityConstants.INNER).getData(); if(rows == 2){ // 注册成功，下发优先级为0的策略 rmMonitorPolicyService.issueDefaultPolicyByClientId(message.getClientId()); // 下发tcpdump默认策略 String detectTimes = "20:00:00"; RmTcpdumpConfig rmTcpdumpConfig = rmTcpdumpConfigService.selectRmTcpdumpConfigByClientId(clientId); if(rmTcpdumpConfig != null){ detectTimes = detectTimes + "," + rmTcpdumpConfig.getDetectTimes(); } RmTcpdumpConfig tcpdumpInsertData = new RmTcpdumpConfig(); tcpdumpInsertData.setClientId(clientId); tcpdumpInsertData.setDetectFlag(1); tcpdumpInsertData.setFrequency("1"); tcpdumpInsertData.setDetectTimes(detectTimes); rmTcpdumpConfigService.insertRmTcpdumpConfig(tcpdumpInsertData); // agent更新表插入数据 // 存储更新结果 // agent更新结果存储 RmAgentManagement query = new RmAgentManagement(); query.setClientId(clientId); List agentManagements = rmAgentManagementService.selectRmAgentManagementList(query); if(agentManagements == null || agentManagements.isEmpty()) { RmAgentManagement insertAgentUpdateData = new RmAgentManagement(); insertAgentUpdateData.setClientId(clientId); rmAgentManagementService.insertRmAgentManagement(insertAgentUpdateData); } } // 处理网卡信息 processNetworkInterfaces(registerMsg, clientId, true); } } /** * agent更新响应 * @param message */ private void handleAgentUpdateRspMessage(DeviceMessage message) { List rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class); if (!rspVoList.isEmpty()) { RspVo rsp = rspVoList.get(0); if(rsp.getResCode() == 1){ RmAgentManagement rmAgentManagement = new RmAgentManagement(); rmAgentManagement.setClientId(message.getClientId()); rmAgentManagement.setLastUpdateResult("1"); rmAgentManagement.setLastUpdateTime(DateUtils.getNowDate()); rmAgentManagementService.updateRmAgentManagementByHardwareSn(rmAgentManagement); }else{ RmAgentManagement rmAgentManagement = new RmAgentManagement(); rmAgentManagement.setClientId(message.getClientId()); rmAgentManagement.setLastUpdateResult("0"); rmAgentManagement.setLastUpdateTime(DateUtils.getNowDate()); rmAgentManagementService.updateRmAgentManagementByHardwareSn(rmAgentManagement); } } } /** * 注册消息处理器 */ private void registerHandler(String dataType, Consumer handler) { messageHandlers.put(dataType, handler); } /** * 处理设备消息（对外暴露的主方法） */ public void handleMessage(DeviceMessage message) { String dataType = message.getDataType(); Consumer handler = messageHandlers.get(dataType); if (handler != null) { handler.accept(message); } else { log.warn("未知数据类型：{}", dataType); } } // ========== 具体的消息处理方法 ========== /** * 网络流量数据入库 * @param message */ private void handleNetMessage(DeviceMessage message) { List interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class); if(!interfaces.isEmpty()){ String clientId = message.getClientId(); // 时间戳转换 long timestamp = interfaces.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime); // 创建比timestamp少5分钟的时间 long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数 Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒 // 查询临时表信息，计算实际流量值 InitialBandwidthTrafficTemp temp = new InitialBandwidthTrafficTemp(); temp.setCreateTime(fiveMinutesEarlierDate); temp.setClientId(clientId); List tempList = initialBandwidthTrafficTempService.selectInitialBandwidthTrafficTempList(temp); if(!tempList.isEmpty()){ // 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键 Map tempMap = tempList.stream() .collect(Collectors.toMap( tempItem -> generateKey(tempItem.getMac(), tempItem.getName()), Function.identity(), (existing, replacement) -> existing )); // 2. 预计算除数（避免重复创建对象） BigDecimal divisor = new BigDecimal(300); interfaces.forEach(iface -> { iface.setClientId(clientId); iface.setCreateTime(createTime); // 设置总流量（转换为比特） iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); // 首次采集，速率设为null iface.setInSpeed(null); iface.setOutSpeed(null); iface.setIpv4InSpeed(null); iface.setIpv4OutSpeed(null); iface.setIpv6InSpeed(null); iface.setIpv6OutSpeed(null); // 使用MAC地址+网卡名称作为查找键 String key = generateKey(iface.getMac(), iface.getName()); InitialBandwidthTrafficTemp tempInfo = tempMap.get(key); if (tempInfo != null) { // 计算总流入速率 if (iface.getTotalInSpeed() != null && tempInfo.getTotalInSpeed() != null) { BigDecimal nowInSpeed = new BigDecimal(iface.getTotalInSpeed()); BigDecimal tempInSpeed = new BigDecimal(tempInfo.getTotalInSpeed()); BigDecimal inDiff = nowInSpeed.subtract(tempInSpeed); if (inDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setInSpeed(inDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算总流出速率 if (iface.getTotalOutSpeed() != null && tempInfo.getTotalOutSpeed() != null) { BigDecimal nowOutSpeed = new BigDecimal(iface.getTotalOutSpeed()); BigDecimal tempOutSpeed = new BigDecimal(tempInfo.getTotalOutSpeed()); BigDecimal outDiff = nowOutSpeed.subtract(tempOutSpeed); if (outDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setOutSpeed(outDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv4流入速率 if (iface.getTotalIpv4InSpeed() != null && tempInfo.getTotalIpv4InSpeed() != null) { BigDecimal nowIpv4In = new BigDecimal(iface.getTotalIpv4InSpeed()); BigDecimal tempIpv4In = new BigDecimal(tempInfo.getTotalIpv4InSpeed()); BigDecimal ipv4InDiff = nowIpv4In.subtract(tempIpv4In); if (ipv4InDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv4InSpeed(ipv4InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv4流出速率 if (iface.getTotalIpv4OutSpeed() != null && tempInfo.getTotalIpv4OutSpeed() != null) { BigDecimal nowIpv4Out = new BigDecimal(iface.getTotalIpv4OutSpeed()); BigDecimal tempIpv4Out = new BigDecimal(tempInfo.getTotalIpv4OutSpeed()); BigDecimal ipv4OutDiff = nowIpv4Out.subtract(tempIpv4Out); if (ipv4OutDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv4OutSpeed(ipv4OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv6流入速率 if (iface.getTotalIpv6InSpeed() != null && tempInfo.getTotalIpv6InSpeed() != null) { BigDecimal nowIpv6In = new BigDecimal(iface.getTotalIpv6InSpeed()); BigDecimal tempIpv6In = new BigDecimal(tempInfo.getTotalIpv6InSpeed()); BigDecimal ipv6InDiff = nowIpv6In.subtract(tempIpv6In); if (ipv6InDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv6InSpeed(ipv6InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv6流出速率 if (iface.getTotalIpv6OutSpeed() != null && tempInfo.getTotalIpv6OutSpeed() != null) { BigDecimal nowIpv6Out = new BigDecimal(iface.getTotalIpv6OutSpeed()); BigDecimal tempIpv6Out = new BigDecimal(tempInfo.getTotalIpv6OutSpeed()); BigDecimal ipv6OutDiff = nowIpv6Out.subtract(tempIpv6Out); if (ipv6OutDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv6OutSpeed(ipv6OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } } }); // 清空临时表对应server信息 initialBandwidthTrafficTempService.deleteTempMsgByClientId(clientId); }else{ interfaces.forEach(iface -> { iface.setClientId(clientId); iface.setCreateTime(createTime); // 设置总流量（转换为比特） iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); // 首次采集，速率设为null iface.setInSpeed(null); iface.setOutSpeed(null); iface.setIpv4InSpeed(null); iface.setIpv4OutSpeed(null); iface.setIpv6InSpeed(null); iface.setIpv6OutSpeed(null); }); } InitialBandwidthTraffic data = new InitialBandwidthTraffic(); // 批量入库集合 data.setList(interfaces); // 临时表 用来计算流量速率 initialBandwidthTrafficTempService.batchInsertServerTemp(interfaces); // 初始流量数据入库 initialBandwidthTrafficService.batchInsert(data); EpsInitialTrafficDataRemote epsInitialTrafficDataRemote = new EpsInitialTrafficDataRemote(); epsInitialTrafficDataRemote.setStartTime(timeStr); epsInitialTrafficDataRemote.setEndTime(timeStr); // 复制到业务初始库 remoteRevenueConfigService.autoSaveServiceTrafficData(epsInitialTrafficDataRemote, SecurityConstants.INNER); }else{ throw new RuntimeException("NET流量data数据为空"); } } /** * 存储时间戳到Redis * @param clientId 客户端ID * @param timestamp 时间戳 */ private void storeTimestampToRedis(String clientId, long timestamp) { try { String redisKey = "net_traffic_timestamps:" + clientId; String existingTimestamps = redisTemplate.opsForValue().get(redisKey); if (existingTimestamps != null && !existingTimestamps.isEmpty()) { // 已存在时间戳，追加新的时间戳 String newTimestamps = existingTimestamps + "," + timestamp; redisTemplate.opsForValue().set(redisKey, newTimestamps); } else { // 首次存储 redisTemplate.opsForValue().set(redisKey, String.valueOf(timestamp)); } } catch (Exception e) { // Redis操作异常处理，记录日志但不中断主流程 log.error("存储时间戳到Redis失败，clientId: {}, timestamp: {}", clientId, timestamp, e); } } private void handleNetRecoverMessage(DeviceMessage message) { String clientId = message.getClientId(); List interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class); if (interfaces.isEmpty()) { return; } boolean isLast = interfaces.get(0).isLastTrafficFlag(); // 1. 消息入队 LinkedBlockingQueue queue = messageBuffer .computeIfAbsent(clientId, k -> new LinkedBlockingQueue<>()); queue.offer(message); // 2. 如果是最后一条消息，开始处理（此时前面的消息肯定都已经到了） if (isLast) { processAllMessagesInOrder(clientId); } } private void processAllMessagesInOrder(String clientId) { LinkedBlockingQueue queue = messageBuffer.get(clientId); if (queue == null || queue.isEmpty()) { return; } // 1. 取出所有消息 List allMessages = new ArrayList<>(); queue.drainTo(allMessages); // 2. 按时间戳排序 allMessages.sort((msg1, msg2) -> { long ts1 = JsonDataParser.parseJsonData(msg1.getData(), InitialBandwidthTraffic.class) .get(0).getTimestamp(); long ts2 = JsonDataParser.parseJsonData(msg2.getData(), InitialBandwidthTraffic.class) .get(0).getTimestamp(); return Long.compare(ts1, ts2); }); // 3. 顺序处理 for (DeviceMessage message : allMessages) { try { processNetRecoverMessageInternal(message); } catch (Exception e) { log.error("处理设备{}消息失败", clientId, e); } } log.info("设备{}顺序处理完成，共处理{}条消息", clientId, allMessages.size()); // 4. 清理 messageBuffer.remove(clientId); } /** * 网络重试流量数据入库 * @param message */ private void processNetRecoverMessageInternal(DeviceMessage message) { List interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class); if(!interfaces.isEmpty()){ String clientId = message.getClientId(); // 时间戳转换 long timestamp = interfaces.get(0).getTimestamp(); boolean lasttrafficFlag = interfaces.get(0).isLastTrafficFlag(); // 时间戳存储到redis storeTimestampToRedis(clientId, timestamp); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime); // 判断数据库中是否已存在数据 InitialBandwidthTraffic countQuery = new InitialBandwidthTraffic(); countQuery.setClientId(clientId); countQuery.setCreateTime(createTime); countQuery.setStartTime(timeStr); int exitsCount = initialBandwidthTrafficService.countByClientIdAndTime(countQuery); // 创建比timestamp少5分钟的时间 long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数 Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒 // 查询临时表信息，计算实际流量值 InitialBandwidthTrafficTemp temp = new InitialBandwidthTrafficTemp(); temp.setCreateTime(fiveMinutesEarlierDate); temp.setClientId(clientId); List tempList = initialBandwidthTrafficTempService.selectInitialBandwidthTrafficRecoverTempList(temp); if(!tempList.isEmpty()){ // 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键 Map tempMap = tempList.stream() .collect(Collectors.toMap( tempItem -> generateKey(tempItem.getMac(), tempItem.getName()), Function.identity(), (existing, replacement) -> existing )); // 2. 预计算除数（避免重复创建对象） BigDecimal divisor = new BigDecimal(300); interfaces.forEach(iface -> { iface.setClientId(clientId); iface.setCreateTime(createTime); // 设置总流量（转换为比特） iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); // 首次采集，速率设为null iface.setInSpeed(null); iface.setOutSpeed(null); iface.setIpv4InSpeed(null); iface.setIpv4OutSpeed(null); iface.setIpv6InSpeed(null); iface.setIpv6OutSpeed(null); // 使用MAC地址+网卡名称作为查找键 String key = generateKey(iface.getMac(), iface.getName()); InitialBandwidthTrafficTemp tempInfo = tempMap.get(key); if (tempInfo != null) { // 计算总流入速率 if (iface.getTotalInSpeed() != null && tempInfo.getTotalInSpeed() != null) { BigDecimal nowInSpeed = new BigDecimal(iface.getTotalInSpeed()); BigDecimal tempInSpeed = new BigDecimal(tempInfo.getTotalInSpeed()); BigDecimal inDiff = nowInSpeed.subtract(tempInSpeed); if (inDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setInSpeed(inDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算总流出速率 if (iface.getTotalOutSpeed() != null && tempInfo.getTotalOutSpeed() != null) { BigDecimal nowOutSpeed = new BigDecimal(iface.getTotalOutSpeed()); BigDecimal tempOutSpeed = new BigDecimal(tempInfo.getTotalOutSpeed()); BigDecimal outDiff = nowOutSpeed.subtract(tempOutSpeed); if (outDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setOutSpeed(outDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv4流入速率 if (iface.getTotalIpv4InSpeed() != null && tempInfo.getTotalIpv4InSpeed() != null) { BigDecimal nowIpv4In = new BigDecimal(iface.getTotalIpv4InSpeed()); BigDecimal tempIpv4In = new BigDecimal(tempInfo.getTotalIpv4InSpeed()); BigDecimal ipv4InDiff = nowIpv4In.subtract(tempIpv4In); if (ipv4InDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv4InSpeed(ipv4InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv4流出速率 if (iface.getTotalIpv4OutSpeed() != null && tempInfo.getTotalIpv4OutSpeed() != null) { BigDecimal nowIpv4Out = new BigDecimal(iface.getTotalIpv4OutSpeed()); BigDecimal tempIpv4Out = new BigDecimal(tempInfo.getTotalIpv4OutSpeed()); BigDecimal ipv4OutDiff = nowIpv4Out.subtract(tempIpv4Out); if (ipv4OutDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv4OutSpeed(ipv4OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv6流入速率 if (iface.getTotalIpv6InSpeed() != null && tempInfo.getTotalIpv6InSpeed() != null) { BigDecimal nowIpv6In = new BigDecimal(iface.getTotalIpv6InSpeed()); BigDecimal tempIpv6In = new BigDecimal(tempInfo.getTotalIpv6InSpeed()); BigDecimal ipv6InDiff = nowIpv6In.subtract(tempIpv6In); if (ipv6InDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv6InSpeed(ipv6InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } // 计算IPv6流出速率 if (iface.getTotalIpv6OutSpeed() != null && tempInfo.getTotalIpv6OutSpeed() != null) { BigDecimal nowIpv6Out = new BigDecimal(iface.getTotalIpv6OutSpeed()); BigDecimal tempIpv6Out = new BigDecimal(tempInfo.getTotalIpv6OutSpeed()); BigDecimal ipv6OutDiff = nowIpv6Out.subtract(tempIpv6Out); if (ipv6OutDiff.compareTo(BigDecimal.ZERO) >= 0) { iface.setIpv6OutSpeed(ipv6OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); } } } }); // 清空临时表对应server信息 InitialBandwidthTrafficTemp delQuery = new InitialBandwidthTrafficTemp(); delQuery.setClientId(clientId); delQuery.setCreateTime(fiveMinutesEarlierDate); initialBandwidthTrafficTempService.deleteTempMsgByClientIdAndTime(delQuery); } else { interfaces.forEach(iface -> { iface.setClientId(clientId); iface.setCreateTime(createTime); // 设置总流量（转换为比特） iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); // 首次采集，速率设为null iface.setInSpeed(null); iface.setOutSpeed(null); iface.setIpv4InSpeed(null); iface.setIpv4OutSpeed(null); iface.setIpv6InSpeed(null); iface.setIpv6OutSpeed(null); }); } // 只有在数据不存在时才执行入库操作 if (exitsCount == 0) { InitialBandwidthTraffic data = new InitialBandwidthTraffic(); // 批量入库集合 data.setList(interfaces); // 临时表 用来计算流量速率 initialBandwidthTrafficTempService.batchInsertServerRecoverTemp(interfaces); // 初始流量数据入库 initialBandwidthTrafficService.batchInsertRecoverTraffic(data); EpsInitialTrafficDataRemote epsInitialTrafficDataRemote = new EpsInitialTrafficDataRemote(); epsInitialTrafficDataRemote.setStartTime(timeStr); epsInitialTrafficDataRemote.setEndTime(timeStr); // 复制到业务初始库 remoteRevenueConfigService.autoSaveServiceRecoverTrafficData(epsInitialTrafficDataRemote, SecurityConstants.INNER); } else { // 数据已存在时，只更新临时表，确保后续计算正确 initialBandwidthTrafficTempService.batchInsertServerRecoverTemp(interfaces); } if(lasttrafficFlag){ // 把redis中存储的时间戳提取出来，删除redis中的时间戳 processExitsTraffic(clientId); } } else { throw new RuntimeException("NET流量data数据为空"); } } private void processExitsTraffic(String clientId) { try { String redisKey = "net_traffic_timestamps:" + clientId; String storedTimestamps = redisTemplate.opsForValue().get(redisKey); if (storedTimestamps != null && !storedTimestamps.isEmpty()) { // 删除Redis中的时间戳数据 redisTemplate.delete(redisKey); // 构建磁盘下发信息 PolicyTypeVo policyTypeVo = new PolicyTypeVo(); policyTypeVo.setTrafficUsedTimestamp(storedTimestamps); MessageProducer messageProducer = new MessageProducer(); String configJson = JSONObject.toJSONString(policyTypeVo); DeviceMessage message = new DeviceMessage(); message.setClientId(clientId); message.setData(configJson); message.setDataType(MsgEnum.获取最新策略应答.getValue()); messageProducer.sendAsyncProducerMessage( producerMode.getAgentTopic(), "", "", JSONObject.toJSONString(message) ); } } catch (Exception e) { log.error("提取并删除Redis时间戳失败，clientId: {}", clientId, e); } } private String generateKey(String mac, String name) { if (mac == null) { mac = ""; } if (name == null) { name = ""; } return mac + "|" + name; } /** * docker数据入库 * @param message */ private void handleDockerMessage(DeviceMessage message) { List dockers = JsonDataParser.parseJsonData(message.getData(), InitialDockerInfo.class); if(!dockers.isEmpty()){ // 时间戳转换 long timestamp = dockers.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 dockers.forEach(iface -> { iface.setClientId(message.getClientId()); iface.setCreateTime(createTime); }); // 初始容器数据入库 initialDockerInfoService.batchInsertInitialDockerInfo(dockers, createTime); }else{ throw new RuntimeException("DOCKER容器data数据为空"); } } /** * cpu数据入库 * @param message */ private void handleCpuMessage(DeviceMessage message) { List cpus = JsonDataParser.parseJsonData(message.getData(),InitialCpuInfo.class); // 时间戳转换 long timestamp = cpus.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 if(!cpus.isEmpty()){ cpus.forEach(iface -> { iface.setClientId(message.getClientId()); iface.setCreateTime(createTime); if(iface.getTemperature() != null && iface.getTemperature() == 0L){ iface.setTemperature(null); } }); // 初始CPU数据入库 initialCpuInfoService.batchInsertInitialCpuInfo(cpus, createTime); }else{ throw new RuntimeException("CPUdata数据为空"); } } /** * 磁盘数据入库 * @param message */ private void handleDiskMessage(DeviceMessage message) { List disks = JsonDataParser.parseJsonData(message.getData(), InitialDiskInfo.class); String clientId = message.getClientId(); if (disks == null || disks.isEmpty()) { throw new RuntimeException("磁盘data数据为空"); } long timestamp = disks.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // disks.forEach(disk -> { // disk.setClientId(clientId); // disk.setCreateTime(createTime); // }); // 关键：每个clientId有自己独立的key String diskCountKey = DISK_COUNT_PREFIX + clientId; // 1. 给这个客户端的所有磁盘次数+1 Map diskCountMap = redisTemplate.opsForHash().entries(diskCountKey); for (Map.Entry entry : diskCountMap.entrySet()) { String diskName = (String) entry.getKey(); String countStr = (String) entry.getValue(); try { int count = Integer.parseInt(countStr) + 1; redisTemplate.opsForHash().put(diskCountKey, diskName, String.valueOf(count)); } catch (NumberFormatException e) { redisTemplate.opsForHash().put(diskCountKey, diskName, "1"); } } // 2. 处理本次上报的磁盘 Set reportedDisks = new HashSet<>(); disks.forEach(disk -> { String diskName = disk.getName(); reportedDisks.add(diskName); // 本次上报的磁盘，次数重置为0 redisTemplate.opsForHash().put(diskCountKey, diskName, "0"); disk.setClientId(clientId); disk.setCreateTime(createTime); }); // 3. 检查次数≥3的磁盘 diskCountMap = redisTemplate.opsForHash().entries(diskCountKey); List disksToRemove = new ArrayList<>(); for (Map.Entry entry : diskCountMap.entrySet()) { String diskName = (String) entry.getKey(); String countStr = (String) entry.getValue(); try { int count = Integer.parseInt(countStr); // 如果次数≥3且本次没上报 if (count >= 3 && !reportedDisks.contains(diskName)) { AllDiskName allDiskName = new AllDiskName(); allDiskName.setStatus(0); allDiskName.setClientId(clientId); allDiskName.setName(diskName); allDiskNameService.updateAllDiskName(allDiskName); disksToRemove.add(diskName); // 磁盘缺失，触发告警 RmAlarmLog rmAlarmLog = new RmAlarmLog(); rmAlarmLog.setClientId(clientId); rmAlarmLog.setAlarmTime(DateUtils.getNowDate()); rmAlarmLog.setAlarmType(AlarmTypeEnum.磁盘缺失.getCode()); rmAlarmLog.setAlarmContent("服务器" + clientId + "磁盘缺失，磁盘名称:" + diskName); rmAlarmLogService.insertRmAlarmLog(rmAlarmLog); sendAlarmPushUtil.sendAlarmPush(rmAlarmLog, AlarmTypeEnum.磁盘缺失.getMsg()); } } catch (NumberFormatException e) { disksToRemove.add(diskName); } } // 4. 删除已处理的磁盘记录 if (!disksToRemove.isEmpty()) { redisTemplate.opsForHash().delete(diskCountKey, disksToRemove.toArray()); } // 5. 数据入库 initialDiskInfoService.batchInsertInitialDiskInfo(disks, createTime); } /** * 内存数据入库 * @param message */ private void handleMemoryMessage(DeviceMessage message) { List memorys = JsonDataParser.parseJsonData(message.getData(), InitialMemoryInfo.class); if(!memorys.isEmpty()){ // 时间戳转换 long timestamp = memorys.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 memorys.forEach(iface -> { iface.setClientId(message.getClientId()); iface.setCreateTime(createTime); }); // 初始内存数据入库 initialMemoryInfoService.batchInsertInitialMemoryInfo(memorys); }else{ throw new RuntimeException("内存data数据为空"); } } /** * 挂载点数据入库 * @param message */ private void handleMountPointMessage(DeviceMessage message) { List mountPointInfos = JsonDataParser.parseJsonData(message.getData(), InitialMountPointInfo.class); if(!mountPointInfos.isEmpty()){ // 时间戳转换 long timestamp = mountPointInfos.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 mountPointInfos.forEach(iface -> { iface.setClientId(message.getClientId()); iface.setCreateTime(createTime); }); // 初始挂载点数据入库 initialMountPointInfoService.batchInsertInitialMountPointInfo(mountPointInfos, createTime); }else{ throw new RuntimeException("挂载点data数据为空"); } } /** * 系统其他信息 * @param message */ private void handleOtherSystemMessage(DeviceMessage message) { List otherData = JsonDataParser.parseJsonData(message.getData(), CollectDataVo.class); if(!otherData.isEmpty()){ CollectDataVo systemDataVo = otherData.get(0); if (systemDataVo != null){ try { InitialSystemOtherCollectData insertData = new InitialSystemOtherCollectData(); // 时间戳转换 long timestamp = systemDataVo.getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 insertData.setClientId(message.getClientId()); insertData.setCreateTime(createTime); String collectType = systemDataVo.getType(); String collectValue = systemDataVo.getValue(); insertData.setCollectType(collectType); insertData.setCollectValue(collectValue); // 定义需要先删后插的collectType列表 List specialCollectTypes = Arrays.asList( "kernelMaxprocCollect", "memorySizeTotalCollect", "systemBoottimeCollect", "systemCpuNum", "systemDiskSizeTotalCollect", "systemLocaltimeCollect", "systemSwArchCollect", "systemSwOsCollect", "systemUnameCollect", "systemUptimeCollect", "procNumCollect" ); // 如果collectType在特殊列表中，先删除上个时间段的记录 if (specialCollectTypes.contains(collectType)) { // 删除相同clientId、collectType和时间范围内的记录 InitialSystemOtherCollectData deleteData = new InitialSystemOtherCollectData(); deleteData.setClientId(message.getClientId()); deleteData.setCollectType(collectType); deleteData.setCreateTime(createTime); iInitialSystemOtherCollectDataService.deleteInitialSystemOtherCollectData(deleteData); } iInitialSystemOtherCollectDataService.insertInitialSystemOtherCollectData(insertData); } catch (Exception e) { log.error("解析JSON数据失败: {}, value: {}", e.getMessage(), systemDataVo.getValue()); // 可以选择保存原始数据或进行其他错误处理 } } } } /** * 监听心跳 * @param message */ private void handleHeartbeatMessage(DeviceMessage message) { List heartbeats = JsonDataParser.parseJsonData(message.getData(), InitialHeartbeatListen.class); if(!heartbeats.isEmpty()){ InitialHeartbeatListen heartbeat = heartbeats.get(0); String clientId = message.getClientId(); String version = heartbeat.getVersion(); String name = heartbeat.getName(); log.debug("处理心跳消息，客户端ID: {}, 时间: {}", clientId, heartbeat.getTimestamp()); // 使用Redis存储状态 String statusKey = HEARTBEAT_STATUS_PREFIX + clientId; String timeKey = HEARTBEAT_TIME_PREFIX + clientId; String recoveryCountKey = HEARTBEAT_RECOVERY_COUNT_PREFIX + clientId; String heartbeatCountKey = HEARTBEAT_COUNT_PREFIX + clientId; String alertKey = HEARTBEAT_ALERT_PREFIX + clientId; try { // 记录处理前状态（调试用） String prevStatus = redisTemplate.opsForValue().get(statusKey); String prevTime = redisTemplate.opsForValue().get(timeKey); String prevHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey); Boolean prevAlertStatus = redisTemplate.hasKey(alertKey); // log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}", // clientId, prevStatus, prevTime, prevHeartbeatCount, prevAlertStatus); // 原子递增心跳计数（线程安全） Long newHeartbeatCount = redisTemplate.opsForValue().increment(heartbeatCountKey); // 使用事务更新状态和时间 redisTemplate.execute(new SessionCallback() { @Override public Object execute(RedisOperations operations) throws DataAccessException { operations.multi(); // 重置丢失计数为0，设置最后心跳时间 operations.opsForValue().set(statusKey, "0"); operations.opsForValue().set(timeKey, String.valueOf(System.currentTimeMillis())); return operations.exec(); } }); // log.debug("客户端ID: {} 心跳处理完成，当前心跳次数: {}", clientId, newHeartbeatCount); // 检查是否之前有告警状态（心跳恢复检测） if (Boolean.TRUE.equals(redisTemplate.hasKey(alertKey))) { log.info("客户端ID: {} 检测到心跳恢复", clientId); // 原子递增恢复计数 Long recoveryCount = redisTemplate.opsForValue().increment(recoveryCountKey); log.debug("客户端ID: {} 恢复计数: {}", clientId, recoveryCount); if (recoveryCount >= 2) { // 达到2次恢复，清除告警状态 log.warn("客户端ID: {} 心跳恢复达到{}次，清除告警状态", clientId, recoveryCount); insertHeartbeatLog(clientId, "2", "心跳恢复,设备在线状态改为在线"); // 清理告警相关key redisTemplate.delete(alertKey); redisTemplate.delete(recoveryCountKey); // 修改资源状态为在线 updateResourceStatus(clientId, "1"); log.info("客户端ID: {} 告警状态已清除", clientId); } else { // 未达到2次，只记录恢复次数 log.info("客户端ID: {} 心跳恢复第{}次", clientId, recoveryCount); } } // 只有达到3次心跳才执行数据库操作 if (newHeartbeatCount >= 3) { // log.debug("客户端ID: {} 达到{}次心跳，开始执行数据库操作", clientId, newHeartbeatCount); // 添加逻辑节点标识 RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote(); updateData.setClientId(message.getClientId()); updateData.setLogicalNodeId(heartbeat.getLogicalNode()); updateData.setHardwareSn(heartbeat.getSn()); updateData.setOnlineStatus("1"); updateData.setAgentVersion(version); updateData.setOnboardTime(DateUtils.getNowDate()); remoteRevenueConfigService.innerUpdateRegist(updateData, SecurityConstants.INNER); // agent更新结果存储 RmAgentManagement query = new RmAgentManagement(); query.setClientId(clientId); List agentManagements = rmAgentManagementService.selectRmAgentManagementList(query); if(!agentManagements.isEmpty()){ RmAgentManagement rmAgentManagement = agentManagements.get(0); if(!StringUtils.equals(rmAgentManagement.getAgentVersion(), version)){ // 存储更新结果 RmAgentManagement updateResultData = new RmAgentManagement(); updateResultData.setId(rmAgentManagement.getId()); updateResultData.setAgentVersion(version); updateResultData.setLastUpdateTime(DateUtils.getNowDate()); updateResultData.setLastUpdateResult("1"); rmAgentManagementService.updateRmAgentManagementByHardwareSn(updateResultData); } } // 交换机在线状态改为在线（只在第3次心跳时执行） if(newHeartbeatCount == 3){ RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote(); rmSwitchManagementRemote.setClientId(clientId); R> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER); if(rmSwitchManagementRemoteListR != null && rmSwitchManagementRemoteListR.getData()!=null && !rmSwitchManagementRemoteListR.getData().isEmpty()){ RmSwitchManagementRemote switchUpdate = new RmSwitchManagementRemote(); switchUpdate.setClientId(clientId); switchUpdate.setOnlineStatus("1"); remoteRevenueConfigService.updateSwitchMsgByClientId(switchUpdate, SecurityConstants.INNER); log.info("客户端ID: {} 交换机状态更新完成", clientId); } else { log.debug("客户端ID: {} 未找到对应的交换机信息", clientId); } } } else { log.debug("客户端ID: {} 当前心跳次数: {}，未达到3次，跳过数据库操作", clientId, newHeartbeatCount); } // 记录处理后状态（调试用） String currentStatus = redisTemplate.opsForValue().get(statusKey); String currentTime = redisTemplate.opsForValue().get(timeKey); String currentHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey); Boolean currentAlertStatus = redisTemplate.hasKey(alertKey); // log.debug("客户端ID: {} 处理后状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}", // clientId, currentStatus, currentTime, currentHeartbeatCount, currentAlertStatus); } catch (Exception e) { log.error("处理心跳消息异常, clientId: {}", clientId, e); } } } // 添加一个定时任务方法，定期检查心跳状态 @Scheduled(fixedRate = 60000) // 每60s检查一次 public void checkHeartbeatStatus() { long currentTime = System.currentTimeMillis(); log.debug("开始心跳状态检查，当前时间: {}", currentTime); // 获取所有客户端时间键0 Set timeKeys = redisTemplate.keys(HEARTBEAT_TIME_PREFIX + "*"); if (timeKeys == null) { log.debug("未找到任何心跳时间键"); return; } log.debug("找到 {} 个客户端需要检查", timeKeys.size()); for (String timeKey : timeKeys) { String clientId = timeKey.substring(HEARTBEAT_TIME_PREFIX.length()); String statusKey = HEARTBEAT_STATUS_PREFIX + clientId; String alertKey = HEARTBEAT_ALERT_PREFIX + clientId; try { // 检查是否已经存在告警 String existingAlert = redisTemplate.opsForValue().get(alertKey); if ("1".equals(existingAlert)) { log.debug("客户端ID: {} 已有告警，跳过检查", clientId); continue; // 如果已有告警，跳过处理 } String lastTimeStr = redisTemplate.opsForValue().get(timeKey); if (lastTimeStr == null) { log.debug("客户端ID: {} 时间键为空，跳过", clientId); continue; } long lastHeartbeatTime = Long.parseLong(lastTimeStr); long timeDiff = currentTime - lastHeartbeatTime; log.debug("客户端ID: {} 最后心跳: {}, 时间差: {}ms, 超时阈值: {}ms", clientId, lastHeartbeatTime, timeDiff, HEARTBEAT_TIMEOUT); if (timeDiff > HEARTBEAT_TIMEOUT) { // 心跳超时处理 - 使用原子操作增加计数 Long lostCount = redisTemplate.opsForValue().increment(statusKey); if (lostCount == 1) { // 确保第一次增加时值为1 redisTemplate.opsForValue().set(statusKey, "1"); lostCount = 1L; } log.warn("客户端ID: {} 心跳超时，连续次数: {}, 时间差: {}ms", clientId, lostCount, timeDiff); if (lostCount >= 3) { log.warn("客户端ID: {} 连续三次心跳丢失，触发告警", clientId); insertHeartbeatLog(clientId, "3", "连续三次心跳丢失"); // 告警 insertAlarmRecords(clientId); redisTemplate.opsForValue().set(HEARTBEAT_ALERT_PREFIX + clientId, "1"); // 设置告警后删除timeKey和statusKey redisTemplate.delete(timeKey); redisTemplate.delete(statusKey); log.info("客户端ID: {} 已设置告警并清理心跳记录", clientId); // frpc状态改为未运行 RmFrpcConfigManage updateFrp = new RmFrpcConfigManage(); updateFrp.setClientId(clientId); updateFrp.setFrpcStatus("0"); rmFrpcConfigManageService.updateRmFrpcConfigManage(updateFrp); // 修改资源状态 updateResourceStatus(clientId, "0"); } } else { // 如果心跳正常，重置丢失次数 String currentStatus = redisTemplate.opsForValue().get(statusKey); if (!"0".equals(currentStatus)) { redisTemplate.opsForValue().set(statusKey, "0"); log.debug("客户端ID: {} 心跳正常，重置丢失次数从 {} 到 0", clientId, currentStatus); } else { log.debug("客户端ID: {} 心跳正常，状态已是0", clientId); } } } catch (Exception e) { log.error("检查心跳状态异常, clientId: {}", clientId, e); } } log.debug("心跳状态检查完成"); } // 更新资源状态的公共方法 private void updateResourceStatus(String clientId, String status) { log.info("开启更新资源状态========"); RmResourceRegistrationRemote query = new RmResourceRegistrationRemote(); query.setClientId(clientId); R registerMsgR = remoteRevenueConfigService.getListByHardwareSn(query, SecurityConstants.INNER); if(registerMsgR != null && registerMsgR.getData() != null){ RmResourceRegistrationRemote rmResourceRegistrationRemote = new RmResourceRegistrationRemote(); rmResourceRegistrationRemote.setOnlineStatus(status); rmResourceRegistrationRemote.setClientId(clientId); remoteRevenueConfigService.updateStatusByResource(rmResourceRegistrationRemote, SecurityConstants.INNER); } RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote(); rmSwitchManagementRemote.setClientId(clientId); R> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER); if(rmSwitchManagementRemoteListR != null && rmSwitchManagementRemoteListR.getData()!=null && !rmSwitchManagementRemoteListR.getData().isEmpty()){ RmSwitchManagementRemote switchUpdate = new RmSwitchManagementRemote(); switchUpdate.setClientId(clientId); switchUpdate.setOnlineStatus(status); remoteRevenueConfigService.updateSwitchMsgByClientId(switchUpdate, SecurityConstants.INNER); } } // 插入心跳日志到数据库 private void insertHeartbeatLog(String machineId, String status, String remark) { try { InitialHeartbeatListenLog listenLog = new InitialHeartbeatListenLog(); listenLog.setClientId(machineId); listenLog.setStatus(status); // 0-离线 1-在线 2-恢复 3-三次丢失 listenLog.setRemark(remark); listenLog.setCreateTime(new Date()); // 调用DAO或Service插入日志 initialHeartbeatListenLog.insertInitialHeartbeatListenLog(listenLog); log.info("已记录心跳日志，客户端ID: {}, 状态: {}", machineId, status); } catch (Exception e) { log.error("插入心跳日志失败", e); } } /** * 添加告警日志并推送消息到企业微信 * @param clientId */ private void insertAlarmRecords(String clientId) { // 查询clientId是否为交换机唯一标识 boolean isSwitch = false; String switchName = ""; RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote(); rmSwitchManagementRemote.setClientId(clientId); R> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER); if (rmSwitchManagementRemoteListR != null && rmSwitchManagementRemoteListR.getData() != null && !rmSwitchManagementRemoteListR.getData().isEmpty()) { isSwitch = true; switchName = rmSwitchManagementRemoteListR.getData().get(0).getSwitchName(); } // 创建告警日志记录 RmAlarmLog rmAlarmLog = createAlarmLog(clientId, isSwitch, switchName); // 插入告警日志 rmAlarmLogService.insertRmAlarmLog(rmAlarmLog); // 发送告警推送 sendAlarmPush(rmAlarmLog, isSwitch); } /** * 创建告警日志记录 */ private RmAlarmLog createAlarmLog(String clientId, boolean isSwitch, String switchName) { RmAlarmLog rmAlarmLog = new RmAlarmLog(); String alarmContent = clientId + "下线"; if (isSwitch) { rmAlarmLog.setClientId(switchName); rmAlarmLog.setAlarmType("2"); alarmContent = switchName + "下线"; } else { // 查询服务器信息 RmResourceRegistrationRemote query = new RmResourceRegistrationRemote(); query.setClientId(clientId); R registerMsgR = remoteRevenueConfigService.getListByHardwareSn(query, SecurityConstants.INNER); if(registerMsgR != null && registerMsgR.getData() != null){ RmResourceRegistrationRemote serverMsg = registerMsgR.getData(); rmAlarmLog.setMgmPublicIp(serverMsg.getIp1PublicIp()); rmAlarmLog.setBusinessName(serverMsg.getBusinessName()); rmAlarmLog.setRemark(serverMsg.getRemark()); } rmAlarmLog.setClientId(clientId); rmAlarmLog.setAlarmType("1"); } rmAlarmLog.setAlarmContent(alarmContent); rmAlarmLog.setAlarmTime(DateUtils.getNowDate()); return rmAlarmLog; } /** * 发送告警推送 */ private void sendAlarmPush(RmAlarmLog rmAlarmLog, boolean isSwitch) { String alarmTypeCode = isSwitch ? AlarmTypeEnum.交换机下线.getCode() : AlarmTypeEnum.服务器下线.getCode(); String alarmTypeMsg = isSwitch ? AlarmTypeEnum.交换机下线.getMsg() : AlarmTypeEnum.服务器下线.getMsg(); rmAlarmLog.setAlarmType(alarmTypeCode); sendAlarmPushUtil.sendAlarmPush(rmAlarmLog, alarmTypeMsg); } /** * 应答信息 * @param message */ private RspVo handleResponseMessage(DeviceMessage message) { List rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class); if (!rspVoList.isEmpty()) { RspVo rsp = rspVoList.get(0); log.info("应答信息：{}",rsp); return rsp; } return null; } /** * 脚本策略应答信息 * @param message */ private void handleScriptRspMessage(DeviceMessage message) { List rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class); if (!rspVoList.isEmpty()) { RspVo rsp = rspVoList.get(0); // 时间戳转换 long timestamp = rsp.getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 if(rsp.getResCode() == 1){ if(rsp.getResult() != null){ List resultVos = JsonDataParser.parseJsonData(rsp.getResult(), RspResultVo.class); RspResultVo rspResultVo = resultVos.get(0); JSONObject jsonObject = JSONObject.parseObject(rsp.getResult()); String resOut = jsonObject.getString("resOut"); // 查询执行的脚本id RmDeploymentPolicy deploymentPolicy = rmDeploymentPolicyService.selectRmDeploymentPolicyById(Long.valueOf(rspResultVo.getScriptId())); Long businessScriptId = deploymentPolicy.getScriptId(); // 判断失败返回结果是否包含脚本执行失败关键字 R businessScriptMsgR = remoteRevenueConfigService.getBusinessScriptMsgByScriptId(businessScriptId, SecurityConstants.INNER); // 构建脚本执行结果实体类 RmResourceRemote insertData = new RmResourceRemote(); insertData.setClientId(message.getClientId()); insertData.setScriptId(Long.valueOf(rspResultVo.getScriptId())); if(businessScriptMsgR != null && businessScriptMsgR.getData() != null){ EpsBusinessScriptRemote businessScriptMsg = businessScriptMsgR.getData(); String failedKeywords = businessScriptMsg.getFailedKeywords(); if(failedKeywords != null && resOut.contains(failedKeywords)){ insertData.setResultFlag(0); }else { insertData.setResultFlag(1); } }else{ insertData.setResultFlag(1); } insertData.setDescription(rsp.getResult()); insertData.setCreateTime(createTime); // 执行插入sql rmResourceRemoteService.insertRmResourceRemote(insertData); log.info("脚本执行结果入库成功：{}",rsp); } }else{ // 构建脚本执行结果实体类 RmResourceRemote insertData = new RmResourceRemote(); insertData.setClientId(message.getClientId()); insertData.setResultFlag(0); if(rsp.getResult() != null){ List resultVos = JsonDataParser.parseJsonData(rsp.getResult(), RspResultVo.class); RspResultVo rspResultVo = resultVos.get(0); insertData.setDescription(rsp.getResult()); insertData.setScriptId(Long.valueOf(rspResultVo.getScriptId())); } insertData.setCreateTime(createTime); // 执行插入sql rmResourceRemoteService.insertRmResourceRemote(insertData); log.error("脚本执行失败:{}", rsp); } } } /** * 公共方法：处理网卡信息 * @param registerMsg 注册消息 * @param clientId 客户端ID * @param isRegister 是否为注册消息 */ private void processNetworkInterfaces(RegisterMsgVo registerMsg, String clientId, boolean isRegister) { // 时间戳转换 long timestamp = registerMsg.getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); List networkInfoList = registerMsg.getNetworkInfo(); if (networkInfoList.isEmpty()) { return; } boolean isSingleInterface = networkInfoList.size() == 1; // 查询数据库中当前的网卡数量 RmNetworkInterface countQuery = new RmNetworkInterface(); countQuery.setClientId(clientId); countQuery.setNewFlag(1); List currentInterfaces = rmNetworkInterfaceService.selectRmNetworkInterfaceList(countQuery); // 构建数据库中当前网卡MAC地址的集合 Set currentMacSet = currentInterfaces.stream() .map(RmNetworkInterface::getMacAddress) .filter(Objects::nonNull) .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); } // 获取当前客户端的所有子网卡 RmNetworkInterfaceChild childQuery = new RmNetworkInterfaceChild(); childQuery.setClientId(clientId); List existingChildren = rmNetworkInterfaceChildService.selectRmNetworkInterfaceChildList(childQuery); if(existingChildren == null){ existingChildren = new ArrayList<>(); } // 构建数据库中所有子网卡的key集合 (clientId_mac_name) Map existingChildMap = new HashMap<>(); for (RmNetworkInterfaceChild child : existingChildren) { String key = child.getClientId() + "_" + child.getMacAddress() + "_" + child.getInterfaceName(); existingChildMap.put(key, child); } // 构建本次接收到的所有子网卡key集合 Set newChildKeys = new HashSet<>(); for (NetworkInfo networkInfo : networkInfoList) { List childList = networkInfo.getSubInterfaces(); // 查询该网卡信息是否存在 RmNetworkInterface queryParam = new RmNetworkInterface(); queryParam.setClientId(clientId); queryParam.setMacAddress(networkInfo.getMac()); queryParam.setNewFlag(1); List exits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(queryParam); if (exits.isEmpty()) { // 新增网卡信息 RmNetworkInterface insertData = new RmNetworkInterface(); setNetworkInterfaceData(insertData, networkInfo, clientId); insertData.setCreateTime(createTime); // 设置bindIp if (isSingleInterface) { insertData.setBindIp("3"); } rmNetworkInterfaceService.insertRmNetworkInterface(insertData); // 处理子网卡 if (childList != null && !childList.isEmpty()) { for (NetworkInfo info : childList) { RmNetworkInterfaceChild insertChild = new RmNetworkInterfaceChild(); setNetworkInterfaceChildData(insertChild, info, clientId, networkInfo.getName()); insertChild.setUpdateTime(DateUtils.getNowDate()); // 设置bindIp if (isSingleInterface) { insertChild.setBindIp("3"); } // 构建子网卡唯一key String childKey = clientId + "_" + info.getMac() + "_" + info.getName(); newChildKeys.add(childKey); // 插入或更新子网卡 rmNetworkInterfaceChildService.insertRmNetworkInterfaceChild(insertChild); } } // 如果网卡数量有变动，需要更新网卡绑定状态 if (!isRegister && interfaceCountChanged && !isSingleInterface) { RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote(); updateParam.setClientId(clientId); updateParam.setMultiPublicIpStatus("0"); remoteRevenueConfigService.updateStatusByResource(updateParam, SecurityConstants.INNER); } } else { // 更新网卡信息 RmNetworkInterface oldInterfaceMsg = exits.get(0); // 处理子网卡 if (childList != null && !childList.isEmpty()) { for (NetworkInfo info : childList) { RmNetworkInterfaceChild insertChild = new RmNetworkInterfaceChild(); setNetworkInterfaceChildData(insertChild, info, clientId, networkInfo.getName()); insertChild.setUpdateTime(DateUtils.getNowDate()); // 构建子网卡唯一key String childKey = clientId + "_" + info.getMac() + "_" + info.getName(); newChildKeys.add(childKey); // 插入或更新子网卡 rmNetworkInterfaceChildService.insertRmNetworkInterfaceChild(insertChild); } } // 判断是否需要创建新记录 boolean needCreateNew = (!StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName()) || !StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway())) && StringUtils.equals(networkInfo.getMac(), oldInterfaceMsg.getMacAddress()); // 更新现有记录 updateNetworkInterface(networkInfo, clientId, oldInterfaceMsg, isSingleInterface, isRegister); } } // 删除数据库中不存在于本次接收数据中的子网卡 // 计算需要删除的子网卡：数据库中有的，但本次没有接收到的 Set childKeysToDelete = new HashSet<>(existingChildMap.keySet()); childKeysToDelete.removeAll(newChildKeys); if (!childKeysToDelete.isEmpty() && !isRegister) { for (String keyToDelete : childKeysToDelete) { String[] parts = keyToDelete.split("_", 3); if (parts.length == 3) { String clientIdPart = parts[0]; String mac = parts[1]; String name = parts[2]; // 只删除当前客户端的子网卡 if (clientId.equals(clientIdPart)) { deleteChildNetworkInterfaces(clientId, mac, name); } } } } } /** * 删除过时的网卡信息 * @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); } /** * 删除过时的子网卡信息 * @param clientId * @param name * @param mac */ private void deleteChildNetworkInterfaces(String clientId, String mac, String name) { RmNetworkInterfaceChild query = new RmNetworkInterfaceChild(); query.setClientId(clientId); query.setMacAddress(mac); query.setInterfaceName(name); query.setNewFlag(1); List oldExits = rmNetworkInterfaceChildService.selectRmNetworkInterfaceChildList(query); if(!oldExits.isEmpty()) { oldExits.forEach(oldMsg -> { rmNetworkInterfaceChildService.deleteRmNetworkInterfaceChildById(oldMsg.getId()); }); } // 记录删除操作日志 log.info("删除客户端 {} 的过时子网卡信息，MAC地址: {},网卡名称：{}", clientId, mac, name); } /** * 设置网卡信息公共字段 */ private void setNetworkInterfaceData(RmNetworkInterface networkInterface, NetworkInfo networkInfo, String clientId) { networkInterface.setClientId(clientId); networkInterface.setIsp(networkInfo.getCarrier()); networkInterface.setCity(networkInfo.getCity()); networkInterface.setGateway(networkInfo.getGateway()); networkInterface.setInterfaceName(networkInfo.getName()); networkInterface.setIpv4Address(networkInfo.getIpv4()); networkInterface.setIpv6Address(networkInfo.getIpv6()); networkInterface.setMacAddress(networkInfo.getMac()); networkInterface.setProvince(networkInfo.getProvince()); networkInterface.setPublicIp(networkInfo.getPublicIp()); networkInterface.setInterfaceType(networkInfo.getType()); } private void setNetworkInterfaceChildData(RmNetworkInterfaceChild networkInterface, NetworkInfo networkInfo, String clientId, String parentName) { networkInterface.setClientId(clientId); networkInterface.setParentInterface(parentName); networkInterface.setIsp(networkInfo.getCarrier()); networkInterface.setCity(networkInfo.getCity()); networkInterface.setGateway(networkInfo.getGateway()); networkInterface.setInterfaceName(networkInfo.getName()); networkInterface.setIpv4Address(networkInfo.getIpv4()); networkInterface.setIpv6Address(networkInfo.getIpv6()); networkInterface.setMacAddress(networkInfo.getMac()); networkInterface.setProvince(networkInfo.getProvince()); networkInterface.setPublicIp(networkInfo.getPublicIp()); networkInterface.setInterfaceType(networkInfo.getType()); networkInterface.setStatus(networkInfo.getStatus()); networkInterface.setNetCreateTime(networkInfo.getNetCreateTime()); } /** * 清理旧记录 */ private void cleanOldRecords(String clientId, String macAddress) { RmNetworkInterface query = new RmNetworkInterface(); query.setClientId(clientId); query.setMacAddress(macAddress); query.setNewFlag(999); List oldExits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(query); if(!oldExits.isEmpty()) { oldExits.forEach(oldMsg -> { rmNetworkInterfaceService.deleteRmNetworkInterfaceById(oldMsg.getId()); }); } } /** * 清理旧记录 */ private void cleanChildOldRecords(String clientId, String macAddress) { RmNetworkInterfaceChild childQuery = new RmNetworkInterfaceChild(); childQuery.setClientId(clientId); childQuery.setMacAddress(macAddress); List oldChildExits = rmNetworkInterfaceChildService.selectRmNetworkInterfaceChildList(childQuery); if(!oldChildExits.isEmpty()) { oldChildExits.forEach(oldMsg -> { rmNetworkInterfaceChildService.deleteRmNetworkInterfaceChildById(oldMsg.getId()); }); } } /** * 更新网卡信息 */ private void updateNetworkInterface(NetworkInfo networkInfo, String clientId, RmNetworkInterface oldInterfaceMsg, boolean isSingleInterface, boolean isRegister) { RmNetworkInterface updateData = new RmNetworkInterface(); BeanUtils.copyProperties(oldInterfaceMsg, updateData); boolean needUpdate = false; // 逐个字段比较是否需要更新 if (networkInfo.getCity() != null && !StringUtils.equals(networkInfo.getCity(), oldInterfaceMsg.getCity())) { updateData.setCity(networkInfo.getCity()); needUpdate = true; } if (!StringUtils.equals(networkInfo.getIpv4(), oldInterfaceMsg.getIpv4Address())) { updateData.setIpv4Address(networkInfo.getIpv4()); needUpdate = true; } if (!StringUtils.equals(networkInfo.getIpv6(), oldInterfaceMsg.getIpv6Address())) { updateData.setIpv6Address(networkInfo.getIpv6()); needUpdate = true; } if (networkInfo.getProvince() != null && !StringUtils.equals(networkInfo.getProvince(), oldInterfaceMsg.getProvince())) { updateData.setProvince(networkInfo.getProvince()); needUpdate = true; } if (networkInfo.getPublicIp() != null && !StringUtils.equals(networkInfo.getPublicIp(), oldInterfaceMsg.getPublicIp())) { updateData.setPublicIp(networkInfo.getPublicIp()); needUpdate = true; } if (networkInfo.getCarrier() != null && !StringUtils.equals(networkInfo.getCarrier(), oldInterfaceMsg.getIsp())) { updateData.setIsp(networkInfo.getCarrier()); needUpdate = true; } if (!StringUtils.equals(networkInfo.getType(), oldInterfaceMsg.getInterfaceType())) { updateData.setInterfaceType(networkInfo.getType()); needUpdate = true; } if (!StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway())) { updateData.setGateway(networkInfo.getGateway()); needUpdate = true; } if (!StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())) { if(networkInfo.getName() != null){ // 添加业务变更记录 EpsMethodChangeRecordVo recordAddData = new EpsMethodChangeRecordVo(); recordAddData.setClientId(clientId); recordAddData.setTrafficPort(networkInfo.getName()); recordAddData.setUpdateTime(DateUtils.getNowDate()); recordAddData.setCreateTime(DateUtils.getNowDate()); recordAddData.setUpdateBy(SecurityUtils.getUsername()); recordAddData.setCreatBy(SecurityUtils.getUsername()); StringBuilder content = new StringBuilder(); content.append("流量网口设置为").append(networkInfo.getName()); recordAddData.setChangeContent(content.toString()); rmNetworkInterfaceService.addTrafficPortChangeRecord(recordAddData); updateData.setInterfaceName(networkInfo.getName()); needUpdate = true; } } // 只有有字段变化时才执行更新 if (needUpdate) { updateData.setClientId(clientId); updateData.setMacAddress(oldInterfaceMsg.getMacAddress()); rmNetworkInterfaceService.updateNetMsgByMac(updateData); } } public String checkIPVersion(String ipAddress) { // 检查是否为 IPv4 地址 if (ipAddress.contains(".")) { // 简单的格式验证：IPv4 应该有 4 个部分，每个部分用点分隔 String[] parts = ipAddress.split("\\."); if (parts.length == 4) { try { // 验证每个部分是否在 0-255 范围内 for (String part : parts) { int num = Integer.parseInt(part); if (num < 0 || num > 255) { return "Invalid IP Address"; } } return "IPv4"; } catch (NumberFormatException e) { return "Invalid IP Address"; } } } // 检查是否为 IPv6 地址 if (ipAddress.contains(":")) { // 简单的格式验证：IPv6 应该包含冒号分隔的十六进制数 String[] parts = ipAddress.split(":"); if (parts.length >= 3 && parts.length <= 8) { try { // 验证每个部分是否为有效的十六进制数 for (String part : parts) { if (!part.isEmpty()) { // 允许空的部分（表示连续的零） Integer.parseInt(part, 16); } } return "IPv6"; } catch (NumberFormatException e) { return "Invalid IP Address"; } } } return "Invalid IP Address"; } /** * 查询IP地址归属地信息，返回运营商和省份 */ private Map queryIpLocation(String ip) { String apiUrl = "http://172.16.15.103:10000/?ip=" + ip; HttpGet httpGet = new HttpGet(apiUrl); try (CloseableHttpResponse response = httpClient.execute(httpGet)) { if (response.getStatusLine().getStatusCode() == 200) { String jsonResponse = EntityUtils.toString(response.getEntity(), "UTF-8"); ObjectMapper mapper = new ObjectMapper(); Map resultMap = mapper.readValue(jsonResponse, new TypeReference>() {}); return extractLocationInfo(resultMap); } } catch (Exception e) { System.err.println("查询IP归属地时发生异常: " + e.getMessage()); e.printStackTrace(); } return null; } /** * 从JSON Map中提取运营商和省份 */ private Map extractLocationInfo(Map resultMap) { Map info = new HashMap<>(); // 提取运营商（从as.info获取中文名称） if (resultMap.containsKey("as")) { Object asObj = resultMap.get("as"); if (asObj instanceof Map) { Map asMap = (Map) asObj; if (asMap.containsKey("info")) { info.put("operator", asMap.get("info").toString()); } } } // 提取省份 if (resultMap.containsKey("regions_short")) { Object regionsObj = resultMap.get("regions_short"); if (regionsObj instanceof List && ((List) regionsObj).size() > 0) { info.put("province", ((List) regionsObj).get(0).toString()); } } else if (resultMap.containsKey("regions")) { Object regionsObj = resultMap.get("regions"); if (regionsObj instanceof List && ((List) regionsObj).size() > 0) { String province = ((List) regionsObj).get(0).toString(); // 处理"浙江省" -> "浙江" if (province.endsWith("省")) { province = province.substring(0, province.length() - 1); } info.put("province", province); } } return info; } }