区分配置文件

.gitignore

@@ -0,0 +1,46 @@
+######################################################################
+# Build Tools
+
+.gradle
+/build/
+!gradle/wrapper/gradle-wrapper.jar
+
+target/
+!.mvn/wrapper/maven-wrapper.jar
+
+######################################################################
+# IDE
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### JRebel ###
+rebel.xml
+### NetBeans ###
+nbproject/private/
+build/*
+nbbuild/
+dist/
+nbdist/
+.nb-gradle/
+
+######################################################################
+# Others
+*.log
+*.xml.versionsBackup
+*.swp
+
+!*/build/*.java
+!*/build/*.html
+!*/build/*.xml

src/main/java/com/tongran/agent/server/controller/TcpService.java

@@ -1,24 +1,13 @@
 package com.tongran.agent.server.controller;
 
-import cn.hutool.core.util.ObjectUtil;
-import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONObject;
-import com.tongran.agent.server.core.enums.MsgEnum;
 import com.tongran.agent.server.core.session.SessionManager;
-import com.tongran.agent.server.netty.MultiTargetNettyClient;
-import com.tongran.agent.server.netty.annotation.AgentDispatcher;
 import com.tongran.agent.server.netty.basics.AgentDispatcherManager;
-import com.tongran.agent.server.netty.basics.AgentHandler;
-import com.tongran.agent.server.netty.model.Message;
 import com.tongran.agent.server.rockermq.AgentProducer;
-import com.tongran.agent.server.rockermq.model.MqMsg;
 import com.tongran.agent.server.utils.AssertLog;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
-import java.util.Objects;
 
 @Slf4j
 @Service
@@ -28,8 +17,8 @@ public class TcpService {
     @Resource
     private AgentDispatcherManager agentDispatcherManager;
 
-    @Resource
+//    @Resource
-    private MultiTargetNettyClient client;
+//    private MultiTargetNettyClient client;
 
     @Resource
     private AgentProducer agentProducer;
@@ -41,88 +30,88 @@ public class TcpService {
     public void create(String clientId){
         boolean success1 = false;
         // 动态创建多个不同IP和端口的连接
-        if(StringUtils.equals(clientId, "clientId007")){
+//        if(StringUtils.equals(clientId, "clientId007")){
-            success1 = client.createConnection(clientId, "127.0.0.1", 6610, 5);
+//            success1 = client.createConnection(clientId, "127.0.0.1", 6610, 5);
-        }else{
+//        }else{
-            success1 = client.createConnection(clientId, "127.0.0.1", 6710, 5);
+//            success1 = client.createConnection(clientId, "127.0.0.1", 6710, 5);
-        }
+//        }
-        System.out.println("当前活跃连接数: " + client.getActiveConnections());
+//        System.out.println("当前活跃连接数: " + client.getActiveConnections());
-        if(success1){
+//        if(success1){
-            // 连接建立后可以发送登录认证消息
+//            // 连接建立后可以发送登录认证消息
-            long timestamp = System.currentTimeMillis();
+//            long timestamp = System.currentTimeMillis();
-            JSONObject object = new JSONObject();
+//            JSONObject object = new JSONObject();
-            object.put("uuid",clientId);
+//            object.put("uuid",clientId);
-            object.put("timestamp",timestamp);
+//            object.put("timestamp",timestamp);
-            Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
+//            Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
-            // 将对象转为 JSON 字符串
+//            // 将对象转为 JSON 字符串
-            String json = JSON.toJSONString(message);
+//            String json = JSON.toJSONString(message);
-            System.out.println("连接建立成功，发送登录认证="+json);
+//            System.out.println("连接建立成功，发送登录认证="+json);
-            client.sendMessages(clientId, message);
+//            client.sendMessages(clientId, message);
-        }
+//        }
     }
 
     public void sendMessage(String message){
         AssertLog.info("consumer==> received down message: {}", message);
-        try {
+//        try {
-            Message dto = JSON.parseObject(message, Message.class);
+//            Message dto = JSON.parseObject(message, Message.class);
-            if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
+//            if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
-                JSONObject jsonObject = JSONObject.parseObject(dto.getData());
+//                JSONObject jsonObject = JSONObject.parseObject(dto.getData());
-                String clientIp = "";
+//                String clientIp = "";
-                int clientPort = 0;
+//                int clientPort = 0;
-                String switchBoard = "";
+//                String switchBoard = "";
-                long timestamp = 0;
+//                long timestamp = 0;
-                if(jsonObject.containsKey("clientIp")){
+//                if(jsonObject.containsKey("clientIp")){
-                    clientIp = jsonObject.getString("clientIp");
+//                    clientIp = jsonObject.getString("clientIp");
-                }
+//                }
-                if(jsonObject.containsKey("clientPort")){
+//                if(jsonObject.containsKey("clientPort")){
-                    clientPort = jsonObject.getInteger("clientPort");
+//                    clientPort = jsonObject.getInteger("clientPort");
-                }
+//                }
-                if(jsonObject.containsKey("switchBoard")){
+//                if(jsonObject.containsKey("switchBoard")){
-                    switchBoard = jsonObject.getString("switchBoard");
+//                    switchBoard = jsonObject.getString("switchBoard");
-                }
+//                }
-                if(jsonObject.containsKey("timestamp")){
+//                if(jsonObject.containsKey("timestamp")){
-                    timestamp = jsonObject.getLongValue("timestamp");
+//                    timestamp = jsonObject.getLongValue("timestamp");
-                }
+//                }
-                if(StringUtils.isBlank(clientIp) || clientPort == 0){
+//                if(StringUtils.isBlank(clientIp) || clientPort == 0){
-                    //返回提醒ip 端口 空
+//                    //返回提醒ip 端口 空
-                    JSONObject json = new JSONObject();
+//                    JSONObject json = new JSONObject();
-                    json.put("clientId",dto.getClientId());
+//                    json.put("clientId",dto.getClientId());
-                    json.put("dataType",MsgEnum.注册应答.getValue());
+//                    json.put("dataType",MsgEnum.注册应答.getValue());
-                    JSONObject dataJson = new JSONObject();
+//                    JSONObject dataJson = new JSONObject();
-                    dataJson.put("resCode",0);
+//                    dataJson.put("resCode",0);
-                    dataJson.put("resMag","客户端IP、端口不能为空");
+//                    dataJson.put("resMag","客户端IP、端口不能为空");
-                    dataJson.put("timestamp", System.currentTimeMillis());
+//                    dataJson.put("timestamp", System.currentTimeMillis());
-                    json.put("data",dataJson.toString());
+//                    json.put("data",dataJson.toString());
-                    MqMsg mqMsg = MqMsg.builder().topic("tongran_agent_down").content(json.toString()).build();
+//                    MqMsg mqMsg = MqMsg.builder().topic("tongran_agent_down").content(json.toString()).build();
-                    agentProducer.asyncSend(mqMsg);
+//                    agentProducer.asyncSend(mqMsg);
-                    return;
+//                    return;
-                }
+//                }
-                boolean success = client.createConnection(dto.getClientId(), clientIp, clientPort, 5);
+//                boolean success = client.createConnection(dto.getClientId(), clientIp, clientPort, 5);
-                if(success){
+//                if(success){
-                    //连接成功
+//                    //连接成功
-                    // 连接建立后可以发送登录认证消息
+//                    // 连接建立后可以发送登录认证消息
-                    JSONObject object = new JSONObject();
+//                    JSONObject object = new JSONObject();
-                    object.put("timestamp", timestamp);
+//                    object.put("timestamp", timestamp);
-                    if(StringUtils.isNotBlank(switchBoard)){
+//                    if(StringUtils.isNotBlank(switchBoard)){
-                        object.put("switchBoard",switchBoard);
+//                        object.put("switchBoard",switchBoard);
-                    }
+//                    }
-                    Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
+//                    Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
-                    // 将对象转为 JSON 字符串
+//                    // 将对象转为 JSON 字符串
-                    client.sendMessages(dto.getClientId(), msg);
+//                    client.sendMessages(dto.getClientId(), msg);
-                }
+//                }
-            } else {
+//            } else {
-                AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
+//                AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
-                if (ObjectUtil.isNotEmpty(msgHandler)) {
+//                if (ObjectUtil.isNotEmpty(msgHandler)) {
-                    String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
+//                    String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
-                    Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
+//                    Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
-                    if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
+//                    if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
-                        sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
+//                        sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
-                    }
+//                    }
-                }
+//                }
-            }
+//            }
-        } catch (Exception e) {
+//        } catch (Exception e) {
-            AssertLog.error("=====rocketmq-exception{}=====" + e.getMessage());
+//            AssertLog.error("=====rocketmq-exception{}=====" + e.getMessage());
-        }
+//        }
     }
 
 

src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java

@@ -11,11 +11,18 @@ import lombok.NoArgsConstructor;
 @AllArgsConstructor
 @NoArgsConstructor
 public enum MsgEnum {
+    建立连接("CONNECT"),
+
+    建立连接应答("CONNECT_RSP"),
 
     注册("REGISTER"),
 
     注册应答("REGISTER_RSP"),
 
+    获取最新策略("GET_POLICY"),
+
+    获取最新策略应答("GET_POLICY_RSP"),
+
     断开("DISCONNECT"),
 
     断开应答("DISCONNECT_RSP"),
@@ -34,6 +41,10 @@ public enum MsgEnum {
 
     网络上报("NET"),
 
+    网络上报重试("NET_RECOVER"),
+
+    业务网络上报("BUSINESS_NET"),
+
     挂载上报("POINT"),
 
     系统其他上报("OTHER_SYSTEM"),
@@ -68,7 +79,19 @@ public enum MsgEnum {
 
     Agent版本更新("AGENT_VERSION_UPDATE"),
 
-    Agent版本更新应答("AGENT_VERSION_UPDATE_RSP");
+    Agent版本更新应答("AGENT_VERSION_UPDATE_RSP"),
+
+    修改frp配置文件应答("UPDATE_FRP_RSP"),
+
+    macvlan状态上报("MACVLAN_STATUS_RSP"),
+
+    内存详情上报("MEMORY_DETAILS"),
+
+    iops结果上报("IOPS_RESULT"),
+
+    tcpdump结果上报("TCPDUMP_RESULT"),
+
+    多网IP探测上报("NETWORK_DETECT");
 
     private String value;
 

src/main/java/com/tongran/agent/server/netty/AgentNettyConfig.java

@@ -0,0 +1,22 @@
+package com.tongran.agent.server.netty;
+
+import com.tongran.agent.server.netty.config.BaseNettyConfig;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Netty配置
+ */
+@Data
+@Configuration
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+@ConfigurationProperties(prefix = "tcp.netty.charge")
+public class AgentNettyConfig extends BaseNettyConfig {
+
+    private static final long serialVersionUID = -4284214721383107291L;
+
+}

