From 68dedde23d3f5b6d913917d410107c14c7845ca5 Mon Sep 17 00:00:00 2001 From: baoqm <976302356@qq.com> Date: Sat, 23 Aug 2025 23:52:33 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=BE=E7=BD=AE=E9=87=87=E9=9B=86=E9=97=B4?= =?UTF-8?q?=E9=9A=94=E5=88=9D=E5=A7=8B=E5=8C=96=E9=BB=98=E8=AE=A4=E5=80=BC?= =?UTF-8?q?5=E5=88=86=E9=92=9F=20=E5=A2=9E=E5=8A=A0=E6=8C=87=E4=BB=A4?= =?UTF-8?q?=E6=9B=B4=E6=96=B0=E9=87=87=E9=9B=86=E9=97=B4=E9=9A=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../agentserver/core/enums/MsgEnum.java | 36 ++++++++-- .../agentserver/scheduler/CpuScheduler.java | 2 +- .../scheduler/DockerScheduler.java | 5 +- .../agentserver/scheduler/NetScheduler.java | 2 +- .../scheduler/SwitchBoardScheduler.java | 2 +- .../server/collect/config/GlobalConfig.java | 67 +++++++++++++++++++ .../impl/SwitchBoardServiceImpl.java | 12 ++-- .../server/netty/NettyTcpClient.java | 4 +- .../netty/basics/AgentDispatcherManager.java | 46 +++++++++++++ .../server/netty/handler/DecoderHandler.java | 41 +++++++----- .../server/netty/handler/EncoderHandler.java | 41 ++++++++++++ 11 files changed, 222 insertions(+), 36 deletions(-) create mode 100644 src/main/java/com/tongran/agentserver/server/collect/config/GlobalConfig.java create mode 100644 src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java create mode 100644 src/main/java/com/tongran/agentserver/server/netty/handler/EncoderHandler.java diff --git a/src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java b/src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java index e4b4a52..62b85c3 100644 --- a/src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java +++ b/src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java @@ -12,12 +12,40 @@ import lombok.NoArgsConstructor; @NoArgsConstructor public enum MsgEnum { - 登录认证(0x01, 6, "0000"); + 更新CPU采集间隔("TIME_CPU"), - private Integer value; + 更新CPU采集间隔应答("TIME_CPU_RSP"), + + 更新容器采集间隔("TIME_DOCKER"), + + 更新容器采集间隔应答("TIME_DOCKER_RSP"), + + 更新网卡采集间隔("TIME_NET"), + + 更新网卡采集间隔应答("TIME_NET_RSP"), + + 更新交换机采集间隔("TIME_SWITCH"), + + 更新交换机采集间隔应答("TIME_SWITCH_RSP"), + + 更新磁盘采集间隔("TIME_DISK"), + + 更新磁盘采集间隔应答("TIME_DISK_RSP"), + + 更新挂载采集间隔("TIME_POINT"), + + 更新挂载采集间隔应答("TIME_POINT_RSP"), + + 更新内存采集间隔("TIME_MEMORY"), + + 更新内存采集间隔应答("TIME_MEMORY_RSP"), + + 更新系统采集间隔("TIME_SYSTEM"), + + 更新系统采集间隔应答("TIME_SYSTEM_RSP"); + + private String value; - private Integer start; - private String serialNumber; } diff --git a/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java index 8052bfa..1866fc9 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java @@ -39,7 +39,7 @@ public class CpuScheduler { } catch (InterruptedException e) { e.printStackTrace(); } - startTask(180000); // 默认3分钟间隔启动 + startTask(300000); // 默认3分钟间隔启动 } public void startTask(long intervalMillis) { diff --git a/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java index b0d4146..f686982 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java @@ -12,6 +12,7 @@ import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.stereotype.Component; +import javax.annotation.PostConstruct; import javax.annotation.Resource; import java.time.LocalDateTime; import java.util.List; @@ -32,7 +33,7 @@ public class DockerScheduler { @Resource private DockerService dockerService; -// @PostConstruct + @PostConstruct public void init() { // 等待42秒 try { @@ -40,7 +41,7 @@ public class DockerScheduler { } catch (InterruptedException e) { e.printStackTrace(); } - startTask(180000); // 默认180秒间隔启动 + startTask(300000); // 默认180秒间隔启动 } public void startTask(long intervalMillis) { diff --git a/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java index 6e55fb9..5b2f81b 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java @@ -41,7 +41,7 @@ public class NetScheduler { } catch (InterruptedException e) { e.printStackTrace(); } - startTask(180000); // 默认180秒间隔启动 + startTask(300000); // 默认180秒间隔启动 } public void startTask(long intervalMillis) { diff --git a/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java index ae90c88..b9fd5de 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java @@ -40,7 +40,7 @@ public class SwitchBoardScheduler { } catch (InterruptedException e) { e.printStackTrace(); } - startTask(80000); // 默认3分钟间隔启动 + startTask(300000); // 默认3分钟间隔启动 } public void startTask(long intervalMillis) { diff --git a/src/main/java/com/tongran/agentserver/server/collect/config/GlobalConfig.java b/src/main/java/com/tongran/agentserver/server/collect/config/GlobalConfig.java new file mode 100644 index 0000000..db927d7 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/server/collect/config/GlobalConfig.java @@ -0,0 +1,67 @@ +package com.tongran.agentserver.server.collect.config; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Properties; + +/** + * 全局配置类 + */ +public class GlobalConfig { + + // 定义全局变量 + public static String COMMUNITY; + public static String SWITCH_IP; + public static int PORT; + + // 静态代码块，在类加载时执行 + static { + loadConfigFromFile(); + } + + /** + * 从配置文件加载配置 + */ + private static void loadConfigFromFile() { + Properties props = new Properties(); + + try (InputStream input = Files.newInputStream(Paths.get("/data/agent-server/config/switch_config.txt"))) { + // 加载配置文件 + props.load(input); + + // 为全局变量赋值 + COMMUNITY = props.getProperty("switch.core-switch.community", "public"); + SWITCH_IP = props.getProperty("witch.core-switch.ip", "127.0.0.1"); + PORT = Integer.parseInt(props.getProperty("switch.core-switch.port", "161")); + + + System.out.println("配置文件加载成功"); + + } catch (IOException e) { + System.err.println("无法加载配置文件，使用默认值"); + setDefaultValues(); + } catch (NumberFormatException e) { + System.err.println("配置文件格式错误: " + e.getMessage()); + setDefaultValues(); + } + } + + /** + * 设置默认值 + */ + private static void setDefaultValues() { + COMMUNITY = "public"; + SWITCH_IP = "127.0.0.1"; + PORT = 161; + } + + /** + * 重新加载配置（可选） + */ + public static void reloadConfig() { + loadConfigFromFile(); + } + +} diff --git a/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java index 8ca140b..dae6f02 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java +++ b/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java @@ -1,5 +1,6 @@ package com.tongran.agentserver.server.collect.switchboard.impl; +import com.tongran.agentserver.server.collect.config.GlobalConfig; import com.tongran.agentserver.server.collect.switchboard.SwitchBoardService; import com.tongran.agentserver.server.collect.vo.SwitchBoardVO; import org.snmp4j.*; @@ -18,15 +19,10 @@ import java.util.List; @Service public class SwitchBoardServiceImpl implements SwitchBoardService { - private static final String COMMUNITY = "sjz_huiri"; // 默认community字符串 - private static final String SWITCH_IP = "172.16.15.2"; // 交换机IP - private static final int PORT = 161; // SNMP端口 - @Override public List query() { List list = new ArrayList<>(); - SwitchBoardVO boardVO = SwitchBoardVO.builder().build(); - System.out.println("1111111111==================== 交换机流量信息 ===================="); + System.out.println("==================== 交换机流量信息 ===================="); try { // 1. 创建传输映射 TransportMapping transport = new DefaultUdpTransportMapping(); @@ -35,8 +31,8 @@ public class SwitchBoardServiceImpl implements SwitchBoardService { // 2. 创建目标对象 CommunityTarget target = new CommunityTarget(); - target.setCommunity(new OctetString(COMMUNITY)); - target.setAddress(GenericAddress.parse("udp:" + SWITCH_IP + "/" + PORT)); + target.setCommunity(new OctetString(GlobalConfig.COMMUNITY)); + target.setAddress(GenericAddress.parse("udp:" + GlobalConfig.SWITCH_IP + "/" + GlobalConfig.PORT)); target.setRetries(2); target.setTimeout(5000); target.setVersion(SnmpConstants.version2c); diff --git a/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java b/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java index 9342cae..5cb1fee 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java +++ b/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java @@ -2,13 +2,13 @@ package com.tongran.agentserver.server.netty; import com.tongran.agentserver.server.netty.config.AgentNettyConfig; import com.tongran.agentserver.server.netty.handler.DecoderHandler; +import com.tongran.agentserver.server.netty.handler.EncoderHandler; import com.tongran.agentserver.server.netty.handler.NettyClientHandler; import io.netty.bootstrap.Bootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; -import io.netty.handler.codec.string.StringEncoder; import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,7 +60,7 @@ public class NettyTcpClient { ChannelPipeline pipeline = ch.pipeline(); // 添加编解码器 pipeline.addLast("decoder", new DecoderHandler()); - pipeline.addLast("encoder", new StringEncoder()); + pipeline.addLast("encoder", new EncoderHandler()); // 添加心跳机制 pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, 90, TimeUnit.SECONDS)); diff --git a/src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java b/src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java new file mode 100644 index 0000000..58cc4ab --- /dev/null +++ b/src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java @@ -0,0 +1,46 @@ +package com.tongran.agentserver.server.netty.basics; + +import com.tongran.agentserver.server.netty.annotation.AgentDispatcher; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +@Component +public class AgentDispatcherManager implements ApplicationContextAware { + + /** + * 所有实现的包处理器 + */ + private static Map MSG_HANDLER_MAP; + + /** + * 唤醒时 初始化 packHandlerMap + */ + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + // 仅一次性初始化完成 + if (MSG_HANDLER_MAP == null) { + MSG_HANDLER_MAP = new ConcurrentHashMap<>(); + Map handlers = applicationContext.getBeansWithAnnotation(AgentDispatcher.class); + if (!CollectionUtils.isEmpty(handlers)) { + handlers.values().forEach(tempHandler -> { + boolean result = tempHandler.getClass().isAnnotationPresent(AgentDispatcher.class); + if (result) { + AgentDispatcher annotation = tempHandler.getClass().getAnnotation(AgentDispatcher.class); + MSG_HANDLER_MAP.put(annotation.msgId().getValue() + "&" + annotation.version().value, (AgentHandler) tempHandler); + } + }); + } + } + } + + public AgentHandler getHandler(String msgIdVersion) { + return MSG_HANDLER_MAP == null ? null : MSG_HANDLER_MAP.get(msgIdVersion); + } + +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java b/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java index 3b0e864..9264a57 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java +++ b/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java @@ -1,7 +1,11 @@ package com.tongran.agentserver.server.netty.handler; -import cn.hutool.cache.Cache; -import cn.hutool.cache.CacheUtil; +import com.alibaba.fastjson2.JSONObject; +import com.tongran.agentserver.server.netty.annotation.AgentDispatcher; +import com.tongran.agentserver.server.netty.basics.AgentDispatcherManager; +import com.tongran.agentserver.server.netty.basics.AgentHandler; +import com.tongran.agentserver.server.netty.model.Message; +import com.tongran.agentserver.server.netty.model.UpMsgResponse; import com.tongran.agentserver.utils.AssertLog; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandler; @@ -10,14 +14,14 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.util.CharsetUtil; import org.springframework.stereotype.Component; +import javax.annotation.Resource; + @Component @ChannelHandler.Sharable public class DecoderHandler extends ChannelInboundHandlerAdapter { - - - // 用来临时保留没有处理过的请求报文 - private final Cache lruCache = CacheUtil.newLRUCache(3000); + @Resource + private AgentDispatcherManager agentDispatcherManager; /** * 消息解码器 @@ -29,17 +33,20 @@ public class DecoderHandler extends ChannelInboundHandlerAdapter { ByteBuf byteBuf = (ByteBuf) msg; String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码 AssertLog.info("<<[up]:[up-content]==>{}", messages); - -// JSONObject jsonObject = JSONObject.parseObject(messages); -// String clientId = jsonObject.getString("clientId"); -// Message electricMessage = Message.builder().build(); -// electricMessage.setClientId(clientId); -// JSONObject object = new JSONObject(); -// object.put("dateType","register"); -// object.put("stats","1"); -// electricMessage.setContent(object.toString()); -// ctx.fireChannelRead(electricMessage);//传递到下一个handler - + boolean startsWith = messages.startsWith("agent-server:"); + boolean endsWith = messages.endsWith("@tong-ran"); + if(startsWith && endsWith){ + JSONObject jsonObject = JSONObject.parseObject(messages); + String clientId = jsonObject.getString("clientId"); + String dataType = jsonObject.getString("dataType"); + AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value); + UpMsgResponse response = msgHandler.upHandle(messages,clientId); + Message agentMessage = Message.builder().build(); + agentMessage.setClientId(clientId); + agentMessage.setDataType(response.getDataType()); + agentMessage.setData(response.getContent()); + ctx.fireChannelRead(agentMessage);//传递到下一个handler + } byteBuf.release(); // 释放 ByteBuf 资源（重要！） } else { System.out.println("Unexpected message type: " + msg.getClass()); diff --git a/src/main/java/com/tongran/agentserver/server/netty/handler/EncoderHandler.java b/src/main/java/com/tongran/agentserver/server/netty/handler/EncoderHandler.java new file mode 100644 index 0000000..012e678 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/server/netty/handler/EncoderHandler.java @@ -0,0 +1,41 @@ +package com.tongran.agentserver.server.netty.handler; + +import com.alibaba.fastjson2.JSON; +import com.tongran.agentserver.server.netty.model.Message; +import com.tongran.agentserver.utils.AssertLog; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Component; + +import java.nio.charset.StandardCharsets; + +@Component +@ChannelHandler.Sharable +public class EncoderHandler extends ChannelOutboundHandlerAdapter { + + /** + * 消息解码器 + */ + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + if (msg instanceof Message) { + Message entity = (Message) msg; + if (StringUtils.isBlank(entity.getData())) { + AssertLog.error(">>[down]:errorContent:{}", msg); + return; + } + AssertLog.info(">>[down]:[content]==>{}", entity.getData()); + String json = "agent-tcp:"+JSON.toJSONString(entity)+"@tong-ran"; + byte[] bytes = json.getBytes(StandardCharsets.UTF_8); // 显式指定 UTF-8 +// byte[] bytes = EscapeUtil.hexStringToByteArray(entity.getContent()); + ByteBuf buf = Unpooled.wrappedBuffer(bytes); + ctx.write(buf, promise); + } + } + +}