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

agent2.0版本数据适配，开发脚本策略接口

Browse Source
This commit is contained in:
gaoyutao
2025-09-23 18:21:05 +08:00
parent a9270027f4
commit d5f232e295
62 changed files with 5286 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/DeviceMessageHandler.java
+283 -20
View File
@@ -1,13 +1,15 @@
package com.ruoyi.rocketmq.handler;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.common.core.constant.SecurityConstants;
import com.ruoyi.common.core.domain.R;
import com.ruoyi.common.core.enums.MsgEnum;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.common.core.utils.StringUtils;
import com.ruoyi.rocketmq.domain.*;
import com.ruoyi.rocketmq.domain.vo.CollectDataVo;
import com.ruoyi.rocketmq.domain.vo.RspVo;
import com.ruoyi.rocketmq.domain.vo.SwitchDataVo;
import com.ruoyi.rocketmq.service.*;
import com.ruoyi.rocketmq.utils.JsonDataParser;
import com.ruoyi.system.api.RemoteRevenueConfigService;
@@ -71,6 +73,20 @@ public class DeviceMessageHandler {
private IInitialSwitchInfoTempService initialSwitchInfoTempService;
@Autowired
private IInitialHeartbeatListenLogService initialHeartbeatListenLog;
@Autowired
private IInitialSwitchPowerSupplyService initialSwitchPowerSupplyService;
@Autowired
private IInitialSwitchFanInfoService initialSwitchFanInfoService;
@Autowired
private IInitialSwitchMpuInfoService initialSwitchMpuInfoService;
@Autowired
private IInitialSwitchOpticalModuleService initialSwitchOpticalModuleService;
@Autowired
private IInitialSwitchOtherCollectDataService insertInitialSwitchOtherInfo;
@Autowired
private IInitialSystemOtherCollectDataService iInitialSystemOtherCollectDataService;
// 在类中添加
private static final ObjectMapper objectMapper = new ObjectMapper();
/**
@@ -90,13 +106,13 @@ public class DeviceMessageHandler {
registerHandler(MsgEnum.Agent版本更新应答.getValue(), this::handleResponseMessage);
// 其他类型消息可以单独注册处理器
// 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::handleSystemMessage);
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::handleSwitchDataMessage);
registerHandler(MsgEnum.心跳上报.getValue(), this::handleHeartbeatMessage);
}
@@ -133,7 +149,7 @@ public class DeviceMessageHandler {
if(!interfaces.isEmpty()){
// 时间戳转换
long timestamp = interfaces.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime);
InitialBandwidthTraffic data = new InitialBandwidthTraffic();
@@ -163,7 +179,7 @@ public class DeviceMessageHandler {
if(!dockers.isEmpty()){
// 时间戳转换
long timestamp = dockers.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
dockers.forEach(iface -> {
iface.setClientId(message.getClientId());
@@ -183,7 +199,7 @@ public class DeviceMessageHandler {
List<InitialCpuInfo> cpus = JsonDataParser.parseJsonData(message.getData(),InitialCpuInfo.class);
// 时间戳转换
long timestamp = cpus.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
if(!cpus.isEmpty()){
cpus.forEach(iface -> {
@@ -204,7 +220,7 @@ public class DeviceMessageHandler {
List<InitialDiskInfo> disks = JsonDataParser.parseJsonData(message.getData(), InitialDiskInfo.class);
// 时间戳转换
long timestamp = disks.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
if(!disks.isEmpty()){
disks.forEach(iface -> {
@@ -226,7 +242,7 @@ public class DeviceMessageHandler {
if(!memorys.isEmpty()){
// 时间戳转换
long timestamp = memorys.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
memorys.forEach(iface -> {
iface.setClientId(message.getClientId());
@@ -247,7 +263,7 @@ public class DeviceMessageHandler {
if(!mountPointInfos.isEmpty()){
// 时间戳转换
long timestamp = mountPointInfos.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
mountPointInfos.forEach(iface -> {
iface.setClientId(message.getClientId());
@@ -264,29 +280,230 @@ public class DeviceMessageHandler {
* @param message
*/
private void handleSwitchDataMessage(DeviceMessage message) {
List<SwitchDataVo> switchData = JsonDataParser.parseJsonData(message.getData(), SwitchDataVo.class);
List<CollectDataVo> switchData = JsonDataParser.parseJsonData(message.getData(), CollectDataVo.class);
if(!switchData.isEmpty()){
SwitchDataVo switchDataVo = switchData.get(0);
List<InitialSwitchInfo> switchInfos = JsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchInfo.class);
CollectDataVo switchDataVo = switchData.get(0);
switchDataVo.setValue(switchDataVo.getValue().replaceAll("\"noSuchInstance\"", "null"));
switch(switchDataVo.getType()){
case "switchNetCollect":
// handleSwitchMessage(message);
handleSwitchMessage(switchDataVo, message.getClientId());
break;
case "switchPwrCollect":
handleSwitchPwrMessage(switchDataVo, message.getClientId());
break;
case "switchModuleCollect":
handleSwitchModuleMessage(switchDataVo, message.getClientId());
break;
case "switchMpuCollect":
handleSwitchMpuMessage(switchDataVo, message.getClientId());
break;
case "switchFanCollect":
handleSwitchFanMessage(switchDataVo, message.getClientId());
break;
default:
handleSwitchOtherMessage(switchDataVo, message.getClientId());
break;
}
}else{
throw new RuntimeException("交换机data数据为空");
}
}
/**
* 电源发现数据
* @param switchDataVo
*/
private void handleSwitchPwrMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchPowerSupply> powerSupplyList = JsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchPowerSupply.class);
if (!powerSupplyList.isEmpty()){
InitialSwitchPowerSupply insertData = powerSupplyList.get(0);
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
initialSwitchPowerSupplyService.insertInitialSwitchPowerSupply(insertData);
}
}
/**
* 光模块发现数据
* @param switchDataVo
*/
private void handleSwitchModuleMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchOpticalModule> moduleList = JsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchOpticalModule.class);
if (!moduleList.isEmpty()){
InitialSwitchOpticalModule insertData = moduleList.get(0);
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
initialSwitchOpticalModuleService.insertInitialSwitchOpticalModule(insertData);
}
}
/**
* MPU发现数据
* @param switchDataVo
*/
private void handleSwitchMpuMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchMpuInfo> mpuList = JsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchMpuInfo.class);
if (!mpuList.isEmpty()){
InitialSwitchMpuInfo insertData = mpuList.get(0);
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
initialSwitchMpuInfoService.insertInitialSwitchMpuInfo(insertData);
}
}
/**
* 风扇发现数据
* @param switchDataVo
*/
private void handleSwitchFanMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchFanInfo> fanList = JsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchFanInfo.class);
if (!fanList.isEmpty()){
InitialSwitchFanInfo insertData = fanList.get(0);
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
initialSwitchFanInfoService.insertInitialSwitchFanInfo(insertData);
}
}
/**
* 其他发现数据（默认处理）
* @param switchDataVo
*/
private void handleSwitchOtherMessage(CollectDataVo switchDataVo, String clientId){
if (switchDataVo != null){
try {
InitialSwitchOtherCollectData insertData = new InitialSwitchOtherCollectData();
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
String value = switchDataVo.getValue();
// 解析JSON字符串
JsonNode jsonNode = objectMapper.readTree(value);
// 获取第一个字段的key和value
if (jsonNode.isObject() && jsonNode.fields().hasNext()) {
Map.Entry<String, JsonNode> entry = jsonNode.fields().next();
String fieldName = entry.getKey();
String fieldValue = entry.getValue().asText();
insertData.setCollectType(fieldName);
if(!"null".equals(fieldValue)){
insertData.setCollectValue(fieldValue);
}
} else if (jsonNode.isArray() && jsonNode.size() > 0) {
// 处理数组格式 [{}]
JsonNode firstElement = jsonNode.get(0);
if (firstElement.isObject() && firstElement.fields().hasNext()) {
Map.Entry<String, JsonNode> entry = firstElement.fields().next();
String fieldName = entry.getKey();
String fieldValue = entry.getValue().asText();
insertData.setCollectType(fieldName);
if(!"null".equals(fieldValue)){
insertData.setCollectValue(fieldValue);
}
}
}
insertInitialSwitchOtherInfo.insertInitialSwitchOtherCollectData(insertData);
} catch (Exception e) {
log.error("解析JSON数据失败: {}, value: {}", e.getMessage(), switchDataVo.getValue());
// 可以选择保存原始数据或进行其他错误处理
}
}
}
/**
* 交换机数据入库
* @param switchDataVo
*/
private void handleSwitchMessage(CollectDataVo switchDataVo, String clientId) {
List<InitialSwitchInfo> switchInfos = JsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchInfo.class);
if(!switchInfos.isEmpty()){
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime);
// 查询临时表信息，计算实际流量值
InitialSwitchInfoTemp temp = new InitialSwitchInfoTemp();
temp.setClientId(clientId);
List<InitialSwitchInfoTemp> tempList = initialSwitchInfoTempService.selectInitialSwitchInfoTempList(temp);
if(!tempList.isEmpty()){
// 1. 构建快速查找的Map
Map<String, InitialSwitchInfoTemp> tempMap = tempList.stream()
.collect(Collectors.toMap(
InitialSwitchInfoTemp::getName,
Function.identity(),
(existing, replacement) -> existing
));
// 2. 预计算除数（避免重复创建对象）
BigDecimal divisor = new BigDecimal(300);
// 3. 计算速度
switchInfos.forEach(switchInfo -> {
switchInfo.setClientId(clientId);
switchInfo.setCreateTime(createTime);
InitialSwitchInfoTemp tempInfo = tempMap.get(switchInfo.getName());
if (tempInfo != null) {
// 计算inSpeed
if (switchInfo.getInBytes() != null && tempInfo.getInBytes() != null) {
BigDecimal inDiff = switchInfo.getInBytes().subtract(tempInfo.getInBytes());
switchInfo.setInSpeed(inDiff.divide(divisor, 2, RoundingMode.HALF_UP));
}
// 计算outSpeed
if (switchInfo.getOutBytes() != null && tempInfo.getOutBytes() != null) {
BigDecimal outDiff = switchInfo.getOutBytes().subtract(tempInfo.getOutBytes());
switchInfo.setOutSpeed(outDiff.divide(divisor, 2, RoundingMode.HALF_UP));
}
}
});
}else{
switchInfos.forEach(switchInfo -> {
switchInfo.setClientId(clientId);
switchInfo.setCreateTime(createTime);
});
}
// 清空临时表对应switch信息
initialSwitchInfoTempService.truncateSwitchInfoTemp(clientId);
// 临时表 用来计算inSpeed outSeppd
initialSwitchInfoTempService.batchInsertInitialSwitchInfoTemp(switchInfos);
// 初始交换机数据入库
initialSwitchInfoService.batchInsertInitialSwitchInfo(switchInfos);
// 业务表入库
InitialSwitchInfoDetailsRemote detailsRemote = new InitialSwitchInfoDetailsRemote();
detailsRemote.setClientId(clientId);
detailsRemote.setStartTime(timeStr);
detailsRemote.setEndTime(timeStr);
remoteRevenueConfigService.autoSaveSwitchTraffic(detailsRemote, SecurityConstants.INNER);
}else{
throw new RuntimeException("交换机data数据为空");
}
}
/**
* 交换机数据入库
* @param message
@@ -296,7 +513,7 @@ public class DeviceMessageHandler {
if(!switchInfos.isEmpty()){
// 时间戳转换
long timestamp = switchInfos.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime);
// 查询临时表信息，计算实际流量值
@@ -356,6 +573,39 @@ public class DeviceMessageHandler {
throw new RuntimeException("交换机data数据为空");
}
}
/**
* 系统其他信息
* @param message
*/
private void handleOtherSystemMessage(DeviceMessage message) {
List<CollectDataVo> 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);
iInitialSystemOtherCollectDataService.insertInitialSystemOtherCollectData(insertData);
} catch (Exception e) {
log.error("解析JSON数据失败: {}, value: {}", e.getMessage(), systemDataVo.getValue());
// 可以选择保存原始数据或进行其他错误处理
}
}
}
}
/**
* 系统数据入库
* @param message
@@ -365,7 +615,7 @@ public class DeviceMessageHandler {
if(!systemInfos.isEmpty()){
// 时间戳转换
long timestamp = systemInfos.get(0).getTimestamp();
long millis = timestamp < 1_000_000_000L ? timestamp * 1000 : timestamp;
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
systemInfos.forEach(iface -> {
iface.setClientId(message.getClientId());
@@ -544,6 +794,19 @@ public class DeviceMessageHandler {
}
return null;
}
/**
* 脚本策略应答信息
* @param message
*/
private RspVo handleScriptRspMessage(DeviceMessage message) {
List<RspVo> rspVoList = JsonDataParser.parseJsonData(message.getData(), RspVo.class);
if (!rspVoList.isEmpty()) {
RspVo rsp = rspVoList.get(0);
log.info("应答信息：{}",rsp);
return rsp;
}
return null;
}
/**
* 注册应答处理
* @param message