src/main/java/com/tongran/agent/server/netty/AgentNettyServer.java

@@ -0,0 +1,56 @@
+package com.tongran.agent.server.netty;
+
+import com.tongran.agent.server.netty.handler.AgentDecoderHandler;
+import com.tongran.agent.server.netty.handler.AgentDispatcherHandler;
+import com.tongran.agent.server.netty.handler.AgentEncoderHandler;
+import com.tongran.agent.server.netty.handler.TCPListenHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.timeout.IdleStateHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.Resource;
+
+@Configuration
+public class AgentNettyServer extends BaseNettyServer {
+
+    @Resource
+    private AgentNettyConfig config;
+
+    @Resource
+    private TCPListenHandler tcpListenHandler;
+
+    @Resource
+    private AgentDecoderHandler decoderHandler;
+
+    @Resource
+    private AgentEncoderHandler encoderHandler;
+
+    @Resource
+    private AgentDispatcherHandler dispatcherHandler;
+
+
+    protected AgentNettyServer(AgentNettyConfig config) {
+        super(config);
+    }
+
+    @Bean(name = "NettyCharge", initMethod = "start", destroyMethod = "stop")
+    BaseNettyServer run() {
+        config.setHander(new ChannelInitializer<NioSocketChannel>() {
+            @Override
+            public void initChannel(NioSocketChannel ch) throws Exception {
+                ch.pipeline().addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime));// 心跳
+                ch.pipeline().addLast(tcpListenHandler);// 监听器
+                //入栈
+                ch.pipeline().addLast(decoderHandler);//解码器
+                //出栈
+                ch.pipeline().addLast(encoderHandler);//加码器
+                // 业务分发是入站最后一个处理器, 出站第一个处理器, 位置要放在最后
+                ch.pipeline().addLast(businessGroup, dispatcherHandler);//业务分发 这里使用业务线程组
+            }
+        });
+        return new AgentNettyServer(config);
+    }
+
+}

src/main/java/com/tongran/agent/server/netty/MultiTargetNettyClient.java

