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

获取最新策略增加路由信息方法、交换机带宽收益

Browse Source
This commit is contained in:
gaoyutao
2025-10-27 18:16:47 +08:00
parent 36362e2236
commit 46bf4c4114
19 changed files with 376 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/MessageHandler.java
View File

@@ -8,6 +8,7 @@ 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.RegisterMsgVo;
import com.ruoyi.rocketmq.domain.vo.RspResultVo;
import com.ruoyi.rocketmq.domain.vo.RspVo;
import com.ruoyi.rocketmq.service.*;
import com.ruoyi.rocketmq.utils.DataProcessUtil;
@@ -91,18 +92,18 @@ public class MessageHandler {
@PostConstruct
public void init() {
registerHandler(MsgEnum.执行脚本策略应答.getValue(), this::handleScriptRspMessage);
// registerHandler(MsgEnum.Agent版本更新应答.getValue(), this::handleAgentUpdateRspMessage);
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.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);
}
@@ -116,9 +117,11 @@ public class MessageHandler {
RegisterMsgVo registerMsgVo = interfaces.get(0);
String clientId = registerMsgVo.getClientId();
// 如果未下发监控策略，下发
// rmMonitorPolicyService.issuePolicyMsgByClientId(clientId);
rmMonitorPolicyService.issuePolicyMsgByClientId(clientId);
// 如果未下发服务器脚本策略，下发
rmDeploymentPolicyService.issueDeployPolicyMsgByClientId(clientId);
// 如果路由有变化，更新路由信息
rmNetworkInterfaceService.updateRouteMsg(clientId);
}
}
@@ -150,6 +153,7 @@ public class MessageHandler {
// 查询该网卡信息是否存在
RmNetworkInterface queryParam = new RmNetworkInterface();
queryParam.setMacAddress(networkInfo.getMac());
queryParam.setNewFlag(1);
List<RmNetworkInterface> exits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(queryParam);
if(exits.isEmpty()){
// 保存网卡信息
@@ -169,18 +173,64 @@ public class MessageHandler {
rmNetworkInterfaceService.insertRmNetworkInterface(insertData);
}else{
RmNetworkInterface oldInterfaceMsg = exits.get(0);
if(!StringUtils.equals(networkInfo.getPublicIp(),oldInterfaceMsg.getPublicIp())){
if(!StringUtils.equals(networkInfo.getName(),oldInterfaceMsg.getInterfaceName())
|| !StringUtils.equals(networkInfo.getGateway(),oldInterfaceMsg.getGateway())){
// 查询该网卡信息是否旧数据
RmNetworkInterface query = new RmNetworkInterface();
query.setMacAddress(networkInfo.getMac());
query.setNewFlag(999);
List<RmNetworkInterface> oldExits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(queryParam);
if(!oldExits.isEmpty()){
// 先删除旧数据
oldExits.forEach(oldMsg ->{
rmNetworkInterfaceService.deleteRmNetworkInterfaceById(oldMsg.getId());
});
}
// 先将已存在的改为旧数据
RmNetworkInterface oldData = new RmNetworkInterface();
oldData.setNewFlag(0);
oldData.setMacAddress(networkInfo.getMac());
rmNetworkInterfaceService.updateRmNetworkInterfaceByMac(oldData);
RmNetworkInterface insertData = new RmNetworkInterface();
insertData.setMacAddress(networkInfo.getMac());
insertData.setIsp(networkInfo.getCarrier());
insertData.setCity(networkInfo.getCity());
insertData.setGateway(networkInfo.getGateway());
insertData.setInterfaceName(networkInfo.getName());
insertData.setIpv4Address(networkInfo.getIpv4());
insertData.setMacAddress(networkInfo.getMac());
insertData.setProvince(networkInfo.getProvince());
insertData.setPublicIp(networkInfo.getPublicIp());
rmNetworkInterfaceService.insertRmNetworkInterface(insertData);
}else {
RmNetworkInterface updateData = new RmNetworkInterface();
updateData.setMacAddress(networkInfo.getMac());
updateData.setIsp(networkInfo.getCarrier());
updateData.setCity(networkInfo.getCity());
updateData.setGateway(networkInfo.getGateway());
updateData.setInterfaceName(networkInfo.getName());
updateData.setIpv4Address(networkInfo.getIpv4());
updateData.setMacAddress(networkInfo.getMac());
updateData.setProvince(networkInfo.getProvince());
updateData.setPublicIp(networkInfo.getPublicIp());
rmNetworkInterfaceService.updateRmNetworkInterfaceByMac(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.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 (needUpdate) {
rmNetworkInterfaceService.updateRmNetworkInterfaceByMac(updateData);
}
}
}
}else{
@@ -206,25 +256,71 @@ public class MessageHandler {
rmNetworkInterfaceService.insertRmNetworkInterface(insertData);
}else{
RmNetworkInterface oldInterfaceMsg = exits.get(0);
RmNetworkInterface updateData = new RmNetworkInterface();
updateData.setMacAddress(networkInfo.getMac());
updateData.setIsp(networkInfo.getCarrier());
updateData.setCity(networkInfo.getCity());
updateData.setGateway(networkInfo.getGateway());
updateData.setInterfaceName(networkInfo.getName());
updateData.setIpv4Address(networkInfo.getIpv4());
updateData.setMacAddress(networkInfo.getMac());
updateData.setProvince(networkInfo.getProvince());
updateData.setPublicIp(networkInfo.getPublicIp());
if(!StringUtils.equals(networkInfo.getPublicIp(),oldInterfaceMsg.getPublicIp())){
updateData.setBindIp("0");
// 修改绑定公网ip状态
RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote();
updateParam.setClientId(registerMsg.getClientId());
updateParam.setMultiPublicIpStatus("0");
remoteRevenueConfigService.updateStatusByResource(updateParam, SecurityConstants.INNER);
if(!StringUtils.equals(networkInfo.getName(),oldInterfaceMsg.getInterfaceName())
|| !StringUtils.equals(networkInfo.getGateway(),oldInterfaceMsg.getGateway())){
// 查询该网卡信息是否旧数据
RmNetworkInterface query = new RmNetworkInterface();
query.setMacAddress(networkInfo.getMac());
query.setNewFlag(999);
List<RmNetworkInterface> oldExits = rmNetworkInterfaceService.selectRmNetworkInterfaceList(queryParam);
if(!oldExits.isEmpty()){
// 先删除旧数据
oldExits.forEach(oldMsg ->{
rmNetworkInterfaceService.deleteRmNetworkInterfaceById(oldMsg.getId());
});
}
// 先将已存在的改为旧数据
RmNetworkInterface oldData = new RmNetworkInterface();
oldData.setNewFlag(0);
oldData.setMacAddress(networkInfo.getMac());
rmNetworkInterfaceService.updateRmNetworkInterfaceByMac(oldData);
RmNetworkInterface insertData = new RmNetworkInterface();
insertData.setMacAddress(networkInfo.getMac());
insertData.setIsp(networkInfo.getCarrier());
insertData.setCity(networkInfo.getCity());
insertData.setGateway(networkInfo.getGateway());
insertData.setInterfaceName(networkInfo.getName());
insertData.setIpv4Address(networkInfo.getIpv4());
insertData.setMacAddress(networkInfo.getMac());
insertData.setProvince(networkInfo.getProvince());
insertData.setPublicIp(networkInfo.getPublicIp());
rmNetworkInterfaceService.insertRmNetworkInterface(insertData);
}else {
RmNetworkInterface updateData = new RmNetworkInterface();
// 检查其他字段是否需要更新
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.getProvince(), oldInterfaceMsg.getProvince())) {
updateData.setProvince(networkInfo.getProvince());
needUpdate = true;
}
if (!StringUtils.equals(networkInfo.getPublicIp(), oldInterfaceMsg.getPublicIp())) {
updateData.setPublicIp(networkInfo.getPublicIp());
needUpdate = true;
updateData.setBindIp("0");
// 修改绑定公网ip状态
RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote();
updateParam.setClientId(registerMsg.getClientId());
updateParam.setMultiPublicIpStatus("0");
remoteRevenueConfigService.updateStatusByResource(updateParam, SecurityConstants.INNER);
}
if (!StringUtils.equals(networkInfo.getCarrier(), oldInterfaceMsg.getIsp())) {
updateData.setIsp(networkInfo.getCarrier());
needUpdate = true;
}
// 只有有字段变化时才执行更新
if (needUpdate) {
rmNetworkInterfaceService.updateRmNetworkInterfaceByMac(updateData);
}
}
rmNetworkInterfaceService.updateRmNetworkInterfaceByMac(updateData);
}
}
}
@@ -243,13 +339,15 @@ public class MessageHandler {
RspVo rsp = rspVoList.get(0);
if(rsp.getResCode() == 1){
RmAgentManagement rmAgentManagement = new RmAgentManagement();
rmAgentManagement.setHardwareSn(message.getClientId());
rmAgentManagement.setClientId(message.getClientId());
rmAgentManagement.setLastUpdateResult("1");
rmAgentManagement.setLastUpdateTime(DateUtils.getNowDate());
rmAgentManagementService.updateRmAgentManagementByHardwareSn(rmAgentManagement);
}else{
RmAgentManagement rmAgentManagement = new RmAgentManagement();
rmAgentManagement.setHardwareSn(message.getClientId());
rmAgentManagement.setClientId(message.getClientId());
rmAgentManagement.setLastUpdateResult("0");
rmAgentManagement.setLastUpdateTime(DateUtils.getNowDate());
rmAgentManagementService.updateRmAgentManagementByHardwareSn(rmAgentManagement);
}
}
@@ -462,12 +560,29 @@ public class MessageHandler {
if(!heartbeats.isEmpty()){
InitialHeartbeatListen heartbeat = heartbeats.get(0);
String clientId = message.getClientId();
String version = heartbeat.getVersion();
log.debug("处理心跳消息，客户端ID: {}, 时间: {}", clientId, heartbeat.getTimestamp());
// 添加逻辑节点标识
RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote();
updateData.setClientId(message.getClientId());
updateData.setLogicalNodeId(heartbeat.getLogicalNode());
remoteRevenueConfigService.innerUpdateRegist(updateData, SecurityConstants.INNER);
// agent更新结果存储
RmAgentManagement query = new RmAgentManagement();
query.setClientId(clientId);
List<RmAgentManagement> 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);
}
}
// 使用Redis存储状态
String statusKey = HEARTBEAT_STATUS_PREFIX + clientId;
String timeKey = HEARTBEAT_TIME_PREFIX + clientId;
@@ -659,9 +774,12 @@ public class MessageHandler {
RmResourceRegistrationRemote resourceMsgData = resourceMsg.getData();
if(rsp.getResCode() == 1){
if(rsp.getResult() != null){
List<RspResultVo> resultVos = JsonDataParser.parseJsonData(rsp.getResult(), RspResultVo.class);
RspResultVo rspResultVo = resultVos.get(0);
// 构建脚本执行结果实体类
RmResourceRemote insertData = new RmResourceRemote();
insertData.setClientId(message.getClientId());
insertData.setScriptId(Long.valueOf(rspResultVo.getScriptId()));
insertData.setResultFlag(1);
insertData.setDescription(rsp.getResult());
insertData.setCreateTime(createTime);
@@ -675,7 +793,10 @@ public class MessageHandler {
insertData.setClientId(message.getClientId());
insertData.setResultFlag(0);
if(rsp.getResult() != null){
List<RspResultVo> 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