package com.tongran.rocketmq.handler; import com.alibaba.fastjson.JSONObject; 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.PushMethodEnum; import com.tongran.rocketmq.service.*; import com.tongran.rocketmq.utils.DataProcessUtil; import com.tongran.rocketmq.utils.JsonDataParser; import com.tongran.rocketmq.utils.WeChatWorkBot; import com.tongran.system.api.RemoteRevenueConfigService; import com.tongran.system.api.domain.*; import lombok.extern.slf4j.Slf4j; 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.stereotype.Component; import javax.annotation.PostConstruct; import java.math.BigDecimal; import java.math.RoundingMode; import java.util.*; import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; /** * 设备消息处理器 */ @Slf4j @Component @EnableScheduling public class MessageHandler { private final Map> messageHandlers = new HashMap<>(); // 心跳状态 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分钟超时 @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 IRmAlarmLogService rmAlarmLogService; @Autowired private IRmAlarmPushConfigService rmAlarmPushConfigService; @Autowired private IRmFrpcConfigManageService rmFrpcConfigManageService; @Autowired private IRmPppoeConfigSubService rmPppoeConfigSubService; /** * 初始化处理器映射 */ @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::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); } 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()); // 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数据为空"); } } 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); }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); }); // 初始CPU数据入库 initialCpuInfoService.batchInsertInitialCpuInfo(cpus); }else{ throw new RuntimeException("CPUdata数据为空"); } } /** * 磁盘数据入库 * @param message */ private void handleDiskMessage(DeviceMessage message) { List disks = JsonDataParser.parseJsonData(message.getData(), InitialDiskInfo.class); // 时间戳转换 long timestamp = disks.get(0).getTimestamp(); long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 if(!disks.isEmpty()){ disks.forEach(iface -> { iface.setClientId(message.getClientId()); iface.setCreateTime(createTime); }); // 初始磁盘数据入库 initialDiskInfoService.batchInsertInitialDiskInfo(disks); }else{ throw new RuntimeException("磁盘data数据为空"); } } /** * 内存数据入库 * @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); }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); // 获取所有客户端时间键 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 { // 查询管理网公网ip RmNetworkInterface rmNetworkInterface = new RmNetworkInterface(); rmNetworkInterface.setClientId(clientId); rmNetworkInterface.setNewFlag(1); List interfaceList = rmNetworkInterfaceService.selectRmNetworkInterfaceList(rmNetworkInterface); if (interfaceList != null && !interfaceList.isEmpty()) { interfaceList.stream() .filter(info -> "2".equals(info.getBindIp()) || "3".equals(info.getBindIp())) .findFirst() .ifPresent(networkInterface -> { rmAlarmLog.setMgmPublicIp(networkInterface.getPublicIp()); }); } // 查询业务名称 RmResourceRegistrationRemote query = new RmResourceRegistrationRemote(); query.setClientId(clientId); R registrationR = remoteRevenueConfigService.getListByHardwareSn(query, SecurityConstants.INNER); if(registrationR != null){ RmResourceRegistrationRemote registrationMsg = registrationR.getData(); if(registrationMsg != null){ rmAlarmLog.setBusinessName(registrationMsg.getBusinessName()); } } 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(); RmAlarmPushConfig rmAlarmPushConfig = new RmAlarmPushConfig(); rmAlarmPushConfig.setPushMethod(PushMethodEnum.企业微信.getCode()); rmAlarmPushConfig.setPushAlarmTypes(alarmTypeCode); List alarmConfigList = rmAlarmPushConfigService.selectRmAlarmPushConfigList(rmAlarmPushConfig); RmAlarmLog updateData = new RmAlarmLog(); updateData.setId(rmAlarmLog.getId()); if (alarmConfigList != null && !alarmConfigList.isEmpty()) { try { for (RmAlarmPushConfig alarmPushConfig : alarmConfigList) { String contentTemplate = alarmPushConfig.getMessageContent(); String webhookUrl = alarmPushConfig.getPushAddress(); Map alarmMap = new HashMap<>(); alarmMap.put("告警时间", rmAlarmLog.getAlarmTime()); alarmMap.put("IP", rmAlarmLog.getMgmPublicIp()); alarmMap.put("告警类型", alarmTypeMsg); alarmMap.put("告警设备", rmAlarmLog.getClientId()); alarmMap.put("告警内容", rmAlarmLog.getAlarmContent()); alarmMap.put("业务名称", rmAlarmLog.getBusinessName()); if (alarmPushConfig.getContactPhones() != null) { String[] phones = alarmPushConfig.getContactPhones().split(","); WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap, rmAlarmLog.getAlarmContent(), phones, false); } else { WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap); } } updateData.setPushFlag(1L); } catch (Exception e){ updateData.setPushFlag(0L); log.error("消息推送失败：{}", e.getMessage()); } }else{ updateData.setPushFlag(2L); } rmAlarmLogService.updateRmAlarmLog(updateData); } /** * 应答信息 * @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) .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(); // 查询该网卡信息是否存在 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()){ rmNetworkInterfaceChildService.deleteRmNetworkInterfaceChildByClientId(clientId); for (NetworkInfo info : childList) { RmNetworkInterfaceChild insertChild = new RmNetworkInterfaceChild(); setNetworkInterfaceChildData(insertChild, info, clientId, networkInfo.getName()); // 设置bindIp if (isSingleInterface) { insertChild.setBindIp("3"); } 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); // cleanChildOldRecords(clientId, networkInfo.getMac()); // 判断是否需要创建新记录 boolean needCreateNew = (!StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName()) || !StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway())) && StringUtils.equals(networkInfo.getMac(), oldInterfaceMsg.getMacAddress()); if(childList != null && !childList.isEmpty()){ rmNetworkInterfaceChildService.deleteRmNetworkInterfaceChildByClientId(clientId); for (NetworkInfo info : childList) { RmNetworkInterfaceChild insertChild = new RmNetworkInterfaceChild(); setNetworkInterfaceChildData(insertChild, info, clientId, networkInfo.getName()); rmNetworkInterfaceChildService.insertRmNetworkInterfaceChild(insertChild); } } // 更新现有记录 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); } /** * 设置网卡信息公共字段 */ 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()); } /** * 清理旧记录 */ 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 (!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 (!StringUtils.equals(networkInfo.getProvince(), oldInterfaceMsg.getProvince())) { updateData.setProvince(networkInfo.getProvince()); needUpdate = true; } if (!StringUtils.equals(networkInfo.getPublicIp(), oldInterfaceMsg.getPublicIp())) { updateData.setPublicIp(networkInfo.getPublicIp()); needUpdate = true; } if (!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.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); } } }