@@ -1,301 +0,0 @@
-package com.tongran.agent.server.netty;
-
-import com.tongran.agent.server.netty.handler.AgentDecoderHandler;
-import com.tongran.agent.server.netty.handler.AgentDispatcherHandler;
-import com.tongran.agent.server.netty.handler.AgentEncoderHandler;
-import com.tongran.agent.server.netty.model.Message;
-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.util.AttributeKey;
-import org.springframework.stereotype.Component;
-
-import javax.annotation.Resource;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-@Component
-public class MultiTargetNettyClient {
-    // 连接标识属性键
-    private static final AttributeKey<String> CONNECTION_KEY = AttributeKey.valueOf("connectionKey");
-    
-    // 连接管理器
-    private final Map<String, Channel> connectionMap = new ConcurrentHashMap<>();
-    private final Map<Channel, String> reverseConnectionMap = new ConcurrentHashMap<>();
-    private EventLoopGroup workerGroup;
-
-    @Resource
-    private AgentDecoderHandler decoderHandler;
-
-    @Resource
-    private AgentEncoderHandler encoderHandler;
-
-    @Resource
-    private AgentDispatcherHandler dispatcherHandler;
-
-    /**
-     * 动态创建TCP连接
-     * @param connectionKey 连接唯一标识（格式：ip:port 或自定义）
-     * @param host 目标主机
-     * @param port 目标端口
-     * @param timeout 超时时间（秒）
-     * @return 是否创建成功
-     */
-    public boolean createConnection(String connectionKey, String host, int port, int timeout) {
-        if (workerGroup == null) {
-            workerGroup = new NioEventLoopGroup();
-        }
-
-        if (connectionMap.containsKey(connectionKey)) {
-            System.out.println("连接已存在: " + connectionKey);
-            return true;
-        }
-
-        Bootstrap bootstrap = new Bootstrap();
-        bootstrap.group(workerGroup)
-                .channel(NioSocketChannel.class)
-                .option(ChannelOption.TCP_NODELAY, true)
-                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout * 1000)
-                .handler(new ChannelInitializer<SocketChannel>() {
-                    @Override
-                    protected void initChannel(SocketChannel ch) {
-                        ChannelPipeline pipeline = ch.pipeline();
-                        // 添加编解码器
-                        pipeline.addLast(decoderHandler);
-                        pipeline.addLast(encoderHandler);
-                        // 添加心跳机制
-//                        pipeline.addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
-                        // 添加自定义处理器
-                        pipeline.addLast(dispatcherHandler);
-                    }
-                });
-
-        try {
-            CountDownLatch latch = new CountDownLatch(1);
-            final boolean[] success = {false};
-
-            bootstrap.connect(host, port).addListener((ChannelFuture future) -> {
-                if (future.isSuccess()) {
-                    Channel channel = future.channel();
-                    connectionMap.put(connectionKey, channel);
-                    reverseConnectionMap.put(channel, connectionKey);
-                    success[0] = true;
-                    System.out.println("连接建立成功: " + connectionKey + " -> " + host + ":" + port);
-                } else {
-                    System.err.println("连接建立失败: " + connectionKey + " - " + future.cause().getMessage());
-                    success[0] = false;
-                }
-                latch.countDown();
-            });
-
-            // 等待连接结果
-            return latch.await(timeout, TimeUnit.SECONDS) && success[0];
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            return false;
-        }
-    }
-
-    /**
-     * 向指定连接发送消息
-     * @param connectionKey 连接标识
-     * @param message 消息内容
-     * @return 是否发送成功
-     */
-    public boolean sendMessage(String connectionKey, String message) {
-        Channel channel = connectionMap.get(connectionKey);
-        if (channel == null) {
-            System.err.println("连接不存在: " + connectionKey);
-            return false;
-        }
-
-        if (!channel.isActive()) {
-            System.err.println("连接已断开: " + connectionKey);
-            removeConnection(connectionKey);
-            return false;
-        }
-
-        try {
-            ChannelFuture future = channel.writeAndFlush(message).sync();
-            System.out.println("消息发送成功到: " + connectionKey + " - " + message);
-            return future.isSuccess();
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            return false;
-        }
-    }
-
-    public boolean sendMessages(String connectionKey, Message message) {
-        Channel channel = connectionMap.get(connectionKey);
-        if (channel == null) {
-            System.err.println("连接不存在: " + connectionKey);
-            return false;
-        }
-
-        if (!channel.isActive()) {
-            System.err.println("连接已断开: " + connectionKey);
-            removeConnection(connectionKey);
-            return false;
-        }
-
-        try {
-            ChannelFuture future = channel.writeAndFlush(message).sync();
-            System.out.println("消息发送成功到: " + connectionKey + " - " + message);
-            return future.isSuccess();
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            return false;
-        }
-    }
-
-    /**
-     * 批量发送消息到不同连接
-     * @param messages Map<连接标识, 消息内容>
-     */
-    public void sendMessages(Map<String, String> messages) {
-        messages.forEach((connectionKey, message) -> {
-            new Thread(() -> {
-                if (sendMessage(connectionKey, message)) {
-                    System.out.println("批量发送成功: " + connectionKey);
-                } else {
-                    System.err.println("批量发送失败: " + connectionKey);
-                }
-            }).start();
-        });
-    }
-
-    /**
-     * 关闭指定连接
-     * @param connectionKey 连接标识
-     */
-    public void closeConnection(String connectionKey) {
-        Channel channel = connectionMap.get(connectionKey);
-        if (channel != null) {
-            channel.close();
-            removeConnection(connectionKey);
-            System.out.println("连接已关闭: " + connectionKey);
-        }
-    }
-
-    /**
-     * 移除连接记录
-     * @param connectionKey 连接标识
-     */
-    private void removeConnection(String connectionKey) {
-        Channel channel = connectionMap.remove(connectionKey);
-        if (channel != null) {
-            reverseConnectionMap.remove(channel);
-        }
-    }
-
-    /**
-     * 关闭所有连接
-     */
-    public void shutdown() {
-        // 关闭所有连接
-        connectionMap.forEach((key, channel) -> {
-            if (channel.isActive()) {
-                channel.close();
-            }
-        });
-        
-        connectionMap.clear();
-        reverseConnectionMap.clear();
-
-        // 关闭EventLoopGroup
-        if (workerGroup != null) {
-            workerGroup.shutdownGracefully();
-        }
-        System.out.println("所有连接已关闭");
-    }
-
-    /**
-     * 获取连接状态
-     * @return 当前活跃连接数
-     */
-    public int getActiveConnections() {
-        return (int) connectionMap.values().stream()
-                .filter(Channel::isActive)
-                .count();
-    }
-
-//    /**
-//     * 客户端处理器
-//     */
-//    private class ClientHandler extends SimpleChannelInboundHandler<String> {
-//        @Override
-//        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
-//            String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
-//            System.out.println("收到来自 " + connectionKey + " 的消息: " + msg);
-//        }
-//
-//        @Override
-//        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-//            String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
-//            System.err.println("连接 " + connectionKey + " 发生异常: " + cause.getMessage());
-//            ctx.close();
-//        }
-//
-//        @Override
-//        public void channelInactive(ChannelHandlerContext ctx) {
-//            String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
-//            System.out.println("连接断开: " + connectionKey);
-//            removeConnection(connectionKey);
-//        }
-//    }
-
-    /**
-     * 使用示例
-     */
-    public static void main(String[] args) throws InterruptedException {
-        MultiTargetNettyClient client = new MultiTargetNettyClient();
-
-        try {
-            // 动态创建多个不同IP和端口的连接
-            boolean success1 = client.createConnection("server1", "127.0.0.1", 8080, 5);
-            boolean success2 = client.createConnection("server2", "192.168.1.100", 8081, 5);
-            boolean success3 = client.createConnection("server3", "10.0.0.50", 8082, 5);
-
-            // 等待连接建立
-            Thread.sleep(2000);
-
-            if (success1) {
-                client.sendMessage("server1", "Hello Server 1 from 8080");
-            }
-
-            if (success2) {
-                client.sendMessage("server2", "Hello Server 2 from 8081");
-            }
-
-            if (success3) {
-                client.sendMessage("server3", "Hello Server 3 from 8082");
-            }
-
-//            // 批量发送示例
-//            Map<String, String> batchMessages = Map.of(
-//                "server1", "Batch message to server 1",
-//                "server2", "Batch message to server 2",
-//                "server3", "Batch message to server 3"
-//            );
-//
-//            client.sendMessages(batchMessages);
-
-            // 保持运行一段时间
-            Thread.sleep(5000);
-
-            System.out.println("当前活跃连接数: " + client.getActiveConnections());
-
-        } finally {
-            client.shutdown();
-        }
-    }
-
-    public Channel get(String connectionKey){
-        Channel channel = connectionMap.get(connectionKey);
-        return channel;
-    }
-}

