Agent/agent-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

设置采集间隔初始化默认值5分钟

Browse Source 
增加指令更新采集间隔
This commit is contained in:
baoqm
2025-08-23 23:52:33 +08:00
parent c926ce7db8
commit 68dedde23d
11 changed files with 222 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java
+2 -2
View File
@@ -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));
src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java
+46
View File
@@ -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<String, AgentHandler> MSG_HANDLER_MAP;
/**
* 唤醒时 初始化 packHandlerMap
*/
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
// 仅一次性初始化完成
if (MSG_HANDLER_MAP == null) {
MSG_HANDLER_MAP = new ConcurrentHashMap<>();
Map<String, Object> 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);
}
}
src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java
+24 -17
View File
@@ -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<String, ByteBuf> 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());
src/main/java/com/tongran/agentserver/server/netty/handler/EncoderHandler.java
+41
View File
@@ -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);
}
}
}