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

服务器注册接口、snmp采集工具优化、其他bug修复

Browse Source
This commit is contained in:
gaoyutao
2025-10-17 18:56:32 +08:00
parent b3f16d2c8f
commit b0e63880fc
47 changed files with 1856 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/controller/InitialDiskInfoController.java
+11
View File
@@ -147,4 +147,15 @@ public class InitialDiskInfoController extends BaseController
Map<String, Object> echartsData = initialDiskInfoService.rwBytesEcharts(initialDiskInfo);
return success(echartsData);
}
/**
* 获取指定服务器的磁盘名称
* @param initialDiskInfo
* @return
*/
@RequiresPermissions("rocketmq:mountPointInfo:list")
@PostMapping("/getAllDistName")
public AjaxResult getAllDistName(@RequestBody InitialDiskInfo initialDiskInfo){
List<Map> list = initialDiskInfoService.getAllDistName(initialDiskInfo);
return success(list);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/controller/InitialDockerInfoController.java
+11
View File
@@ -146,4 +146,15 @@ public class InitialDockerInfoController extends BaseController
Map<String, Object> echartsData = initialDockerInfoService.netSpeedEcharts(initialDockerInfo);
return success(echartsData);
}
/**
* 获取指定服务器的容器id
* @param initialDockerInfo
* @return
*/
@RequiresPermissions("rocketmq:mountPointInfo:list")
@PostMapping("/getAllDockerId")
public AjaxResult getAllDockerId(@RequestBody InitialDockerInfo initialDockerInfo){
List<Map> list = initialDockerInfoService.getAllDockerId(initialDockerInfo);
return success(list);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/controller/InitialMountPointInfoController.java
+11
View File
@@ -130,5 +130,16 @@ public class InitialMountPointInfoController extends BaseController
Map<String, Object> echartsData = initialMountPointInfoService.spaceRateEcharts(initialMountPointInfo);
return success(echartsData);
}
/**
* 获取指定服务器的挂载文件系统名
* @param initialMountPointInfo
* @return
*/
@RequiresPermissions("rocketmq:mountPointInfo:list")
@PostMapping("/getAllMountName")
public AjaxResult getAllMountName(@RequestBody InitialMountPointInfo initialMountPointInfo){
List<Map> list = initialMountPointInfoService.getAllMountName(initialMountPointInfo);
return success(list);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/controller/RmMonitorPolicyController.java
+11
View File
@@ -107,6 +107,17 @@ public class RmMonitorPolicyController extends BaseController
{
return toAjax(rmMonitorPolicyService.issuePolicy(id));
}
/**
* 资源监控策略下发
*/
// @RequiresPermissions("rocketmq:policy:edit")
@Log(title = "issueSwitchPolicy", businessType = BusinessType.UPDATE)
@GetMapping("/issueSwitchPolicy")
public AjaxResult issueSwitchPolicy(Long id)
{
return toAjax(rmMonitorPolicyService.issueSwitchPolicy(id));
}
/**
* 删除监控模板
*/
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/InitialDiskInfo.java
+4
View File
@@ -57,5 +57,9 @@ public class InitialDiskInfo extends BaseEntity
/** 磁盘读取总字节数 */
@Excel(name = "磁盘读取总字节数")
private Long readBytes;
/** 开始时间 */
private String startTime;
/** 结束时间 */
private String endTime;
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/InitialDockerInfo.java
+4
View File
@@ -47,4 +47,8 @@ public class InitialDockerInfo extends BaseEntity
@Excel(name = "设备唯一标识")
private String clientId;
/** 开始时间 */
private String startTime;
/** 结束时间 */
private String endTime;
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/InitialMemoryInfo.java
+4
View File
@@ -47,5 +47,9 @@ public class InitialMemoryInfo extends BaseEntity
/** 总内存(字节) */
@Excel(name = "总内存(字节)")
private Long total;
/** 开始时间 */
private String startTime;
/** 结束时间 */
private String endTime;
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/InitialMountPointInfo.java
+6 -89
View File
@@ -2,8 +2,7 @@ package com.ruoyi.rocketmq.domain;
import com.ruoyi.common.core.annotation.Excel;
import com.ruoyi.common.core.web.domain.BaseEntity;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import lombok.Data;
import java.math.BigDecimal;
@@ -13,6 +12,7 @@ import java.math.BigDecimal;
* @author gyt
* @date 2025-08-25
*/
@Data
public class InitialMountPointInfo extends BaseEntity
{
private static final long serialVersionUID = 1L;
@@ -43,91 +43,8 @@ public class InitialMountPointInfo extends BaseEntity
/** 空间利用率(%) */
@Excel(name = "空间利用率")
private BigDecimal vfsUtil;
public void setId(Long id)
{
this.id = id;
}
public Long getId()
{
return id;
}
public void setClientId(String clientId)
{
this.clientId = clientId;
}
public String getClientId()
{
return clientId;
}
public void setMount(String mount)
{
this.mount = mount;
}
public String getMount()
{
return mount;
}
public void setVfsType(String vfsType)
{
this.vfsType = vfsType;
}
public String getVfsType()
{
return vfsType;
}
public void setVfsFree(Long vfsFree)
{
this.vfsFree = vfsFree;
}
public Long getVfsFree()
{
return vfsFree;
}
public void setVfsTotal(Long vfsTotal)
{
this.vfsTotal = vfsTotal;
}
public Long getVfsTotal()
{
return vfsTotal;
}
public void setVfsUtil(BigDecimal vfsUtil)
{
this.vfsUtil = vfsUtil;
}
public BigDecimal getVfsUtil()
{
return vfsUtil;
}
@Override
public String toString() {
return new ToStringBuilder(this,ToStringStyle.MULTI_LINE_STYLE)
.append("id", getId())
.append("clientId", getClientId())
.append("mount", getMount())
.append("vfsType", getVfsType())
.append("vfsFree", getVfsFree())
.append("vfsTotal", getVfsTotal())
.append("vfsUtil", getVfsUtil())
.append("createBy", getCreateBy())
.append("updateBy", getUpdateBy())
.append("createTime", getCreateTime())
.append("updateTime", getUpdateTime())
.toString();
}
/** 开始时间 */
private String startTime;
/** 结束时间 */
private String endTime;
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/InitialSwitchInfo.java
+4
View File
@@ -62,5 +62,9 @@ public class InitialSwitchInfo extends BaseEntity
/** 端口索引 */
private String ifIndex;
private String startTime;
private String endTime;
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/vo/SwitchOidVo.java
+20
View File
@@ -1,6 +1,8 @@
package com.ruoyi.rocketmq.domain.vo;
import lombok.Data;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.security.SecurityLevel;
import java.util.List;
import java.util.Map;
@@ -21,4 +23,22 @@ public class SwitchOidVo {
private Integer port;
/** 过滤值 */
private Map<String, List<String>> filters;
private int version = SnmpConstants.version2c; // 默认v2c
private int timeout = 5000;
private int retries = 2;
// SNMP v3特有参数
private String securityName;
private String authProtocol;
private String authPassword;
private String privProtocol;
private String privPassword;
private int securityLevel = SecurityLevel.AUTH_PRIV;
/**
* 判断是否为SNMP v3
*/
public boolean isV3() {
return version == SnmpConstants.version3;
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/mapper/InitialDiskInfoMapper.java
+8
View File
@@ -4,6 +4,7 @@ import com.ruoyi.rocketmq.domain.InitialDiskInfo;
import org.springframework.data.repository.query.Param;
import java.util.List;
import java.util.Map;
/**
* 磁盘监控信息Mapper接口
@@ -75,4 +76,11 @@ public interface InitialDiskInfoMapper
* @return
*/
InitialDiskInfo getDistDetailsMsgByClientId(InitialDiskInfo initialDiskInfo);
/**
* 获取指定服务器的磁盘名称
* @param initialDiskInfo
* @return
*/
List<Map> getAllDistName(InitialDiskInfo initialDiskInfo);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/mapper/InitialDockerInfoMapper.java
+8
View File
@@ -4,6 +4,7 @@ import com.ruoyi.rocketmq.domain.InitialDockerInfo;
import org.springframework.data.repository.query.Param;
import java.util.List;
import java.util.Map;
/**
* 容器监控信息Mapper接口
@@ -69,4 +70,11 @@ public interface InitialDockerInfoMapper
* @return
*/
InitialDockerInfo getDockerDetailsMsg(InitialDockerInfo initialDockerInfo);
/**
* 获取指定服务器的容器id
* @param initialDockerInfo
* @return
*/
List<Map> getAllDockerId(InitialDockerInfo initialDockerInfo);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/mapper/InitialMountPointInfoMapper.java
+8
View File
@@ -4,6 +4,7 @@ import com.ruoyi.rocketmq.domain.InitialMountPointInfo;
import org.springframework.data.repository.query.Param;
import java.util.List;
import java.util.Map;
/**
* 挂载点监控信息Mapper接口
@@ -74,4 +75,11 @@ public interface InitialMountPointInfoMapper
* @return
*/
InitialMountPointInfo pointDetailsMsg(String clientId);
/**
* 获取指定服务器的挂载文件系统名
* @param initialMountPointInfo
* @return
*/
List<Map> getAllMountName(InitialMountPointInfo initialMountPointInfo);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/IInitialDiskInfoService.java
+7
View File
@@ -92,4 +92,11 @@ public interface IInitialDiskInfoService
* @return
*/
Map<String, Object> rwBytesEcharts(InitialDiskInfo initialDiskInfo);
/**
* 获取指定服务器的磁盘名称
* @param initialDiskInfo
* @return
*/
List<Map> getAllDistName(InitialDiskInfo initialDiskInfo);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/IInitialDockerInfoService.java
+7
View File
@@ -94,4 +94,11 @@ public interface IInitialDockerInfoService
* @return
*/
Map<String, Object> netSpeedEcharts(InitialDockerInfo initialDockerInfo);
/**
* 获取指定服务器的容器id
* @param initialDockerInfo
* @return
*/
List<Map> getAllDockerId(InitialDockerInfo initialDockerInfo);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/IInitialMountPointInfoService.java
+7
View File
@@ -83,4 +83,11 @@ public interface IInitialMountPointInfoService
* @return
*/
Map<String, Object> spaceRateEcharts(InitialMountPointInfo initialMountPointInfo);
/**
* 获取指定服务器的挂载文件系统名
* @param initialMountPointInfo
* @return
*/
List<Map> getAllMountName(InitialMountPointInfo initialMountPointInfo);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/IRmMonitorPolicyService.java
+1
View File
@@ -73,6 +73,7 @@ public interface IRmMonitorPolicyService
Map<String, Object> getRmMonitorPolicyMsgById(Long id);
int issuePolicy(Long id);
int issueSwitchPolicy(Long id);
int updateResourcePolicy(RmMonitorPolicy rmMonitorPolicy);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/impl/InitialDiskInfoServiceImpl.java
+10
View File
@@ -169,4 +169,14 @@ public class InitialDiskInfoServiceImpl implements IInitialDiskInfoService
extractors.put("writeBytesData", info -> info.getWriteBytes());
return EchartsDataUtils.buildEchartsData(list, InitialDiskInfo::getCreateTime, extractors);
}
/**
* 获取指定服务器的磁盘名称
* @param initialDiskInfo
* @return
*/
@Override
public List<Map> getAllDistName(InitialDiskInfo initialDiskInfo) {
return initialDiskInfoMapper.getAllDistName(initialDiskInfo);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/impl/InitialDockerInfoServiceImpl.java
+10
View File
@@ -167,4 +167,14 @@ public class InitialDockerInfoServiceImpl implements IInitialDockerInfoService
extractors.put("netOutSpeedData", info -> info.getNetOutSpeed());
return EchartsDataUtils.buildEchartsData(list, InitialDockerInfo::getCreateTime, extractors);
}
/**
* 获取指定服务器的容器id
* @param initialDockerInfo
* @return
*/
@Override
public List<Map> getAllDockerId(InitialDockerInfo initialDockerInfo) {
return initialDockerInfoMapper.getAllDockerId(initialDockerInfo);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/impl/InitialMountPointInfoServiceImpl.java
+10
View File
@@ -150,4 +150,14 @@ public class InitialMountPointInfoServiceImpl implements IInitialMountPointInfoS
extractors.put("vfsUtilData", InitialMountPointInfo::getVfsUtil);
return EchartsDataUtils.buildEchartsData(list,InitialMountPointInfo::getCreateTime, extractors);
}
/**
* 获取指定服务器的挂载文件系统名
* @param initialMountPointInfo
* @return
*/
@Override
public List<Map> getAllMountName(InitialMountPointInfo initialMountPointInfo) {
return initialMountPointInfoMapper.getAllMountName(initialMountPointInfo);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/impl/RmMonitorPolicyServiceImpl.java
+200
View File
@@ -3,6 +3,7 @@ package com.ruoyi.rocketmq.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.core.enums.MsgEnum;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.common.core.utils.StringUtils;
import com.ruoyi.common.security.utils.SecurityUtils;
import com.ruoyi.rocketmq.domain.*;
import com.ruoyi.rocketmq.domain.vo.CollectVo;
@@ -13,10 +14,14 @@ import com.ruoyi.rocketmq.mapper.*;
import com.ruoyi.rocketmq.model.ProducerMode;
import com.ruoyi.rocketmq.producer.MessageProducer;
import com.ruoyi.rocketmq.service.IRmMonitorPolicyService;
import com.ruoyi.rocketmq.snmp.scheduler.MultiSwitchCollectionScheduler;
import com.ruoyi.rocketmq.utils.DataProcessUtil;
import com.ruoyi.rocketmq.utils.FieldNameConverterUtil;
import com.ruoyi.system.api.domain.RmResourceRegistrationRemote;
import com.ruoyi.system.api.domain.RmSwitchManagementRemote;
import lombok.extern.slf4j.Slf4j;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.security.SecurityLevel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
@@ -55,6 +60,11 @@ public class RmMonitorPolicyServiceImpl implements IRmMonitorPolicyService
@Autowired
private DataProcessUtil dataProcessUtil;
private final MultiSwitchCollectionScheduler scheduler;
public RmMonitorPolicyServiceImpl(MultiSwitchCollectionScheduler scheduler) {
this.scheduler = scheduler;
}
/**
@@ -355,6 +365,22 @@ public class RmMonitorPolicyServiceImpl implements IRmMonitorPolicyService
// 1. 获取资源监控策略基础信息
RmMonitorPolicy rmMonitorPolicy = rmMonitorPolicyMapper.selectRmMonitorPolicyById(id);
if(rmMonitorPolicy != null){
String clientIds = rmMonitorPolicy.getDeployDevice();
if(clientIds != null){
String[] clientArr = clientIds.split("\n");
if(StringUtils.equals("switch",rmMonitorPolicy.getResourceType())){
List<String> names = new ArrayList<>();
for (String clientId : clientArr) {
// 根据clientId查询设备名称
String switchName = dataProcessUtil.getDeviceNameByClientId(clientId);
names.add(switchName);
}
rmMonitorPolicy.setDeployDevice(String.join(",", names));
}
if(StringUtils.equals("linux",rmMonitorPolicy.getResourceType())){
rmMonitorPolicy.setDeployDevice(String.join(",", clientArr));
}
}
if(rmMonitorPolicy.getResourceGroupId()!=null){
rmMonitorPolicy.setResourceGroupName(dataProcessUtil.getResourceGroupNameById(rmMonitorPolicy.getResourceGroupId()));
}
@@ -379,6 +405,47 @@ public class RmMonitorPolicyServiceImpl implements IRmMonitorPolicyService
return result;
}
/**
* 下发策略v1.1
* @param id
* @return
*/
@Override
public int issueSwitchPolicy(Long id) {
try {
// 获取策略详情和资源设备
RmMonitorPolicy policy = rmMonitorPolicyMapper.selectRmMonitorPolicyById(id);
if (policy == null) {
log.error("策略不存在，id: {}", id);
return 0;
}
// 构建并发送采集配置
Map<String, Object> policyDetails = getRmMonitorPolicyMsgById(id);
boolean isSwitch = false;
if(policyDetails.get("switch") != null){
isSwitch = true;
}
List<CollectVo> collectVos = buildCollectConfigurations(policyDetails);
// 去重
List<CollectVo> uniqueList = collectVos.stream().distinct().collect(Collectors.toList());
if(isSwitch){
SwitchOidVo switchOidVo = buildOids(policyDetails);
startCollectSwitchMsgToDevices(policy.getDeployDevice(), uniqueList, switchOidVo);
}
// 更新策略状态为已下发
// RmMonitorPolicy policyUpdate = new RmMonitorPolicy();
// policyUpdate.setId(id);
// policyUpdate.setStatus("1");
// rmMonitorPolicyMapper.updateRmMonitorPolicy(policyUpdate);
return 1;
} catch (Exception e) {
log.error("下发策略失败，id: {}", id, e);
return 0;
}
}
/**
* 下发策略
@@ -691,4 +758,137 @@ public class RmMonitorPolicyServiceImpl implements IRmMonitorPolicyService
}
}
}
/**
* 开始采集交换机信息 (支持SNMP v1/v2c/v3)
*/
private void startCollectSwitchMsgToDevices(String clientIds, List<CollectVo> collectVos, SwitchOidVo switchOidVo) {
String[] clientIdArray = clientIds.split("\n");
List<MultiSwitchCollectionScheduler.DeviceConfig> deviceConfigs = new ArrayList<>();
for (String clientId : clientIdArray) {
try {
// 根据clientId 查询 交换机信息
RmSwitchManagementRemote switchMsg = dataProcessUtil.getSwitchMsg(clientId);
String switchIp = switchMsg.getSnmpAddress();
String community = switchMsg.getCommunityName();
int port = Math.toIntExact(switchMsg.getSnmpPort());
// 获取SNMP版本
String snmpVersion = switchMsg.getSnmpVersion();
int version = parseSnmpVersion(snmpVersion);
// 创建设备配置
MultiSwitchCollectionScheduler.DeviceConfig deviceConfig =
new MultiSwitchCollectionScheduler.DeviceConfig();
deviceConfig.setSwitchIp(switchIp);
deviceConfig.setPort(port);
deviceConfig.setClientId(clientId);
deviceConfig.setCollectVos(new ArrayList<>(collectVos));
deviceConfig.setSwitchOidVo(switchOidVo);
deviceConfig.setVersion(version);
// 根据SNMP版本设置不同的参数
if (version == SnmpConstants.version3) {
// SNMP v3配置
setupV3DeviceConfig(deviceConfig, switchMsg);
} else {
// SNMP v1/v2c配置
deviceConfig.setCommunity(community != null ? community : "public");
}
deviceConfigs.add(deviceConfig);
} catch (Exception e) {
log.error("处理设备配置失败，deviceId: {}", clientId, e);
}
}
// 批量启动交换机采集任务
scheduler.batchProcessSwitchConfigurations(deviceConfigs);
}
/**
* 解析SNMP版本字符串为对应的常量
*/
private int parseSnmpVersion(String snmpVersion) {
if (snmpVersion == null) {
return SnmpConstants.version2c; // 默认v2c
}
switch (snmpVersion.toLowerCase()) {
case "0":
return SnmpConstants.version1;
case "1":
return SnmpConstants.version2c;
case "3":
return SnmpConstants.version3;
default:
return SnmpConstants.version2c; // 默认v2c
}
}
/**
* 设置SNMP v3设备配置
*/
private void setupV3DeviceConfig(MultiSwitchCollectionScheduler.DeviceConfig deviceConfig,
RmSwitchManagementRemote switchMsg) {
// 设置安全名称（用户名）
String securityName = switchMsg.getSwitchUser();
if (securityName == null || securityName.trim().isEmpty()) {
securityName = switchMsg.getCommunityName(); // 如果没有用户名，使用团体名作为fallback
}
deviceConfig.setSecurityName(securityName != null ? securityName : "defaultUser");
// 设置认证协议
String authProtocol = parseAuthProtocol(switchMsg.getEncryptionMethod());
deviceConfig.setAuthProtocol(authProtocol);
// 设置认证密码
String authPassword = switchMsg.getSwitchPassword();
deviceConfig.setAuthPassword(authPassword != null ? authPassword : "");
// 设置加密协议（默认使用AES128）
deviceConfig.setPrivProtocol("AES128");
deviceConfig.setPrivPassword(authPassword != null ? authPassword : ""); // 通常加密密码与认证密码相同
// 设置安全级别
int securityLevel = parseSecurityLevel(switchMsg.getSecurityLevel());
deviceConfig.setSecurityLevel(securityLevel);
}
/**
* 解析认证协议
*/
private String parseAuthProtocol(String encryptionMethod) {
if (encryptionMethod == null) {
return "MD5"; // 默认MD5
}
switch (encryptionMethod.toLowerCase()) {
case "md5":
return "MD5";
case "sha":
case "sha1":
return "SHA";
default:
return "MD5"; // 默认MD5
}
}
/**
* 解析安全级别
*/
private int parseSecurityLevel(String securityLevel) {
if (securityLevel == null) {
return SecurityLevel.AUTH_PRIV; // 默认认证+加密
}
switch (securityLevel.toLowerCase()) {
case "1":
return SecurityLevel.NOAUTH_NOPRIV;
case "2":
return SecurityLevel.AUTH_NOPRIV;
case "3":
default:
return SecurityLevel.AUTH_PRIV;
}
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/snmp/DynamicOidCollector.java
+209 -57
View File
@@ -3,14 +3,15 @@ package com.ruoyi.rocketmq.snmp;
import com.ruoyi.common.core.utils.StringUtils;
import com.ruoyi.rocketmq.snmp.dto.CollectionResult;
import com.ruoyi.rocketmq.snmp.dto.SwitchOidDto;
import org.snmp4j.CommunityTarget;
import org.snmp4j.PDU;
import org.snmp4j.Snmp;
import org.snmp4j.TransportMapping;
import lombok.extern.slf4j.Slf4j;
import org.snmp4j.*;
import org.snmp4j.event.ResponseEvent;
import org.snmp4j.mp.MPv3;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.security.*;
import org.snmp4j.smi.*;
import org.snmp4j.transport.DefaultUdpTransportMapping;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.*;
@@ -21,27 +22,49 @@ import java.util.concurrent.Executors;
import java.util.stream.Collectors;
/**
* 动态OID多交换机采集管理器
* 动态OID多交换机采集管理器 (SNMP v3版本)
*/
@Service
@Slf4j
public class DynamicOidCollector {
// 全局SNMP实例
private Snmp snmp;
private boolean initialized = false;
private final Object initLock = new Object();
// 线程池
private final ExecutorService switchExecutor = Executors.newFixedThreadPool(10);
private final ExecutorService deviceTypeExecutor = Executors.newFixedThreadPool(20);
/**
* 初始化SNMP
* 初始化SNMP（线程安全）
*/
public void init() throws IOException {
TransportMapping<?> transport = new DefaultUdpTransportMapping();
snmp = new Snmp(transport);
transport.listen();
System.out.println("SNMP服务初始化完成");
synchronized (initLock) {
if (!initialized) {
TransportMapping<?> transport = new DefaultUdpTransportMapping();
snmp = new Snmp(transport);
// 添加SNMP v3支持
USM usm = new USM(SecurityProtocols.getInstance(),
new OctetString(MPv3.createLocalEngineID()), 0);
SecurityModels.getInstance().addSecurityModel(usm);
transport.listen();
initialized = true;
System.out.println("SNMP服务初始化完成");
}
}
}
/**
* 确保SNMP已初始化
*/
private void ensureInitialized() throws IOException {
if (!initialized) {
init();
}
}
/**
@@ -64,24 +87,24 @@ public class DynamicOidCollector {
* 根据设备类型获取信息（核心方法 - 动态OID版本）
*/
public CollectionResult getInterfaceInfoByType(SwitchOidDto switchVo, String deviceType) {
// 获取该设备类型的OID映射
Map<String, String> oidParams = switchVo.getOidsByType(deviceType);
if (oidParams == null || oidParams.isEmpty()) {
return new CollectionResult(switchVo.getIp(), deviceType, "该设备类型未配置OID");
}
List<Map<String, String>> results = new ArrayList<>();
try {
ensureInitialized();
// 获取该设备类型的OID映射
Map<String, String> oidParams = switchVo.getOidsByType(deviceType);
if (oidParams == null || oidParams.isEmpty()) {
return new CollectionResult(switchVo.getIp(), deviceType, "该设备类型未配置OID");
}
List<Map<String, String>> results = new ArrayList<>();
// 创建目标
CommunityTarget target = createTarget(switchVo);
Target target = createTarget(switchVo);
// 判断是否需要索引处理
boolean needIndex = needIndexProcessing(oidParams, deviceType);
// 获取过滤条件
List<String> filterValues = switchVo.getFilterByType(deviceType);
System.out.printf("采集交换机 %s 的 %s 设备信息，筛选条件: %s%n",
log.info("采集交换机 {} 的 {} 设备信息，筛选条件: {}",
switchVo.getIp(), deviceType, filterValues);
if(needIndex){
String indexOID = getIndexOidFromOidParams(oidParams);
// 获取索引OID（使用第一个OID作为索引基础）
@@ -91,7 +114,7 @@ public class DynamicOidCollector {
// 获取索引列表
List<Integer> indexes = getIndexes(target, indexOID, filterValues);
if (indexes.isEmpty()) {
System.out.printf("交换机 %s 的 %s 设备索引为空%n", switchVo.getIp(), deviceType);
log.info("交换机 {} 的 {} 设备索引为空", switchVo.getIp(), deviceType);
return new CollectionResult(switchVo.getIp(), deviceType, results, true);
}
@@ -113,12 +136,13 @@ public class DynamicOidCollector {
}
return new CollectionResult(switchVo.getIp(), deviceType, results, true);
} catch (Exception e) {
String errorMsg = String.format("采集失败: %s", e.getMessage());
System.err.printf("采集交换机 %s 的 %s 信息失败: %s%n",
String errorMsg = String.format("采集失败: {}", e.getMessage());
System.err.printf("采集交换机 {} 的 {} 信息失败: {}",
switchVo.getIp(), deviceType, errorMsg);
return new CollectionResult(switchVo.getIp(), deviceType, errorMsg);
}
}
/**
* 判断是否需要索引处理
* 针对特定的系统信息OID，不需要索引处理
@@ -162,7 +186,7 @@ public class DynamicOidCollector {
/**
* 获取设备索引列表
*/
private List<Integer> getIndexes(CommunityTarget target, String indexOID, List<String> filterValues)
private List<Integer> getIndexes(Target target, String indexOID, List<String> filterValues)
throws IOException {
List<Integer> indexes = new ArrayList<>();
@@ -216,7 +240,7 @@ public class DynamicOidCollector {
* @param oidParams OID参数映射
* @return 设备信息
*/
private Map<String, String> getDeviceInfo(CommunityTarget target, int index,
private Map<String, String> getDeviceInfo(Target target, int index,
Map<String, String> oidParams)
throws IOException {
Map<String, String> deviceInfo = new HashMap<>();
@@ -251,19 +275,118 @@ public class DynamicOidCollector {
}
/**
* 创建SNMP目标
* 创建SNMP目标 (支持v2c和v3)
*/
private CommunityTarget createTarget(SwitchOidDto switchVo) {
private Target createTarget(SwitchOidDto switchVo) {
Address targetAddress = new UdpAddress(switchVo.getIp() + "/" + switchVo.getPort());
// 根据版本选择创建不同的Target
if (switchVo.isV3()) {
return createV3Target(switchVo, targetAddress);
} else {
return createV2cTarget(switchVo, targetAddress);
}
}
/**
* 创建SNMP v2c目标
*/
private CommunityTarget createV2cTarget(SwitchOidDto switchVo, Address targetAddress) {
CommunityTarget target = new CommunityTarget();
target.setCommunity(new OctetString(switchVo.getCommunity()));
target.setAddress(targetAddress);
target.setRetries(2);
target.setTimeout(5000);
target.setRetries(switchVo.getRetries());
target.setTimeout(switchVo.getTimeout());
target.setVersion(SnmpConstants.version2c);
return target;
}
/**
* 创建SNMP v3目标
*/
private UserTarget createV3Target(SwitchOidDto switchVo, Address targetAddress) {
UserTarget target = new UserTarget();
target.setAddress(targetAddress);
target.setRetries(switchVo.getRetries());
target.setTimeout(switchVo.getTimeout());
target.setVersion(SnmpConstants.version3);
// 设置安全名称
target.setSecurityName(new OctetString(switchVo.getSecurityName()));
// 设置安全级别
target.setSecurityLevel(switchVo.getSecurityLevel());
return target;
}
/**
* 添加SNMP v3用户到SNMP引擎
*/
public void addV3User(SwitchOidDto switchVo) {
if (!switchVo.isV3()) {
return;
}
OctetString securityName = new OctetString(switchVo.getSecurityName());
OID authProtocol = getAuthProtocol(switchVo.getAuthProtocol());
OID privProtocol = getPrivProtocol(switchVo.getPrivProtocol());
UsmUser user = new UsmUser(
securityName,
authProtocol,
new OctetString(switchVo.getAuthPassword()),
privProtocol,
new OctetString(switchVo.getPrivPassword())
);
snmp.getUSM().addUser(securityName, user);
}
/**
* 获取认证协议OID
*/
private OID getAuthProtocol(String authProtocol) {
if (authProtocol == null) {
return AuthMD5.ID;
}
switch (authProtocol.toLowerCase()) {
case "md5":
return AuthMD5.ID;
case "sha":
return AuthSHA.ID;
case "sha1":
return AuthSHA.ID;
default:
return AuthMD5.ID;
}
}
/**
* 获取加密协议OID
*/
private OID getPrivProtocol(String privProtocol) {
if (privProtocol == null) {
return PrivDES.ID;
}
switch (privProtocol.toLowerCase()) {
case "des":
return PrivDES.ID;
case "3des":
return Priv3DES.ID;
case "aes128":
return PrivAES128.ID;
case "aes192":
return PrivAES192.ID;
case "aes256":
return PrivAES256.ID;
default:
return PrivDES.ID;
}
}
/**
* 采集单个交换机的所有设备类型
*/
@@ -273,10 +396,15 @@ public class DynamicOidCollector {
// 获取交换机支持的所有设备类型
List<String> deviceTypes = switchVo.getSupportedDeviceTypes();
if (deviceTypes.isEmpty()) {
System.out.printf("交换机 %s 未配置任何设备类型的OID%n", switchVo.getIp());
log.info("交换机 {} 未配置任何设备类型的OID", switchVo.getIp());
return results;
}
// 如果是v3版本，先添加用户
if (switchVo.isV3()) {
addV3User(switchVo);
}
// 并发采集所有设备类型
List<CompletableFuture<Void>> futures = deviceTypes.stream()
.map(deviceType -> CompletableFuture.runAsync(() -> {
@@ -284,10 +412,10 @@ public class DynamicOidCollector {
results.put(deviceType, result);
if (result.isSuccess()) {
System.out.printf("交换机 %s 的 %s 采集完成: %d 个设备%n",
log.info("交换机 {} 的 {} 采集完成: %d 个设备",
switchVo.getIp(), deviceType, result.getData().size());
} else {
System.out.printf("交换机 %s 的 %s 采集失败: %s%n",
log.info("交换机 {} 的 {} 采集失败: {}",
switchVo.getIp(), deviceType, result.getErrorMessage());
}
}, deviceTypeExecutor))
@@ -307,12 +435,12 @@ public class DynamicOidCollector {
List<CompletableFuture<Void>> futures = switches.stream()
.map(switchVo -> CompletableFuture.runAsync(() -> {
System.out.printf("开始采集交换机: %s%n", switchVo.getIp());
log.info("开始采集交换机: {}", switchVo.getIp());
Map<String, CollectionResult> switchResults = collectSwitch(switchVo);
allResults.put(switchVo.getIp(), switchResults);
System.out.printf("完成采集交换机: %s, 采集了 %d 种设备类型%n",
log.info("完成采集交换机: {}, 采集了 %d 种设备类型",
switchVo.getIp(), switchResults.size());
}, switchExecutor))
.collect(Collectors.toList());
@@ -324,10 +452,12 @@ public class DynamicOidCollector {
}
/**
* 创建示例交换机配置
* 创建示例交换机配置 (支持v2c和v3)
*/
public static SwitchOidDto createExampleSwitch(String ip, String community) {
SwitchOidDto switchVo = new SwitchOidDto(ip, community);
switchVo.setVersion(SnmpConstants.version2c); // 默认v2c
// 配置系统信息OID
Map<String, String> otherOids = new LinkedHashMap<>();
otherOids.put("1.3.6.1.2.1.1.1.0", "ifDescr");
@@ -338,6 +468,7 @@ public class DynamicOidCollector {
otherOids.put("1.3.6.1.2.1.1.6.0", "sysLocation");
otherOids.put("1.3.6.1.4.1.2011.5.25.183.1.4.0", "hwStackSystemMac");
switchVo.setOtherOID(otherOids);
// 配置网络接口OID
Map<String, String> netOids = new LinkedHashMap<>();
netOids.put("1.3.6.1.2.1.2.2.1.2", "ifDescr");
@@ -345,25 +476,46 @@ public class DynamicOidCollector {
netOids.put("1.3.6.1.2.1.2.2.1.8", "ifOperStatus");
switchVo.setNetOID(netOids);
// 配置电源OID
// Map<String, String> pwrOids = new LinkedHashMap<>();
// pwrOids.put("1.3.6.1.2.1.47.1.1.1.1.5", "pwrEntIndex");
// pwrOids.put("1.3.6.1.2.1.47.1.1.1.1.7", "pwrName");
// pwrOids.put("1.3.6.1.4.1.2011.5.25.31.1.1.18.1.6", "pwrEntityPwrState");
// pwrOids.put("1.3.6.1.4.1.2011.5.25.31.1.1.18.1.7", "pwrEntityPwrCurrent");
// pwrOids.put("1.3.6.1.4.1.2011.5.25.31.1.1.18.1.8", "pwrEntityPwrVoltage");
// switchVo.setPwrOID(pwrOids);
return switchVo;
}
// 配置过滤条件
Map<String, List<String>> filters = new HashMap<>();
// filters.put("pwr", Arrays.asList("6")); // 只采集正常和警告状态的电源
switchVo.setFilters(filters);
/**
* 创建SNMP v3示例交换机配置
*/
public static SwitchOidDto createV3ExampleSwitch(String ip, String securityName,
String authProtocol, String authPassword,
String privProtocol, String privPassword) {
SwitchOidDto switchVo = new SwitchOidDto(ip, null);
switchVo.setVersion(SnmpConstants.version3);
switchVo.setSecurityName(securityName);
switchVo.setAuthProtocol(authProtocol);
switchVo.setAuthPassword(authPassword);
switchVo.setPrivProtocol(privProtocol);
switchVo.setPrivPassword(privPassword);
switchVo.setSecurityLevel(org.snmp4j.security.SecurityLevel.AUTH_PRIV); // 认证+加密
// 配置系统信息OID
Map<String, String> otherOids = new LinkedHashMap<>();
otherOids.put("1.3.6.1.2.1.1.1.0", "ifDescr");
otherOids.put("1.3.6.1.2.1.1.2.0", "sysObjectID");
otherOids.put("1.3.6.1.2.1.1.3.0", "sysUpTime");
otherOids.put("1.3.6.1.2.1.1.4.0", "sysContact");
otherOids.put("1.3.6.1.2.1.1.5.0", "sysName");
otherOids.put("1.3.6.1.2.1.1.6.0", "sysLocation");
switchVo.setOtherOID(otherOids);
// 配置网络接口OID
Map<String, String> netOids = new LinkedHashMap<>();
netOids.put("1.3.6.1.2.1.2.2.1.2", "ifDescr");
netOids.put("1.3.6.1.2.1.2.2.1.3", "ifType");
netOids.put("1.3.6.1.2.1.2.2.1.8", "ifOperStatus");
switchVo.setNetOID(netOids);
return switchVo;
}
/**
* 主程序入口
* 主程序入口 (支持v2c和v3测试)
*/
public static void main(String[] args) {
DynamicOidCollector collector = new DynamicOidCollector();
@@ -372,10 +524,10 @@ public class DynamicOidCollector {
// 初始化
collector.init();
// 创建动态配置的交换机列表
// 创建动态配置的交换机列表 (支持v2c和v3混合)
List<SwitchOidDto> switches = Arrays.asList(
createExampleSwitch("123.182.89.17", "chengde_200G"),
createExampleSwitch("1.194.193.81", "henan_anyang_10G")
createExampleSwitch("123.182.89.17", "chengde_200G"), // v2c
createV3ExampleSwitch("1.194.193.81", "v3user", "SHA", "auth123", "AES128", "priv123") // v3
);
// 并发采集所有交换机
@@ -399,11 +551,11 @@ public class DynamicOidCollector {
.sum();
totalDevices += switchDevices;
System.out.printf("交换机 %s: %d 个设备%n", ip, switchDevices);
log.info("交换机 {}: %d 个设备", ip, switchDevices);
}
System.out.printf("总计: %d 台交换机, %d 个设备%n", results.size(), totalDevices);
System.out.printf("总采集时间: %.2f 秒%n", (endTime - startTime) / 1000.0);
log.info("总计: %d 台交换机, %d 个设备", results.size(), totalDevices);
log.info("总采集时间: %.2f 秒", (endTime - startTime) / 1000.0);
} catch (Exception e) {
System.err.println("采集程序出错: " + e.getMessage());
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/snmp/config/SchedulerConfig.java
+24
View File
@@ -0,0 +1,24 @@
package com.ruoyi.rocketmq.snmp.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@Configuration
@EnableScheduling
public class SchedulerConfig {
@Bean
public TaskScheduler taskScheduler() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(20); // 根据您的交换机数量调整
scheduler.setThreadNamePrefix("switch-collector-");
scheduler.setAwaitTerminationSeconds(60);
scheduler.setWaitForTasksToCompleteOnShutdown(true);
scheduler.setRemoveOnCancelPolicy(true);
scheduler.initialize(); // 重要：必须调用initialize()
return scheduler;
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/snmp/dto/CollectionResult.java
+1
View File
@@ -17,6 +17,7 @@ public class CollectionResult {
boolean success;
String errorMessage;
public CollectionResult(){}
public CollectionResult(String switchIp, String deviceType,
List<Map<String, String>> data, boolean success) {
this.switchIp = switchIp;
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/snmp/dto/SwitchOidDto.java
+76 -42
View File
@@ -1,56 +1,97 @@
package com.ruoyi.rocketmq.snmp.dto;
import lombok.Data;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.security.SecurityLevel;
import java.util.*;
/**
* 交换机OID配置实体类
* 交换机OID配置DTO (支持SNMP v2c和v3)
*/
@Data
public class SwitchOidDto {
private Map<String, String> netOID; // 网络接口OID映射
private Map<String, String> moduleOID; // 模块OID映射
private Map<String, String> mpuOID; // MPU OID映射
private Map<String, String> pwrOID; // 电源OID映射
private Map<String, String> fanOID; // 风扇OID映射
private Map<String, String> otherOID; // 其他OID映射
/** 团体名 */
private String community;
/** IP地址 */
private String ip;
/** 端口 */
private Integer port;
/** 过滤值 Map<设备类型, 过滤值列表> */
private Map<String, List<String>> filters;
private String community;
private int port = 161;
private int version = SnmpConstants.version2c; // 默认v2c
private int timeout = 5000;
private int retries = 2;
// SNMP v3特有参数
private String securityName;
private String authProtocol;
private String authPassword;
private String privProtocol;
private String privPassword;
private int securityLevel = SecurityLevel.AUTH_PRIV;
// OID配置
private Map<String, String> netOID = new LinkedHashMap<>();
private Map<String, String> otherOID = new LinkedHashMap<>();
private Map<String, String> entityOID = new LinkedHashMap<>();
private Map<String, String> pwrOID = new LinkedHashMap<>();
private Map<String, String> fanOID = new LinkedHashMap<>();
private Map<String, String> mpuOID = new LinkedHashMap<>();
private Map<String, String> moduleOID = new LinkedHashMap<>();
private Map<String, String> powerOID = new LinkedHashMap<>();
// 过滤条件
private Map<String, List<String>> filters = new HashMap<>();
public SwitchOidDto() {
}
// 构造函数
public SwitchOidDto(String ip, String community) {
this.ip = ip;
this.community = community;
this.port = 161;
this.filters = new HashMap<>();
this.netOID = new LinkedHashMap<>();
this.moduleOID = new LinkedHashMap<>();
this.mpuOID = new LinkedHashMap<>();
this.pwrOID = new LinkedHashMap<>();
this.fanOID = new LinkedHashMap<>();
this.otherOID = new LinkedHashMap<>();
}
public SwitchOidDto(String ip, String community, int port) {
this.ip = ip;
this.community = community;
this.port = port;
}
/**
* 获取支持的所有设备类型
*/
public List<String> getSupportedDeviceTypes() {
List<String> types = new ArrayList<>();
if (!netOID.isEmpty()) types.add("net");
if (!otherOID.isEmpty()) types.add("other");
if (!entityOID.isEmpty()) types.add("entity");
if (!pwrOID.isEmpty()) types.add("pwr");
if (!fanOID.isEmpty()) types.add("fan");
if (!mpuOID.isEmpty()) types.add("mpu");
if (!moduleOID.isEmpty()) types.add("module");
if (!powerOID.isEmpty()) types.add("power");
return types;
}
/**
* 根据设备类型获取OID映射
*/
public Map<String, String> getOidsByType(String deviceType) {
switch (deviceType.toLowerCase()) {
case "net": return netOID;
case "module": return moduleOID;
case "mpu": return mpuOID;
case "pwr": return pwrOID;
case "fan": return fanOID;
default: return otherOID;
switch (deviceType) {
case "net":
return netOID;
case "other":
return otherOID;
case "entity":
return entityOID;
case "pwr":
return pwrOID;
case "fan":
return fanOID;
case "mpu":
return mpuOID;
case "module":
return moduleOID;
case "power":
return powerOID;
default:
return Collections.emptyMap();
}
}
@@ -58,20 +99,13 @@ public class SwitchOidDto {
* 根据设备类型获取过滤条件
*/
public List<String> getFilterByType(String deviceType) {
return filters.getOrDefault(deviceType.toLowerCase(), Collections.emptyList());
return filters.get(deviceType);
}
/**
* 获取所有支持的设备类型
* 判断是否为SNMP v3
*/
public List<String> getSupportedDeviceTypes() {
List<String> types = new ArrayList<>();
if (!netOID.isEmpty()) types.add("net");
if (!moduleOID.isEmpty()) types.add("module");
if (!mpuOID.isEmpty()) types.add("mpu");
if (!pwrOID.isEmpty()) types.add("pwr");
if (!fanOID.isEmpty()) types.add("fan");
if (!otherOID.isEmpty()) types.add("other");
return types;
public boolean isV3() {
return version == SnmpConstants.version3;
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/snmp/scheduler/MultiSwitchCollectionScheduler.java
+639
View File
@@ -0,0 +1,639 @@
package com.ruoyi.rocketmq.snmp.scheduler;
import com.ruoyi.rocketmq.domain.vo.CollectVo;
import com.ruoyi.rocketmq.domain.vo.SwitchOidVo;
import com.ruoyi.rocketmq.snmp.DynamicOidCollector;
import com.ruoyi.rocketmq.snmp.dto.CollectionResult;
import com.ruoyi.rocketmq.snmp.dto.SwitchOidDto;
import com.ruoyi.rocketmq.snmp.service.ProcessSwitchCollectDataService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.security.SecurityLevel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
@Service
@Slf4j
public class MultiSwitchCollectionScheduler {
private final TaskScheduler taskScheduler;
private final DynamicOidCollector dynamicOidCollector;
private boolean collectorInitialized = false;
// 存储所有交换机的任务和配置
private final Map<String, Map<String, ScheduledFuture<?>>> allSwitchTasks = new ConcurrentHashMap<>();
private final Map<String, SwitchOidDto> allSwitchConfigs = new ConcurrentHashMap<>();
private final Map<String, List<CollectVo>> allSwitchCollectConfigs = new ConcurrentHashMap<>();
// 使用构造器注入而不是@Autowired
public MultiSwitchCollectionScheduler(TaskScheduler taskScheduler,
DynamicOidCollector dynamicOidCollector) {
this.taskScheduler = taskScheduler;
this.dynamicOidCollector = dynamicOidCollector;
}
@Autowired
private ProcessSwitchCollectDataService processSwitchCollectDataService;
@PostConstruct
public void init() {
try {
// 初始化SNMP采集器
dynamicOidCollector.init();
collectorInitialized = true;
log.info("MultiSwitchCollectionScheduler 初始化完成，SNMP服务已启动");
} catch (Exception e) {
log.error("SNMP服务初始化失败", e);
collectorInitialized = false;
}
}
/**
* 确保采集器已初始化
*/
private boolean ensureCollectorInitialized() {
if (!collectorInitialized) {
try {
dynamicOidCollector.init();
collectorInitialized = true;
log.info("SNMP采集器初始化成功");
} catch (Exception e) {
log.error("SNMP采集器初始化失败", e);
return false;
}
}
return true;
}
/**
* 根据CollectVo列表启动交换机采集任务（支持SNMP v3）
*/
public void startSwitchCollection(String switchIp, String community, int port, String clientId,
List<CollectVo> collectVos, SwitchOidVo switchOidVo) {
// 确保采集器已初始化
if (!ensureCollectorInitialized()) {
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", switchIp);
return;
}
// 停止该交换机的现有任务
stopSwitchCollection(switchIp);
// 创建交换机配置
SwitchOidDto switchVo = createSwitchConfig(switchIp, community, port, collectVos, switchOidVo);
// 初始化存储结构
Map<String, ScheduledFuture<?>> switchTasks = new ConcurrentHashMap<>();
// 根据CollectVo配置创建不同的定时任务
for (CollectVo collectVo : collectVos) {
if (collectVo.isCollect()) { //剔除switchVo中的其他oid
ScheduledFuture<?> task = createCollectionTask(switchIp, clientId, switchVo, collectVo);
if (task != null) {
switchTasks.put(collectVo.getType(), task);
}
}
}
// 存储配置
allSwitchTasks.put(switchIp, switchTasks);
allSwitchConfigs.put(switchIp, switchVo);
allSwitchCollectConfigs.put(switchIp, new ArrayList<>(collectVos));
log.info("启动交换机 {} 的采集任务，包含 {} 个采集类型{}",
switchIp, switchTasks.size());
}
/**
* 启动SNMP v3交换机采集任务
*/
public void startV3SwitchCollection(String switchIp, int port, String clientId, String securityName,
String authProtocol, String authPassword,
String privProtocol, String privPassword,
int securityLevel, List<CollectVo> collectVos,
SwitchOidVo switchOidVo) {
// 停止该交换机的现有任务
stopSwitchCollection(switchIp);
// 创建SNMP v3交换机配置
SwitchOidDto switchVo = createV3SwitchConfig(switchIp, port, securityName,
authProtocol, authPassword, privProtocol, privPassword,
securityLevel, collectVos, switchOidVo);
// 初始化存储结构
Map<String, ScheduledFuture<?>> switchTasks = new ConcurrentHashMap<>();
// 根据CollectVo配置创建不同的定时任务
for (CollectVo collectVo : collectVos) {
if (collectVo.isCollect()) {
ScheduledFuture<?> task = createCollectionTask(switchIp, clientId, switchVo, collectVo);
if (task != null) {
switchTasks.put(collectVo.getType(), task);
}
}
}
// 存储配置
allSwitchTasks.put(switchIp, switchTasks);
allSwitchConfigs.put(switchIp, switchVo);
allSwitchCollectConfigs.put(switchIp, new ArrayList<>(collectVos));
log.info("启动SNMP v3交换机 {} 的采集任务，包含 {} 个采集类型{}",
switchIp, switchTasks.size());
}
/**
* 根据CollectVo创建对应的采集任务
*/
private ScheduledFuture<?> createCollectionTask(String switchIp, String clientId, SwitchOidDto switchVo, CollectVo collectVo) {
String type = collectVo.getType();
long interval = collectVo.getInterval() != null ? collectVo.getInterval() : 300L;
if (isTrafficCollection(type)) {
// 流量采集：固定时间点（0,5,10,15...分钟）
return startTrafficCollection(switchIp, clientId, switchVo, collectVo);
} else {
// 常规采集：相对间隔
return startRegularCollection(switchIp, clientId, switchVo, collectVo);
}
}
/**
* 判断是否为流量采集
*/
private boolean isTrafficCollection(String type) {
return "switchNetCollect".equals(type);
}
/**
* 启动流量采集任务（固定时间点）
*/
private ScheduledFuture<?> startTrafficCollection(String switchIp, String clientId, SwitchOidDto switchVo, CollectVo collectVo) {
// 确保采集器已初始化
if (!ensureCollectorInitialized()) {
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", switchIp);
return null;
}
// 创建Cron表达式：每5分钟的第0秒执行（0,5,10,15...）
String cronExpression = "0 0/5 * * * ?";
Runnable trafficTask = () -> {
try {
// 获取当前时间作为采集时间戳（对齐到5分钟整点）
long collectionTimestamp = getAlignedTimestamp();
log.info("[流量采集] 开始采集交换机 {} 的网络数据, 时间: {}", switchIp, collectionTimestamp);
CollectionResult result = dynamicOidCollector.getInterfaceInfoByType(switchVo, "net");
// 设置采集时间戳
result.setTimestamp(collectionTimestamp);
// 这里可以处理采集结果，比如保存到数据库或发送到消息队列
processCollectionResult(result, clientId);
log.info("[流量采集] 完成采集交换机 {} 的网络数据, 时间戳: {}", switchIp, collectionTimestamp);
} catch (Exception e) {
log.error("[流量采集] 采集交换机 {} 数据失败: {}", switchIp, e.getMessage());
}
};
CronTrigger trigger = new CronTrigger(cronExpression);
ScheduledFuture<?> future = taskScheduler.schedule(trafficTask, trigger);
log.info("交换机 {} 流量采集任务启动，时间点: 每5分钟整点", switchIp);
return future;
}
/**
* 获取对齐的时间戳（5分钟整点，精确到秒）
*/
private long getAlignedTimestamp() {
long currentTime = System.currentTimeMillis() / 1000; // 当前时间戳（秒）
// 计算5分钟的秒数
long fiveMinutes = 5 * 60;
// 对齐到5分钟整点
long alignedTimestamp = (currentTime / fiveMinutes) * fiveMinutes;
return alignedTimestamp;
}
/**
* 处理采集结果
*/
private void processCollectionResult(CollectionResult result, String clientId) {
if (result.isSuccess()) {
log.debug("流量采集成功 - 交换机: {}, 设备类型: {}, 时间: {}, 数据量: {}",
result.getSwitchIp(), result.getDeviceType(),
result.getTimestamp(), result.getData().size());
processSwitchCollectDataService.handleSwitchDataMessage(result, clientId);
} else {
log.warn("流量采集失败 - 交换机: {}, 设备类型: {}, 时间: {}, 错误: {}",
result.getSwitchIp(), result.getDeviceType(),
result.getTimestamp(), result.getErrorMessage());
}
}
/**
* 启动常规采集任务（相对间隔）
*/
private ScheduledFuture<?> startRegularCollection(String switchIp, String clientId, SwitchOidDto switchVo, CollectVo collectVo) {
// 确保采集器已初始化
if (!ensureCollectorInitialized()) {
log.error("SNMP采集器未初始化，无法启动交换机 {} 的采集任务", switchIp);
return null;
}
long intervalSeconds = collectVo.getInterval() != null ? collectVo.getInterval() : 300L;
Instant startTime = calculateAlignedStartTime(intervalSeconds);
Duration interval = Duration.ofSeconds(intervalSeconds);
Runnable regularTask = () -> {
try {
log.info("[{}] 开始采集交换机 {} 的数据, 时间: {}",
collectVo.getType(), switchIp, LocalDateTime.now());
// 获取对齐到整分钟的时间戳
long collectionTimestamp = getAlignedMinuteTimestamp();
// 根据类型执行对应的采集
CollectionResult result = executeCollectionByType(switchVo, collectVo.getType());
// 设置采集时间戳
result.setTimestamp(collectionTimestamp);
// 处理采集结果
processCollectionResult(result, clientId);
log.info("[{}] 完成采集交换机 {} 的数据", collectVo.getType(), switchIp);
} catch (Exception e) {
log.error("[{}] 采集交换机 {} 数据失败: {}",
collectVo.getType(), switchIp, e.getMessage());
}
};
ScheduledFuture<?> future = taskScheduler.scheduleWithFixedDelay(
regularTask, startTime, interval);
log.info("交换机 {} [{}] 采集任务启动，间隔: {}秒, 开始时间: {}",
switchIp, collectVo.getType(), intervalSeconds, startTime);
return future;
}
/**
* 获取对齐到整分钟的时间戳（秒为00）
*/
private long getAlignedMinuteTimestamp() {
long currentTime = System.currentTimeMillis() / 1000; // 当前时间戳（秒）
// 对齐到整分钟（去掉秒部分）
long alignedTimestamp = (currentTime / 60) * 60;
return alignedTimestamp;
}
/**
* 根据采集类型执行对应的采集
*/
private CollectionResult executeCollectionByType(SwitchOidDto switchVo, String type) {
String deviceType;
switch (type) {
case "switchSysDescrCollect":
case "switchSysObjectIDCollect":
case "switchSysUpTimeCollect":
case "switchSysContactCollect":
case "switchSysNameCollect":
case "switchSysLocationCollect":
case "switchHwStackSystemMacCollect":
case "switchEntIndexCollect":
case "switchEntPhysicalNameCollect":
case "switchEntPhysicalSoftwareRevCollect":
case "switchHwEntityCpuUsageCollect":
case "switchHwEntityMemUsageCollect":
case "switchHwAveragePowerCollect":
case "switchHwCurrentPowerCollect":
deviceType = "other";
break;
case "switchPwrCollect":
deviceType = "pwr";
break;
case "switchFanCollect":
deviceType = "fan";
break;
case "switchMpuCollect":
deviceType = "mpu";
break;
case "switchModuleCollect":
deviceType = "module";
break;
default:
// 返回错误结果
throw new RuntimeException(switchVo.getIp() + " 未知的采集类型：" + type);
}
return dynamicOidCollector.getInterfaceInfoByType(switchVo, deviceType);
}
/**
* 计算对齐的开始时间（智能对齐到整秒，使用上海时区）
*/
private Instant calculateAlignedStartTime(long intervalSeconds) {
// 指定上海时区
ZoneId shanghaiZone = ZoneId.of("Asia/Shanghai");
LocalDateTime now = LocalDateTime.now(shanghaiZone);
LocalDateTime alignedTime;
// 检查当前时间是否已经是整秒（秒为0）
if (now.getSecond() == 0) {
// 如果当前时间已经是整秒，立即开始（或者延迟几秒开始）
alignedTime = now.plusSeconds(5); // 延迟5秒开始，避免立即执行可能的问题
} else {
// 否则对齐到下一分钟的00秒
alignedTime = now.plusMinutes(1)
.withSecond(0)
.withNano(0);
}
Instant instant = alignedTime.atZone(shanghaiZone).toInstant();
log.info("任务启动时间: {} (上海时间), 间隔: {}秒", alignedTime, intervalSeconds);
return instant;
}
/**
* 创建交换机配置（根据SwitchOidVo和CollectVo）
*/
private SwitchOidDto createSwitchConfig(String ip, String community, int port,
List<CollectVo> collectVos, SwitchOidVo switchOidVo) {
SwitchOidDto switchVo = new SwitchOidDto(ip, community);
switchVo.setPort(port);
switchVo.setVersion(switchOidVo.getVersion());
// 使用SwitchOidVo中的OID配置
if (switchOidVo.getNetOID() != null && !switchOidVo.getNetOID().isEmpty()) {
switchVo.setNetOID(new LinkedHashMap<>(switchOidVo.getNetOID()));
}
if (switchOidVo.getOtherOID() != null && !switchOidVo.getOtherOID().isEmpty()) {
switchVo.setOtherOID(new LinkedHashMap<>(switchOidVo.getOtherOID()));
}
if (switchOidVo.getModuleOID() != null && !switchOidVo.getModuleOID().isEmpty()) {
// 将moduleOID映射到对应的设备类型
switchVo.setModuleOID(new LinkedHashMap<>(switchOidVo.getModuleOID()));
}
// 设置过滤条件
if (switchOidVo.getFilters() != null && !switchOidVo.getFilters().isEmpty()) {
switchVo.setFilters(new HashMap<>(switchOidVo.getFilters()));
}
// 设置超时和重试次数
if (switchOidVo.getTimeout() > 0) {
switchVo.setTimeout(switchOidVo.getTimeout());
}
if (switchOidVo.getRetries() > 0) {
switchVo.setRetries(switchOidVo.getRetries());
}
return switchVo;
}
/**
* 创建SNMP v3交换机配置
*/
private SwitchOidDto createV3SwitchConfig(String ip, int port, String securityName,
String authProtocol, String authPassword,
String privProtocol, String privPassword,
int securityLevel, List<CollectVo> collectVos,
SwitchOidVo switchOidVo) {
SwitchOidDto switchVo = new SwitchOidDto(ip, null);
switchVo.setPort(port);
switchVo.setVersion(SnmpConstants.version3);
switchVo.setSecurityName(securityName);
switchVo.setAuthProtocol(authProtocol);
switchVo.setAuthPassword(authPassword);
switchVo.setPrivProtocol(privProtocol);
switchVo.setPrivPassword(privPassword);
switchVo.setSecurityLevel(securityLevel);
// 使用SwitchOidVo中的OID配置
if (switchOidVo.getNetOID() != null && !switchOidVo.getNetOID().isEmpty()) {
switchVo.setNetOID(new LinkedHashMap<>(switchOidVo.getNetOID()));
}
if (switchOidVo.getOtherOID() != null && !switchOidVo.getOtherOID().isEmpty()) {
switchVo.setOtherOID(new LinkedHashMap<>(switchOidVo.getOtherOID()));
}
if (switchOidVo.getModuleOID() != null && !switchOidVo.getModuleOID().isEmpty()) {
switchVo.setModuleOID(new LinkedHashMap<>(switchOidVo.getModuleOID()));
}
// 设置过滤条件
if (switchOidVo.getFilters() != null && !switchOidVo.getFilters().isEmpty()) {
switchVo.setFilters(new HashMap<>(switchOidVo.getFilters()));
}
// 设置超时和重试次数
if (switchOidVo.getTimeout() > 0) {
switchVo.setTimeout(switchOidVo.getTimeout());
}
if (switchOidVo.getRetries() > 0) {
switchVo.setRetries(switchOidVo.getRetries());
}
return switchVo;
}
/**
* 停止单个交换机的所有采集任务
*/
public void stopSwitchCollection(String switchIp) {
Map<String, ScheduledFuture<?>> tasks = allSwitchTasks.remove(switchIp);
if (tasks != null) {
tasks.values().forEach(future -> future.cancel(false));
log.info("停止交换机 {} 的所有采集任务{}", switchIp);
}
allSwitchConfigs.remove(switchIp);
allSwitchCollectConfigs.remove(switchIp);
}
/**
* 停止所有交换机的采集任务
*/
public void stopAllSwitchCollections() {
allSwitchTasks.values().forEach(tasks ->
tasks.values().forEach(future -> future.cancel(false)));
allSwitchTasks.clear();
allSwitchConfigs.clear();
allSwitchCollectConfigs.clear();
System.out.println("停止所有交换机的采集任务");
}
/**
* 获取交换机的任务状态
*/
public Map<String, Object> getSwitchStatus(String switchIp) {
Map<String, Object> status = new HashMap<>();
status.put("switchIp", switchIp);
Map<String, ScheduledFuture<?>> tasks = allSwitchTasks.get(switchIp);
List<CollectVo> collectConfigs = allSwitchCollectConfigs.get(switchIp);
if (tasks != null && collectConfigs != null) {
status.put("isRunning", true);
status.put("taskCount", tasks.size());
// 获取交换机配置信息
SwitchOidDto switchConfig = allSwitchConfigs.get(switchIp);
if (switchConfig != null) {
status.put("snmpVersion", switchConfig.getVersion() == SnmpConstants.version3 ? "v3" : "v2c");
status.put("securityName", switchConfig.getSecurityName());
status.put("securityLevel", getSecurityLevelDescription(switchConfig.getSecurityLevel()));
}
List<Map<String, Object>> taskStatusList = new ArrayList<>();
for (CollectVo collectVo : collectConfigs) {
Map<String, Object> taskStatus = new HashMap<>();
taskStatus.put("type", collectVo.getType());
taskStatus.put("collect", collectVo.isCollect());
taskStatus.put("interval", collectVo.getInterval());
ScheduledFuture<?> task = tasks.get(collectVo.getType());
taskStatus.put("isActive", task != null && !task.isCancelled());
taskStatusList.add(taskStatus);
}
status.put("tasks", taskStatusList);
} else {
status.put("isRunning", false);
}
return status;
}
/**
* 获取安全级别描述
*/
private String getSecurityLevelDescription(int securityLevel) {
switch (securityLevel) {
case SecurityLevel.NOAUTH_NOPRIV:
return "noAuthNoPriv";
case SecurityLevel.AUTH_NOPRIV:
return "authNoPriv";
case SecurityLevel.AUTH_PRIV:
return "authPriv";
default:
return "unknown";
}
}
/**
* 获取所有交换机的状态
*/
public Map<String, Map<String, Object>> getAllSwitchesStatus() {
Map<String, Map<String, Object>> allStatus = new HashMap<>();
for (String switchIp : allSwitchTasks.keySet()) {
allStatus.put(switchIp, getSwitchStatus(switchIp));
}
return allStatus;
}
/**
* 更新交换机的采集配置
*/
public void updateSwitchConfig(String switchIp, String community, int port, String clientId,
List<CollectVo> collectVos, SwitchOidVo switchOidVo) {
// 先停止现有任务
stopSwitchCollection(switchIp);
// 重新启动任务
startSwitchCollection(switchIp, community, port, clientId, collectVos, switchOidVo);
log.info("更新交换机 {} 的采集配置{}", switchIp);
}
/**
* 更新SNMP v3交换机的采集配置
*/
public void updateV3SwitchConfig(String switchIp, int port, String clientId, String securityName,
String authProtocol, String authPassword,
String privProtocol, String privPassword,
int securityLevel, List<CollectVo> collectVos,
SwitchOidVo switchOidVo) {
// 先停止现有任务
stopSwitchCollection(switchIp);
// 重新启动任务
startV3SwitchCollection(switchIp, port, clientId, securityName, authProtocol, authPassword,
privProtocol, privPassword, securityLevel, collectVos, switchOidVo);
log.info("更新SNMP v3交换机 {} 的采集配置{}", switchIp);
}
/**
* 批量处理设备配置（支持v2c和v3混合）
*/
public void batchProcessSwitchConfigurations(List<DeviceConfig> devices) {
for (DeviceConfig device : devices) {
try {
if (device.getVersion() == SnmpConstants.version3) {
// SNMP v3设备
startV3SwitchCollection(
device.getSwitchIp(),
device.getPort(),
device.getClientId(),
device.getSecurityName(),
device.getAuthProtocol(),
device.getAuthPassword(),
device.getPrivProtocol(),
device.getPrivPassword(),
device.getSecurityLevel(),
device.getCollectVos(),
device.getSwitchOidVo()
);
} else {
// SNMP v2c设备
startSwitchCollection(
device.getSwitchIp(),
device.getCommunity(),
device.getPort(),
device.getClientId(),
device.getCollectVos(),
device.getSwitchOidVo()
);
}
} catch (Exception e) {
log.error("处理交换机配置失败，switchIp: {}", device.getSwitchIp(), e);
}
}
}
/**
* 设备配置类（用于批量处理，支持v2c和v3）
*/
@Data
public static class DeviceConfig {
private String switchIp;
private String community;
private int port;
private String clientId;
private List<CollectVo> collectVos;
private SwitchOidVo switchOidVo;
// SNMP v3特有字段
private int version = SnmpConstants.version2c; // 默认v2c
private String securityName;
private String authProtocol;
private String authPassword;
private String privProtocol;
private String privPassword;
private int securityLevel = SecurityLevel.AUTH_PRIV; // 默认认证+加密
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/snmp/service/ProcessSwitchCollectDataService.java
+310
View File
@@ -0,0 +1,310 @@
package com.ruoyi.rocketmq.snmp.service;
import com.alibaba.fastjson.JSONObject;
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.utils.DateUtils;
import com.ruoyi.rocketmq.domain.*;
import com.ruoyi.rocketmq.domain.vo.CollectDataVo;
import com.ruoyi.rocketmq.service.*;
import com.ruoyi.rocketmq.snmp.dto.CollectionResult;
import com.ruoyi.rocketmq.utils.SwitchJsonDataParser;
import com.ruoyi.system.api.RemoteRevenueConfigService;
import com.ruoyi.system.api.domain.InitialSwitchInfoDetailsRemote;
import com.ruoyi.system.api.domain.RmResourceRegistrationRemote;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
@Service
@Slf4j
public class ProcessSwitchCollectDataService {
@Autowired
private RemoteRevenueConfigService remoteRevenueConfigService;
@Autowired
private IInitialSwitchInfoService initialSwitchInfoService;
@Autowired
private IInitialSwitchInfoTempService initialSwitchInfoTempService;
@Autowired
private IInitialSwitchPowerSupplyService initialSwitchPowerSupplyService;
@Autowired
private IInitialSwitchFanInfoService initialSwitchFanInfoService;
@Autowired
private IInitialSwitchMpuInfoService initialSwitchMpuInfoService;
@Autowired
private IInitialSwitchOpticalModuleService initialSwitchOpticalModuleService;
@Autowired
private IInitialSwitchOtherCollectDataService insertInitialSwitchOtherInfo;
// 在类中添加
private static final ObjectMapper objectMapper = new ObjectMapper();
/**
* 交换机所有数据入库
* @param result
*/
public void handleSwitchDataMessage(CollectionResult result, String clientId) {
if(result != null){
List<Map<String, String>> resultData = result.getData();
CollectDataVo switchDataVo = new CollectDataVo();
switchDataVo.setTimestamp(result.getTimestamp());
switchDataVo.setValue(JSONObject.toJSONString(resultData));
switch(result.getDeviceType()){
case "net":
// handleSwitchNetMessage(switchDataVo, clientId);
break;
case "pwr":
handleSwitchPwrMessage(switchDataVo, clientId);
break;
case "module":
handleSwitchModuleMessage(switchDataVo, clientId);
break;
case "mpu":
handleSwitchMpuMessage(switchDataVo, clientId);
break;
case "fan":
handleSwitchFanMessage(switchDataVo, clientId);
break;
default:
handleSwitchOtherMessage(switchDataVo, clientId);
break;
}
}else{
throw new RuntimeException("交换机data数据为空");
}
}
/**
* 其他发现数据（默认处理）
* @param switchDataVo
*/
private void handleSwitchOtherMessage(CollectDataVo switchDataVo, String clientId) {
if (switchDataVo != null) {
try {
InitialSwitchOtherCollectData insertData = new InitialSwitchOtherCollectData();
// 设置基本信息
long timestamp = switchDataVo.getTimestamp();
Date createTime = new Date(timestamp * 1000 / 1000 * 1000);
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
String value = switchDataVo.getValue();
JsonNode jsonNode = objectMapper.readTree(value);
// 处理数组中的字符串JSON
if (jsonNode.isArray() && jsonNode.size() > 0) {
JsonNode firstElement = jsonNode.get(0);
if (firstElement.isTextual()) {
// 二次解析JSON字符串
JsonNode innerJsonNode = objectMapper.readTree(firstElement.asText());
if (innerJsonNode.isObject()) {
Iterator<Map.Entry<String, JsonNode>> fields = innerJsonNode.fields();
if (fields.hasNext()) {
Map.Entry<String, JsonNode> entry = fields.next();
String fieldName = entry.getKey();
String fieldValue = entry.getValue().asText();
insertData.setCollectType(fieldName);
if (!"null".equals(fieldValue)) {
insertData.setCollectValue(fieldValue);
// insertInitialSwitchOtherInfo.insertInitialSwitchOtherCollectData(insertData);
}
}
}
}else{
if (firstElement.isObject()) {
Iterator<Map.Entry<String, JsonNode>> fields = firstElement.fields();
if (fields.hasNext()) {
Map.Entry<String, JsonNode> entry = fields.next();
String fieldName = entry.getKey();
String fieldValue = entry.getValue().asText();
insertData.setCollectType(fieldName);
if (!"null".equals(fieldValue)) {
insertData.setCollectValue(fieldValue);
System.out.println("1");
// insertInitialSwitchOtherInfo.insertInitialSwitchOtherCollectData(insertData);
}
}
}
}
}
} catch (Exception e) {
log.error("解析JSON数据失败: {}, value: {}", e.getMessage(), switchDataVo.getValue(), e);
}
}
}
/**
* 交换机网卡信息数据入库
* @param switchDataVo
*/
private void handleSwitchNetMessage(CollectDataVo switchDataVo, String clientId) {
List<InitialSwitchInfo> switchInfos = SwitchJsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchInfo.class);
if(!switchInfos.isEmpty()){
// 根据clientId查询交换机ip
RmResourceRegistrationRemote queryParam = new RmResourceRegistrationRemote();
queryParam.setHardwareSn(clientId);
R<RmResourceRegistrationRemote> registMsgR = remoteRevenueConfigService.getListByHardwareSn(queryParam, SecurityConstants.INNER);
if(registMsgR != null){
RmResourceRegistrationRemote registMsg = registMsgR.getData();
}
// 时间戳转换
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);
if(registMsgR != null && registMsgR.getData()!=null && registMsgR.getData().getSnmpCollectAddr()!=null){
switchInfo.setSwitchIp(registMsgR.getData().getSnmpCollectAddr());
}
InitialSwitchInfoTemp tempInfo = tempMap.get(switchInfo.getName());
if (tempInfo != null) {
// 计算inSpeed
if (switchInfo.getInBytes() != null && tempInfo.getInBytes() != null) {
BigDecimal inDiff = switchInfo.getInBytes().subtract(tempInfo.getInBytes());
// 字节转为bit
BigDecimal inDiffBit = inDiff.multiply(new BigDecimal(8)).setScale(0, RoundingMode.HALF_UP);
switchInfo.setInSpeed(inDiffBit.divide(divisor, 2, RoundingMode.HALF_UP));
}
// 计算outSpeed
if (switchInfo.getOutBytes() != null && tempInfo.getOutBytes() != null) {
BigDecimal outDiff = switchInfo.getOutBytes().subtract(tempInfo.getOutBytes());
// 字节转为bit
BigDecimal outDiffBit = outDiff.multiply(new BigDecimal(8)).setScale(0, RoundingMode.HALF_UP);
switchInfo.setOutSpeed(outDiffBit.divide(divisor, 2, RoundingMode.HALF_UP));
}
}
});
}else{
switchInfos.forEach(switchInfo -> {
switchInfo.setClientId(clientId);
switchInfo.setCreateTime(createTime);
if(registMsgR != null && registMsgR.getData()!=null && registMsgR.getData().getSnmpCollectAddr()!=null){
switchInfo.setSwitchIp(registMsgR.getData().getSnmpCollectAddr());
}
});
}
// 清空临时表对应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 switchDataVo
*/
private void handleSwitchPwrMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchPowerSupply> powerSupplyList = SwitchJsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchPowerSupply.class);
if (!powerSupplyList.isEmpty()){
for (InitialSwitchPowerSupply insertData : powerSupplyList) {
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
}
initialSwitchPowerSupplyService.insertBatchInitialSwitchPowerSupply(powerSupplyList);
}
}
/**
* 光模块发现数据
* @param switchDataVo
*/
private void handleSwitchModuleMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchOpticalModule> moduleList = SwitchJsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchOpticalModule.class);
if (!moduleList.isEmpty()){
for (InitialSwitchOpticalModule insertData : moduleList) {
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
}
initialSwitchOpticalModuleService.batchInitialSwitchOpticalModule(moduleList);
}
}
/**
* MPU发现数据
* @param switchDataVo
*/
private void handleSwitchMpuMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchMpuInfo> mpuList = SwitchJsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchMpuInfo.class);
if (!mpuList.isEmpty()){
for (InitialSwitchMpuInfo insertData : mpuList) {
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
}
initialSwitchMpuInfoService.insertBatchInitialSwitchMpuInfo(mpuList);
}
}
/**
* 风扇发现数据
* @param switchDataVo
*/
private void handleSwitchFanMessage(CollectDataVo switchDataVo, String clientId){
List<InitialSwitchFanInfo> fanList = SwitchJsonDataParser.parseJsonData(switchDataVo.getValue(), InitialSwitchFanInfo.class);
if (!fanList.isEmpty()){
for (InitialSwitchFanInfo insertData : fanList) {
// 时间戳转换
long timestamp = switchDataVo.getTimestamp();
long millis = timestamp * 1000;
Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
insertData.setClientId(clientId);
insertData.setCreateTime(createTime);
}
initialSwitchFanInfoService.insertBatchInitialSwitchFanInfo(fanList);
}
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/DataProcessUtil.java
+42
View File
@@ -10,6 +10,7 @@ import com.ruoyi.rocketmq.mapper.RmTemplateSwitchMapper;
import com.ruoyi.system.api.RemoteRevenueConfigService;
import com.ruoyi.system.api.domain.RmResourceGroupRemote;
import com.ruoyi.system.api.domain.RmResourceRegistrationRemote;
import com.ruoyi.system.api.domain.RmSwitchManagementRemote;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -30,6 +31,26 @@ public class DataProcessUtil {
@Autowired
private RemoteRevenueConfigService remoteRevenueConfigService;
/**
* 根据clinetId获取交换机名称
* @param clientId
* @return
*/
public String getDeviceNameByClientId(String clientId){
try {
RmSwitchManagementRemote queryParam = new RmSwitchManagementRemote();
queryParam.setClientId(clientId);
R<List<RmSwitchManagementRemote>> switchMsg = remoteRevenueConfigService
.getSwitchNameByClientId(queryParam, SecurityConstants.INNER);
if (switchMsg != null && switchMsg.getData() != null) {
return switchMsg.getData().get(0).getClientId();
}
} catch (Exception e) {
log.error("获取交换机名称信息失败，clientId: {}", clientId, e);
}
return null;
}
/**
* 补充资源组信息
*/
@@ -219,4 +240,25 @@ public class DataProcessUtil {
}
}
}
/**
* 根据clientId查询交换机信息
* @param clientId
* @return
*/
public RmSwitchManagementRemote getSwitchMsg(String clientId) {
try {
RmSwitchManagementRemote queryParam = new RmSwitchManagementRemote();
queryParam.setClientId(clientId);
R<List<RmSwitchManagementRemote>> switchMsg = remoteRevenueConfigService
.getSwitchNameByClientId(queryParam, SecurityConstants.INNER);
if (switchMsg != null && switchMsg.getData() != null) {
return switchMsg.getData().get(0);
}
} catch (Exception e) {
log.error("获取交换机名称信息失败，clientId: {}", clientId, e);
}
return null;
}
}