src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java

@@ -2,7 +2,6 @@ package com.tongran.agent.server.netty.enpoint;
 
 import com.alibaba.fastjson2.JSONObject;
 import com.tongran.agent.server.core.enums.MsgEnum;
-import com.tongran.agent.server.netty.MultiTargetNettyClient;
 import com.tongran.agent.server.netty.annotation.AgentDispatcher;
 import com.tongran.agent.server.netty.basics.AgentHandler;
 import com.tongran.agent.server.netty.model.UpMsgResponse;
@@ -15,8 +14,6 @@ import javax.annotation.Resource;
 
 @Component
 public class AgentEndpoint {
-    @Resource
-    private MultiTargetNettyClient client;
 
     @Resource
     private AgentMqConfig agentMqConfig;
@@ -24,20 +21,66 @@ public class AgentEndpoint {
     @Resource
     private AgentProducer agentProducer;
 
-    @AgentDispatcher(msgId = MsgEnum.注册应答)
+    @AgentDispatcher(msgId = MsgEnum.建立连接)
-    public class RegisterRspHandler implements AgentHandler {
+    public class ConnectHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject json = new JSONObject();
+            json.put("resCode",1);
+            return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.建立连接应答.getValue())
+                    .content(json.toString()).build();
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.建立连接应答)
+    public class ConnectRspHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            return null;
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.注册)
+    public class RegisterHandler implements AgentHandler {
         @Override
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = new JSONObject();
             jsonObject.put("clientId", clientId);
-            jsonObject.put("dataType", MsgEnum.注册应答.getValue());
+            jsonObject.put("dataType", MsgEnum.注册.getValue());
             jsonObject.put("data", data);
             MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
             agentProducer.asyncSend(mqMsg);
-            JSONObject json = new JSONObject();
+            return null;
-            json.put("resCode",1);
+        }
-            return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.注册应答.getValue())
+    }
-                    .content(json.toString()).build();
+
+    @AgentDispatcher(msgId = MsgEnum.注册应答)
+    public class RegisterRspHandler implements AgentHandler {
+        @Override
+        public String downHandle(String data, String clientId) {
+            return data;
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.获取最新策略)
+    public class GetPolicyHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.获取最新策略.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
+            return null;
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.获取最新策略应答)
+    public class GetPolicyRspHandler implements AgentHandler {
+        @Override
+        public String downHandle(String data, String clientId) {
+            return data;
         }
     }
 
@@ -64,9 +107,9 @@ public class AgentEndpoint {
             if(json.containsKey("resCode")){
                 resCode = json.getInteger("resCode");
             }
-            if(resCode == 1){
+//            if(resCode == 1){
-                client.closeConnection(clientId);
+//                client.closeConnection(clientId);
-            }
+//            }
             return UpMsgResponse.builder().clientId(clientId).build();
         }
     }
@@ -171,7 +214,39 @@ public class AgentEndpoint {
                     .content(json.toString()).build();
         }
     }
+    @AgentDispatcher(msgId = MsgEnum.网络上报重试)
+    public class NetRecoverHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.网络上报重试.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.sendOrderly(mqMsg, clientId);
+            JSONObject json = new JSONObject();
+            json.put("resCode",1);
+            return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.采集上报应答.getValue())
+                    .content(json.toString()).build();
+        }
+    }
 
+    @AgentDispatcher(msgId = MsgEnum.业务网络上报)
+    public class BusinessNetHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.业务网络上报.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
+            JSONObject json = new JSONObject();
+            json.put("resCode",1);
+            return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.采集上报应答.getValue())
+                    .content(json.toString()).build();
+        }
+    }
     @AgentDispatcher(msgId = MsgEnum.挂载上报)
     public class PointHandler implements AgentHandler {
         @Override
@@ -350,14 +425,94 @@ public class AgentEndpoint {
             if(json.containsKey("resCode")){
                 resCode = json.getInteger("resCode");
             }
-            if(resCode == 1){
+//            if(resCode == 1){
-                client.closeConnection(clientId);
+//                client.closeConnection(clientId);
-            }
+//            }
+            return null;
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.多网IP探测上报)
+    public class DetectNetworkHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.多网IP探测上报.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
+            return null;
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.修改frp配置文件应答)
+    public class UpdateFrpHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.修改frp配置文件应答.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
+            return null;
+        }
+    }
+    @AgentDispatcher(msgId = MsgEnum.macvlan状态上报)
+    public class upMacvlanStatusHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.macvlan状态上报.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
+            return null;
+        }
+    }
+    @AgentDispatcher(msgId = MsgEnum.iops结果上报)
+    public class iopsResultHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.iops结果上报.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
+            return null;
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.tcpdump结果上报)
+    public class tcpdumpResultHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.tcpdump结果上报.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
+            return null;
+        }
+    }
+    @AgentDispatcher(msgId = MsgEnum.内存详情上报)
+    public class MemoryDetailsHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("clientId", clientId);
+            jsonObject.put("dataType", MsgEnum.内存详情上报.getValue());
+            jsonObject.put("data", data);
+            MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
+            agentProducer.asyncSend(mqMsg);
             return null;
         }
     }
 
 
 
-
 }

src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java

@@ -21,6 +21,9 @@ import io.netty.util.CharsetUtil;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 
 @Component
@@ -40,121 +43,181 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
 	AgentDispatcherManager agentDispatcherManager;
 
 	// 用来临时保留没有处理过的请求报文
-	private final Cache<String, String> lruCache = CacheUtil.newLRUCache(5000);
+	private final Cache<String, byte[]> byteCache  = CacheUtil.newLRUCache(5000);
 
 	/**
 	 * 消息解码器
 	 */
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) {
-		// 检查是否为 ByteBuf（未解码的原始数据）
 		if (msg instanceof ByteBuf) {
 			ByteBuf byteBuf = (ByteBuf) msg;
-			String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码
+			boolean released = false; // 标记是否已释放
-			AssertLog.info("<<[up1]:[up-content]==>{}", messages);
+			try {
+				// 使用字节数组读取
+				byte[] bytes = new byte[byteBuf.readableBytes()];
+				byteBuf.readBytes(bytes);
 
-			String content = "";
+				String clientKey = sessionManager.client(ctx);
-			StringBuilder sb = new StringBuilder();
-			boolean isClear = false;
-			String tempMsg = lruCache.get(sessionManager.client(ctx));
-			tempMsg = tempMsg == null ? "" : tempMsg;
-			int tmpMsgSize = tempMsg.length();
 
-			boolean startsWith = messages.startsWith("agent-client:");
+				// 1. 获取缓存的字节数组
-			boolean endsWith = messages.endsWith("@tong-ran");
+				byte[] cachedBytes = byteCache.get(clientKey);
-			String ms = messages.replaceAll("agent-client:","");
+
-			String[] arr_msg = ms.split("@tong-ran");
+				// 2. 合并字节数组
-			if(startsWith){
+				byte[] mergedBytes;
-				//判定是否整包
+				if (cachedBytes != null && cachedBytes.length > 0) {
-				if(arr_msg.length == 1){
+					mergedBytes = new byte[cachedBytes.length + bytes.length];
-					//判定是否拆包
+					System.arraycopy(cachedBytes, 0, mergedBytes, 0, cachedBytes.length);
-					if(endsWith){
+					System.arraycopy(bytes, 0, mergedBytes, cachedBytes.length, bytes.length);
-						lruCache.remove(sessionManager.client(ctx));
+					// 合并后清除旧缓存
-						tmpMsgSize = 0;
+					byteCache.remove(clientKey);
-//						sb.append(arr_msg[0]+"@tong-ran");
+				} else {
-						content = arr_msg[0]+"@tong-ran";
+					mergedBytes = bytes;
-					}else{
+				}
-						lruCache.remove(sessionManager.client(ctx));
+
-						tmpMsgSize = 0;
+				// 3. 尝试解码
-//						sb.append(arr_msg[0]);
+				String rawMessage = new String(mergedBytes, StandardCharsets.UTF_8);
-						content = arr_msg[0];
+
-						lruCache.put(sessionManager.client(ctx),  content, DateUnit.SECOND.getMillis() * 5000);
+				// 4. 检查是否有乱码
+				boolean hasGarbledChars = false;
+				for (int i = 0; i < rawMessage.length(); i++) {
+					if (rawMessage.charAt(i) == '\ufffd') {
+						hasGarbledChars = true;
+						break;
 					}
 				}
-				//判定是否粘包
+
-				if(arr_msg.length > 1){
+				if (hasGarbledChars) {
-					if(!endsWith){
+					// 有乱码，缓存字节数组等待后续数据
-						lruCache.remove(sessionManager.client(ctx));
+					byteCache.put(clientKey, mergedBytes, DateUnit.SECOND.getMillis() * 5000);
-						tmpMsgSize = 0;
+
-						for (int i = 0; i < arr_msg.length; i++) {
+					// 打印调试信息
-							if(i == arr_msg.length -1){
+					int garbledIndex = rawMessage.indexOf('\ufffd');
-								lruCache.put(sessionManager.client(ctx), arr_msg[i], DateUnit.SECOND.getMillis() * 5000);
+					AssertLog.info("<<[up1]:[bytes-cached]==>长度: {}, 乱码位置: {}, 预览: {}",
-							}else{
+							mergedBytes.length, garbledIndex, rawMessage);
-								content += arr_msg[i].replaceAll("agent-client:","")+"@tong-ran";
+
-//								sb.append(arr_msg[i].replaceAll("agent-client:","")+"@tong-ran");
+					byteBuf.release();
-							}
+					released = true;
-						}
+					return; // 跳过处理
-					}else{
-						lruCache.remove(sessionManager.client(ctx));
-						tmpMsgSize = 0;
-//						sb.append(String.join("@tong-ran", arr_msg));
-						content = String.join("@tong-ran", arr_msg);
-					}
-				}
-			}
-//			content = sb.toString();
-			if (tmpMsgSize > 0) {
-				content = tempMsg + messages;
-				endsWith = content.endsWith("@tong-ran");
-				if(endsWith){
-					isClear = true;
-				}else{
-					lruCache.remove(sessionManager.client(ctx));
-					tmpMsgSize = 0;
-					lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 5000);
 				}
 
-			}
+				// 5. 没有乱码，处理完整消息
-			AssertLog.info("<<[up2]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content);
+				AssertLog.info("<<[up1]:[raw-content]==>{}", rawMessage);
-			endsWith = content.endsWith("@tong-ran");
+
-			//判断是否是整包
+				// 6. 处理消息分割（原有逻辑）
-			if(endsWith){
+				// 这里不需要从缓存获取，因为已经在字节层面处理过了
-				content = content.replaceAll("agent-client:","");
+				String fullMessage = rawMessage;
-				String[] arr = content.split("@tong-ran");
+
-				try {
+				// 7. 使用更精确的分割逻辑
-					for (String message : arr) {
+				List<String> completeMessages = new ArrayList<>();
-						JSONObject jsonObject = JSONObject.parseObject(message);
+				StringBuilder currentMsg = new StringBuilder();
-						String clientId = jsonObject.getString("clientId");
+
-						String dataType = jsonObject.getString("dataType");
+				// 按消息边界分割
-						String data = jsonObject.getString("data");
+				String[] parts = fullMessage.split("(?=agent-client:)");
-						if(Objects.nonNull(agentDispatcherManager)){
+				for (String part : parts) {
-							AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
+					if (part.isEmpty()) continue;
-							if (ObjectUtil.isNotEmpty(msgHandler)) {
+
-								AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", clientId, dataType, data);
+					currentMsg.append(part);
-								UpMsgResponse response = msgHandler.upHandle(data, clientId );
+					String current = currentMsg.toString();
-								if(Objects.nonNull(response)){
+
-									Message agentMessage = Message.builder().build();
+					// 检查是否包含完整消息边界
-									agentMessage.setClientId(clientId);
+					if (current.contains("@tong-ran")) {
-									agentMessage.setDataType(response.getDataType());
+						String[] messages = current.split("@tong-ran", -1); // 使用limit=-1保留空字符串
-									agentMessage.setData(response.getContent());
+
-									ctx.fireChannelRead(agentMessage);//传递到下一个handler
+						// 前面的都是完整消息
-								}
+						for (int i = 0; i < messages.length - 1; i++) {
+							String completeMsg = messages[i];
+							if (!completeMsg.isEmpty()) {
+								completeMessages.add(completeMsg);
 							}
 						}
 
+						// 最后一部分可能是未完成的消息
+						String lastPart = messages[messages.length - 1];
+						currentMsg = new StringBuilder(lastPart);
 					}
-				}catch (Exception e){
+				}
-					AssertLog.error("=====channelRead:{}=====" + e.getMessage());
+
+				// 8. 处理缓存逻辑
+				if (currentMsg.length() > 0) {
+					// 有未完成的消息，需要缓存
+					// 注意：这里缓存的是字符串，不是字节数组
+					String remainingMsg = currentMsg.toString();
+					if (remainingMsg.length() > 0) {
+						// 将未完成的字符串转为字节数组缓存
+						byte[] remainingBytes = remainingMsg.getBytes(StandardCharsets.UTF_8);
+						byteCache.put(clientKey, remainingBytes, DateUnit.SECOND.getMillis() * 5000);
+					}
+				} else {
+					// 所有消息都处理完成，清除缓存
+					byteCache.remove(clientKey);
+				}
+
+				// 9. 处理完整的消息
+				for (String message : completeMessages) {
+					processCompleteMessage(ctx, message);
+				}
+
+			} catch (Exception e) {
+				AssertLog.error("=====channelRead error:{}=====", e.getMessage(), e);
+			} finally {
+				if(!released && byteBuf.refCnt() > 0){
+					byteBuf.release(); // 确保资源释放
 				}
 			}
-			if (isClear) {
-				lruCache.remove(sessionManager.client(ctx));
-			}
-			byteBuf.release(); // 释放 ByteBuf 资源（重要！）
 		} else {
 			System.out.println("Unexpected message type: " + msg.getClass());
 		}
+	}
 
+	private void processCompleteMessage(ChannelHandlerContext ctx, String message) {
+		try {
+			// 清理消息格式
+			String cleanMessage = message.replaceAll("^agent-client:", "").trim();
+			if (cleanMessage.isEmpty()) return;
+
+			// 截断过长的日志
+			AssertLog.info("<<[up2]:IP:{},[clean-content]==>{}...",
+					sessionManager.client(ctx), cleanMessage);
+
+			// 解析JSON
+			JSONObject jsonObject = JSONObject.parseObject(cleanMessage);
+			String clientId = jsonObject.getString("clientId");
+			String dataType = jsonObject.getString("dataType");
+			String data = jsonObject.getString("data");
+
+			if (Objects.nonNull(agentDispatcherManager)) {
+				AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
+				if (ObjectUtil.isNotEmpty(msgHandler)) {
+					AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}",
+							clientId, dataType, data);
+					UpMsgResponse response = msgHandler.upHandle(data, clientId);
+
+					if (Objects.nonNull(response)) {
+						Message agentMessage = Message.builder()
+								.clientId(clientId)
+								.dataType(response.getDataType())
+								.data(response.getContent())
+								.build();
+						ctx.fireChannelRead(agentMessage); // 传递到下一个handler
+					}
+				}
+			}
+		} catch (Exception e) {
+			// 如果解析失败，可能是消息不完整，尝试缓存字节
+			try {
+				String clientKey = sessionManager.client(ctx);
+				// 将失败的消息转为字节数组缓存
+				byte[] failedBytes = message.getBytes(StandardCharsets.UTF_8);
+				byteCache.put(clientKey, failedBytes, DateUnit.SECOND.getMillis() * 5000);
+				AssertLog.warn("<<[parse-failed-cached]: 消息解析失败，已缓存字节，长度: {}", failedBytes.length);
+			} catch (Exception cacheEx) {
+				AssertLog.error("=====缓存失败消息时出错:{}=====", cacheEx.getMessage(), cacheEx);
+			}
+
+			AssertLog.error("=====processCompleteMessage error, message: {}, error: {}=====",
+					message, e.getMessage(), e);
+		}
 	}
 
 	@Override

src/main/java/com/tongran/agent/server/netty/handler/TCPListenHandler.java

@@ -0,0 +1,66 @@
+package com.tongran.agent.server.netty.handler;
+
+
+import com.tongran.agent.server.core.session.SessionManager;
+import com.tongran.agent.server.service.AgentService;
+import com.tongran.agent.server.utils.AssertLog;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.io.IOException;
+
+/**
+ * @Description: TCP消息适配器
+ */
+@Component
+@ChannelHandler.Sharable
+public class TCPListenHandler extends ChannelInboundHandlerAdapter {
+
+	private final SessionManager sessionManager;
+
+	@Resource
+	private AgentService agentService;
+
+	public TCPListenHandler() {
+		this.sessionManager = SessionManager.getInstance();
+	}
+
+	@Override
+	public void channelActive(ChannelHandlerContext ctx) {
+		AssertLog.info("<<<<<<[终端连接]{}", ctx.channel().remoteAddress());
+	}
+
+	@Override
+	public void channelInactive(ChannelHandlerContext ctx) {
+		AssertLog.info(">>>>>>[断开连接]{}", sessionManager.client(ctx));
+		sessionManager.remove(ctx.channel());
+	}
+
+	@Override
+	public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
+		if (e instanceof IOException) {
+			AssertLog.info("<<<<<<[终端断开连接]{} {}", sessionManager.client(ctx), e.getMessage());
+		} else {
+			AssertLog.info(">>>>>>[消息处理异常]" + sessionManager.client(ctx), e);
+		}
+	}
+
+	@Override
+	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+		if (evt instanceof IdleStateEvent) {
+			IdleStateEvent event = (IdleStateEvent) evt;
+			IdleState state = event.state();
+			if (state == IdleState.READER_IDLE || state == IdleState.WRITER_IDLE || state == IdleState.ALL_IDLE) {
+				AssertLog.warn(">>>>>>[终端心跳超时]{} {}", state, sessionManager.client(ctx));
+				sessionManager.remove(ctx.channel());
+				ctx.close();
+			}
+		}
+	}
+
+}

src/main/java/com/tongran/agent/server/netty/handler/UDPListenHandler.java

@@ -0,0 +1,24 @@
+package com.tongran.agent.server.netty.handler;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.socket.DatagramPacket;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Description: UDP消息适配器
+ */
+@Component
+@ChannelHandler.Sharable
+public class UDPListenHandler extends ChannelInboundHandlerAdapter {
+
+	@Override
+	public void channelRead(ChannelHandlerContext ctx, Object msg) {
+		DatagramPacket packet = (DatagramPacket) msg;
+		ByteBuf buf = packet.content();
+		buf.clear();// 暂未实现
+	}
+
+}

src/main/java/com/tongran/agent/server/rockermq/AgentProducer.java

@@ -51,4 +51,19 @@ public class AgentProducer implements RocketMqService {
 		rocketMQTemplate.syncSend(msg.getTopic(), MessageBuilder.withPayload(msg.getContent()).build(), 30000, msg.getLevel());
 	}
 
+	// 新增顺序发送
+	@Override
+	public void sendOrderly(MqMsg mqMsg, String shardingKey) {
+		try {
+			AssertLog.info("sendOrderly发送消息:==>{}", mqMsg);
+			rocketMQTemplate.syncSendOrderly(
+					mqMsg.getTopic(),
+					MessageBuilder.withPayload(mqMsg.getContent()).build(),
+					shardingKey
+			);
+		} catch (Exception e) {
+			AssertLog.error("顺序发送消息失败", e);
+		}
+	}
+
 }

src/main/java/com/tongran/agent/server/rockermq/RocketMqService.java

@@ -42,4 +42,11 @@ public interface RocketMqService {
      */
     default void delayedSendOrderly(MqMsg msg){};
 
+    /**
+     * 发送顺序消息
+     * @param mqMsg
+     * @param shardingKey
+     */
+    public void sendOrderly(MqMsg mqMsg, String shardingKey);
+	
 }

src/main/java/com/tongran/agent/server/rockermq/listener/AgentConsumerDownListener.java

@@ -2,19 +2,13 @@ package com.tongran.agent.server.rockermq.listener;
 
 import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONObject;
 import com.tongran.agent.server.core.enums.MsgEnum;
 import com.tongran.agent.server.core.session.SessionManager;
-import com.tongran.agent.server.netty.MultiTargetNettyClient;
 import com.tongran.agent.server.netty.annotation.AgentDispatcher;
 import com.tongran.agent.server.netty.basics.AgentDispatcherManager;
 import com.tongran.agent.server.netty.basics.AgentHandler;
 import com.tongran.agent.server.netty.model.Message;
-import com.tongran.agent.server.rockermq.AgentProducer;
-import com.tongran.agent.server.rockermq.config.AgentMqConfig;
-import com.tongran.agent.server.rockermq.model.MqMsg;
 import com.tongran.agent.server.utils.AssertLog;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -25,7 +19,7 @@ import java.util.Objects;
 
 @Component
 @ConditionalOnProperty(name = "rocketmq.enabled", havingValue = "true")
-@RocketMQMessageListener(topic = "tongran_agent_down", consumerGroup = "tongran_agent_down_group")
+@RocketMQMessageListener(topic = "tr_agent_down", consumerGroup = "tr_agent_down_group")
 public class AgentConsumerDownListener implements RocketMQListener<String> {
 
 	protected final SessionManager sessionManager;
@@ -33,15 +27,6 @@ public class AgentConsumerDownListener implements RocketMQListener<String> {
     @Resource
     private AgentDispatcherManager agentDispatcherManager;
 
-    @Resource
-    private MultiTargetNettyClient client;
-
-    @Resource
-    private AgentMqConfig agentMqConfig;
-
-    @Resource
-    private AgentProducer agentProducer;
-
     public AgentConsumerDownListener() {
         this.sessionManager = SessionManager.getInstance();
     }
@@ -51,112 +36,14 @@ public class AgentConsumerDownListener implements RocketMQListener<String> {
         AssertLog.info("consumer==> received down message: {}", message);
         try {
             Message dto = JSON.parseObject(message, Message.class);
-            if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
+            AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
-                JSONObject jsonObject = JSONObject.parseObject(dto.getData());
+            if (ObjectUtil.isNotEmpty(msgHandler)) {
-                String clientIp = "";
+                String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
-                int clientPort = 0;
+                Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
-                String switchBoard = "";
+                AssertLog.info("连接是否存在：{},clientId:{},session:{}",
-                long timestamp = 0;
+                        Objects.nonNull(sessionManager.getSessionById(dto.getClientId())), dto.getClientId(), sessionManager.getSessionById(dto.getClientId()));
-                if(jsonObject.containsKey("clientIp")){
+                if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
-                    clientIp = jsonObject.getString("clientIp");
+                    sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
-                }
-                if(jsonObject.containsKey("clientPort")){
-                    clientPort = jsonObject.getInteger("clientPort");
-                }
-                if(jsonObject.containsKey("switchBoard")){
-                    switchBoard = jsonObject.getString("switchBoard");
-                }
-                if(jsonObject.containsKey("timestamp")){
-                    timestamp = jsonObject.getLongValue("timestamp");
-                }
-                if(StringUtils.isBlank(clientIp) || clientPort == 0){
-                    //返回提醒ip 端口 空
-                    JSONObject json = new JSONObject();
-                    json.put("clientId",dto.getClientId());
-                    json.put("dataType",MsgEnum.注册应答.getValue());
-                    JSONObject dataJson = new JSONObject();
-                    dataJson.put("resCode",0);
-                    dataJson.put("resMag","客户端IP、端口不能为空");
-                    dataJson.put("timestamp", System.currentTimeMillis());
-                    json.put("data",dataJson.toString());
-                    MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(json.toString()).build();
-                    agentProducer.asyncSend(mqMsg);
-                    return;
-                }
-                boolean success = client.createConnection(dto.getClientId(), clientIp, clientPort, 5);
-                if(success){
-                    //连接成功
-                    // 连接建立后可以发送登录认证消息
-                    JSONObject object = new JSONObject();
-                    object.put("timestamp", timestamp);
-                    if(StringUtils.isNotBlank(switchBoard)){
-                        String community = "";
-                        String switchIp = "";
-                        int switchPort = 0;
-                        String netOID = "";
-                        String moduleOID = "";
-                        String mpuOID = "";
-                        String pwrOID = "";
-                        String fanOID = "";
-                        String otherOID = "";
-                        String filters = "";
-                        JSONObject json = JSONObject.parseObject(switchBoard);
-                        if(json.containsKey("community")){
-                            community = json.getString("community");
-                        }
-                        if(json.containsKey("ip")){
-                            switchIp = json.getString("ip");
-                        }
-                        if(json.containsKey("port")){
-                            switchPort = json.getInteger("port");
-                        }
-                        if(json.containsKey("netOID")){
-                            netOID = json.getString("netOID");
-                        }
-                        if(json.containsKey("moduleOID")){
-                            moduleOID = json.getString("moduleOID");
-                        }
-                        if(json.containsKey("mpuOID")){
-                            mpuOID = json.getString("mpuOID");
-                        }
-                        if(json.containsKey("pwrOID")){
-                            pwrOID = json.getString("pwrOID");
-                        }
-                        if(json.containsKey("fanOID")){
-                            fanOID = json.getString("fanOID");
-                        }
-                        if(json.containsKey("otherOID")){
-                            otherOID = json.getString("otherOID");
-                        }
-                        if(json.containsKey("filters")){
-                            filters = json.getString("filters");
-                        }
-                        JSONObject switchBoardJson = new JSONObject();
-                        switchBoardJson.put("community",community);
-                        switchBoardJson.put("ip",switchIp);
-                        switchBoardJson.put("port",switchPort);
-                        switchBoardJson.put("netOID",netOID);
-                        switchBoardJson.put("moduleOID",moduleOID);
-                        switchBoardJson.put("mpuOID",mpuOID);
-                        switchBoardJson.put("pwrOID",pwrOID);
-                        switchBoardJson.put("fanOID",fanOID);
-                        switchBoardJson.put("otherOID",otherOID);
-                        switchBoardJson.put("filters",filters);
-                        object.put("switchBoard",switchBoardJson.toString());
-                        AssertLog.info("交换机配置信息={}",switchBoardJson.toString());
-                    }
-                    Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
-                    // 将对象转为 JSON 字符串
-                    client.sendMessages(dto.getClientId(), msg);
-                }
-            } else {
-                AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
-                if (ObjectUtil.isNotEmpty(msgHandler)) {
-                    String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
-                    Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
-                    if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
-                        sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
-                    }
                 }
             }
         } catch (Exception e) {

src/main/java/com/tongran/agent/server/service/impl/AgentServiceImpl.java

@@ -3,37 +3,35 @@ package com.tongran.agent.server.service.impl;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
 import com.tongran.agent.server.core.vo.RegisterVO;
-import com.tongran.agent.server.netty.MultiTargetNettyClient;
 import com.tongran.agent.server.netty.model.Message;
 import com.tongran.agent.server.service.AgentService;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
-
 @Service
 public class AgentServiceImpl implements AgentService {
 
-    @Resource
+//    @Resource
-    private MultiTargetNettyClient client;
+//    private MultiTargetNettyClient client;
 
 
     @Override
     public boolean createConn(String clientId, String host, int port, int timeout) {
         // 动态创建不同IP和端口的连接
-        boolean success = client.createConnection(clientId, host, port, timeout);
+//        boolean success = client.createConnection(clientId, host, port, timeout);
-        if(success){
+//        if(success){
-            // 连接建立后可以发送建立消息
+//            // 连接建立后可以发送建立消息
-            long timestamp = System.currentTimeMillis();
+//            long timestamp = System.currentTimeMillis();
-            JSONObject object = new JSONObject();
+//            JSONObject object = new JSONObject();
-            object.put("clientId",clientId);
+//            object.put("clientId",clientId);
-            object.put("timestamp",timestamp);
+//            object.put("timestamp",timestamp);
-            Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
+//            Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
-            // 将对象转为 JSON 字符串
+//            // 将对象转为 JSON 字符串
-            String json = JSON.toJSONString(message);
+//            String json = JSON.toJSONString(message);
-            System.out.println("连接建立成功，发送建立消息="+json);
+//            System.out.println("连接建立成功，发送建立消息="+json);
-            client.sendMessages(clientId, message);
+//            client.sendMessages(clientId, message);
-        }
+//        }
-        return success;
+//        return success;
+        return false;
     }
 
     @Override
@@ -52,6 +50,6 @@ public class AgentServiceImpl implements AgentService {
         // 将对象转为 JSON 字符串
         String json = JSON.toJSONString(message);
         System.out.println("发送注册消息="+json);
-        client.sendMessages(registerVO.getClientId(), message);
+//        client.sendMessages(registerVO.getClientId(), message);
     }
 }

src/main/java/com/tongran/agent/server/utils/ClientExample.java

@@ -40,7 +40,7 @@ public class ClientExample {
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
         } finally {
-            manager.shutdown();
+//            manager.shutdown();
         }
     }
 }

src/main/java/com/tongran/agent/server/utils/EnhancedConnectionManager.java

@@ -1,6 +1,5 @@
 package com.tongran.agent.server.utils;
 
-import com.tongran.agent.server.netty.MultiTargetNettyClient;
 import com.tongran.agent.server.netty.config.ConnectionConfig;
 
 import java.util.HashMap;
@@ -12,11 +11,11 @@ import java.util.concurrent.ConcurrentHashMap;
  * 增强版连接管理器
  */
 public class EnhancedConnectionManager {
-    private final MultiTargetNettyClient client;
+//    private final MultiTargetNettyClient client;
     private final Map<String, ConnectionConfig> connectionConfigs;
 
     public EnhancedConnectionManager() {
-        this.client = new MultiTargetNettyClient();
+//        this.client = new MultiTargetNettyClient();
         this.connectionConfigs = new ConcurrentHashMap<>();
     }
 
@@ -26,18 +25,18 @@ public class EnhancedConnectionManager {
     public Map<String, Boolean> createConnections(List<ConnectionConfig> configs) {
         Map<String, Boolean> results = new HashMap<>();
         
-        configs.forEach(config -> {
+//        configs.forEach(config -> {
-            boolean success = client.createConnection(
+//            boolean success = client.createConnection(
-                config.getConnectionKey(),
+//                config.getConnectionKey(),
-                config.getHost(),
+//                config.getHost(),
-                config.getPort(),
+//                config.getPort(),
-                config.getTimeout()
+//                config.getTimeout()
-            );
+//            );
-            if (success) {
+//            if (success) {
-                connectionConfigs.put(config.getConnectionKey(), config);
+//                connectionConfigs.put(config.getConnectionKey(), config);
-            }
+//            }
-            results.put(config.getConnectionKey(), success);
+//            results.put(config.getConnectionKey(), success);
-        });
+//        });
         
         return results;
     }
@@ -47,21 +46,21 @@ public class EnhancedConnectionManager {
      */
     public void routeMessageByType(String messageType, String message) {
         // 这里可以根据业务逻辑决定发送到哪个连接
-        switch (messageType) {
+//        switch (messageType) {
-            case "TYPE_A":
+//            case "TYPE_A":
-                client.sendMessage("server1", "[TYPE_A] " + message);
+//                client.sendMessage("server1", "[TYPE_A] " + message);
-                break;
+//                break;
-            case "TYPE_B":
+//            case "TYPE_B":
-                client.sendMessage("server2", "[TYPE_B] " + message);
+//                client.sendMessage("server2", "[TYPE_B] " + message);
-                break;
+//                break;
-            case "TYPE_C":
+//            case "TYPE_C":
-                client.sendMessage("server3", "[TYPE_C] " + message);
+//                client.sendMessage("server3", "[TYPE_C] " + message);
-                break;
+//                break;
-            default:
+//            default:
-                // 广播到所有连接
+//                // 广播到所有连接
-                connectionConfigs.keySet().forEach(key -> 
+//                connectionConfigs.keySet().forEach(key ->
-                    client.sendMessage(key, "[BROADCAST] " + message));
+//                    client.sendMessage(key, "[BROADCAST] " + message));
-        }
+//        }
     }
 
     /**
@@ -69,14 +68,14 @@ public class EnhancedConnectionManager {
      */
     public Map<String, Boolean> getConnectionStatus() {
         Map<String, Boolean> status = new HashMap<>();
-        connectionConfigs.forEach((key, config) -> {
+//        connectionConfigs.forEach((key, config) -> {
-            // 这里可以添加更详细的状态检查
+//            // 这里可以添加更详细的状态检查
-            status.put(key, client.sendMessage(key, "PING"));
+//            status.put(key, client.sendMessage(key, "PING"));
-        });
+//        });
         return status;
     }
 
-    public void shutdown() {
+//    public void shutdown() {
-        client.shutdown();
+//        client.shutdown();
-    }
+//    }
 }

src/main/resources/application-dev.yml

@@ -1,5 +1,5 @@
 server:
-  port: 7012
+  port: -1
   servlet:
     context-path: /tr-agent-server
   
@@ -12,21 +12,21 @@ knife4j:
 logging:
   file:
 #    path: /usr/local/tongran/logs
-    path: /data/tr-agent-server/logs
+    path: /usr/local/tongran_server/logs
 #    path: D:/job/agent-logs/logs
 
 rocketmq:
+  #测试环境
+  #name-server: 172.16.15.52:9876
+  #生产环境
   name-server: 172.16.15.103:9876
   enabled: true
 
-netty:
+tcp:
-  server:
+  netty:
-#    host: 172.16.15.103
+    charge:
-    host: 127.0.0.1
+      enable: true
-    port: 6610
+      name: AGENT-SERVER-服务
-  client:
+      port: 6620
-    client-id: client-001
+      readerIdleTime: 300
-    reconnect-interval: 5
+
-    maxReconnectAttempts: 10    # 最大重连次数
-    initialReconnectDelay: 1000 # 初始重连延迟(毫秒)
-    maxReconnectDelay: 30000    # 最大重连延迟(毫秒)

src/main/resources/application.yml

@@ -34,10 +34,10 @@ rocketmq:
   producer:
     group: tongran-group
     topic: default_topic
-    agent-group: tongran_agent_up_group
+    agent-group: tr_agent_up_group
-    agent-topic: tongran_agent_up
+    agent-topic: tr_agent_up
   consumer:
     group: tongran-group
     topic: default_topic
-    agent-group: tongran_agent_down_group
+    agent-group: tr_agent_down_group
-    agent-topic: tongran_agent_down
+    agent-topic: tr_agent_down