采集网口信息增加 协商速度、工作模式

增加客户端处理服务端下发-文件下载
增加客户端处理服务端下发-执行命令

src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java

@@ -42,7 +42,15 @@ public enum MsgEnum {
 
     更新系统采集间隔("TIME_SYSTEM"),
 
-    更新系统采集间隔应答("TIME_SYSTEM_RSP");
+    更新系统采集间隔应答("TIME_SYSTEM_RSP"),
+
+    文件下载("DOWN_FILE"),
+
+    文件下载应答("DOWN_FILE_RSP"),
+
+    执行命令("EXECUTE_COMMAND"),
+
+    执行命令应答("EXECUTE_COMMAND_RSP");
 
     private String value;
 

src/main/java/com/tongran/agentserver/server/collect/net/impl/NetServiceImpl.java

@@ -9,9 +9,9 @@ import oshi.hardware.HardwareAbstractionLayer;
 import oshi.hardware.NetworkIF;
 import oshi.util.FormatUtil;
 
-import java.util.ArrayList;
+import java.io.BufferedReader;
-import java.util.List;
+import java.io.InputStreamReader;
-import java.util.Objects;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
 @Service
@@ -42,6 +42,11 @@ public class NetServiceImpl implements NetService {
                 netVO.setStatus(net.isConnectorPresent() ? "已连接" : "未连接");//运行状态
                 netVO.setType(EscapeUtil.getInterfaceType(net));//接口类型
                 netVO.setIpV4(String.join(", ", net.getIPv4addr()));//IPv4
+                Map<String, String> map = getNetworkMode(net.getName());
+                if (map != null && !map.isEmpty()) {
+                    netVO.setSpeed(map.get("speed"));
+                    netVO.setDuplex(map.get("duplex"));
+                }
                 temp.add(netVO);
             }
             // 2. 实时带宽监控（需要两次采样）
@@ -159,4 +164,36 @@ public class NetServiceImpl implements NetService {
         return bytes * 8.0 / 1_000_000; // bytes to megabits
     }
 
+    public static Map<String, String> getNetworkMode(String interfaceName) {
+        Map<String, String> result = new HashMap<>();
+        ProcessBuilder pb = new ProcessBuilder("ethtool", interfaceName);
+        pb.redirectErrorStream(true);
+
+        try {
+            Process process = pb.start();
+            BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+            String line;
+
+            while ((line = reader.readLine()) != null) {
+                if (line.contains("Speed:")) {
+                    result.put("speed", line.split(":")[1].trim());
+                } else if (line.contains("Duplex:")) {
+                    result.put("duplex", line.split(":")[1].trim());
+                } else if (line.contains("Auto-negotiation:")) {
+                    result.put("auto-negotiation", line.split(":")[1].trim());
+                }
+            }
+
+            int exitCode = process.waitFor();
+            if (exitCode != 0) {
+                result.put("error", "ethtool command failed with exit code " + exitCode);
+            }
+
+        } catch (Exception e) {
+            result.put("error", e.getMessage());
+        }
+
+        return result;
+    }
+
 }

src/main/java/com/tongran/agentserver/server/collect/vo/NetVO.java

@@ -43,8 +43,11 @@ public class NetVO implements Serializable {
     @Schema(description = "接收流量")
     private long inSpeed;
 
-//    @Schema(description = "接收速度")
+    @Schema(description = "协商速度")
-//    private String speed;
+    private String speed;
+
+    @Schema(description = "工作模式")
+    private String duplex;
 
     @Schema(description = "时间戳")
     private long timestamp;

src/main/java/com/tongran/agentserver/server/file/AdvancedAsyncDownloader.java

@@ -0,0 +1,121 @@
+package com.tongran.agentserver.server.file;
+
+import java.io.BufferedInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Consumer;
+
+public class AdvancedAsyncDownloader {
+    private static final ExecutorService downloadExecutor = 
+        Executors.newFixedThreadPool(5);
+    
+    /**
+     * 带进度回调的异步下载
+     */
+    public static CompletableFuture<String> downloadWithProgress(
+            String fileUrl, 
+            String saveDir, 
+            String fileName,
+            Consumer<Double> progressCallback) {
+        
+        return CompletableFuture.supplyAsync(() -> {
+            try {
+                URL url = new URL(fileUrl);
+                HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+                
+                // 获取文件大小
+                long fileSize = connection.getContentLengthLong();
+                
+                // 创建保存目录
+                java.nio.file.Path directory = java.nio.file.Paths.get(saveDir);
+                if (!java.nio.file.Files.exists(directory)) {
+                    java.nio.file.Files.createDirectories(directory);
+                }
+                
+                String filePath = directory.resolve(fileName != null ? fileName : 
+                    extractFileNameFromUrl(fileUrl)).toString();
+                
+                try (BufferedInputStream in = new BufferedInputStream(connection.getInputStream());
+                     FileOutputStream out = new FileOutputStream(filePath)) {
+                    
+                    byte[] buffer = new byte[8192];
+                    int bytesRead;
+                    long totalRead = 0;
+                    
+                    while ((bytesRead = in.read(buffer)) != -1) {
+                        out.write(buffer, 0, bytesRead);
+                        totalRead += bytesRead;
+                        
+                        // 回调进度
+                        if (progressCallback != null && fileSize > 0) {
+                            double progress = (double) totalRead / fileSize * 100;
+                            progressCallback.accept(progress);
+                        }
+                    }
+                }
+                
+                return filePath;
+            } catch (Exception e) {
+                throw new RuntimeException("下载失败", e);
+            }
+        }, downloadExecutor);
+    }
+    
+    private static String extractFileNameFromUrl(String fileUrl) {
+        // 实现文件名提取逻辑
+        return fileUrl.substring(fileUrl.lastIndexOf('/') + 1);
+    }
+
+    /**
+     * 检查目录是否存在，不存在则创建（不创建父目录）
+     * @param directoryPath 目录路径
+     * @return 创建成功返回true，否则返回false
+     */
+    public static boolean createSingleDirectoryIfNotExists(String directoryPath) {
+        try {
+            Path path = Paths.get(directoryPath);
+
+            if (!Files.exists(path)) {
+                // 只创建单级目录（父目录必须存在）
+                Files.createDirectory(path);
+                System.out.println("目录创建成功: " + directoryPath);
+                return true;
+            } else {
+                System.out.println("目录已存在: " + directoryPath);
+                return true;
+            }
+        } catch (IOException e) {
+            System.err.println("创建目录失败: " + directoryPath);
+            System.err.println("错误信息: " + e.getMessage());
+            return false;
+        }
+    }
+
+    /**
+     * 使用示例
+     */
+    public static void main(String[] args) {
+        String fileUrl = "https://www.tzdsp.net/ksc-andromedae";
+        String saveDir = "D:/down";
+        
+        downloadWithProgress(fileUrl, saveDir, null, progress -> {
+            System.out.printf("下载进度: %.1f%%\n", progress);
+        }).thenAccept(filePath -> {
+            System.out.println("下载完成: " + filePath);
+        }).exceptionally(ex -> {
+            System.err.println("下载错误: " + ex.getMessage());
+            return null;
+        });
+        
+        // 主线程继续执行
+        System.out.println("异步下载已启动...");
+    }
+}

src/main/java/com/tongran/agentserver/server/file/AsyncCommandExecutor.java

@@ -0,0 +1,270 @@
+package com.tongran.agentserver.server.file;
+
+import java.io.*;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.attribute.PosixFilePermission;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.*;
+import java.util.function.Consumer;
+
+public class AsyncCommandExecutor {
+    
+    // 使用线程池管理异步任务
+    private static final ExecutorService executorService = Executors.newCachedThreadPool();
+    private static final ScheduledExecutorService timeoutExecutor = Executors.newScheduledThreadPool(5);
+    
+    /**
+     * 异步为文件添加可执行权限
+     */
+    public static CompletableFuture<Boolean> makeFileExecutableAsync(String filePath) {
+        return CompletableFuture.supplyAsync(() -> {
+            try {
+                Path path = Paths.get(filePath);
+                
+                if (!Files.exists(path)) {
+                    throw new RuntimeException("文件不存在: " + filePath);
+                }
+                
+                Set<PosixFilePermission> permissions = Files.getPosixFilePermissions(path);
+                permissions.add(PosixFilePermission.OWNER_EXECUTE);
+                permissions.add(PosixFilePermission.GROUP_EXECUTE);
+                permissions.add(PosixFilePermission.OTHERS_EXECUTE);
+                
+                Files.setPosixFilePermissions(path, permissions);
+                return true;
+                
+            } catch (Exception e) {
+                throw new RuntimeException("设置执行权限失败: " + e.getMessage(), e);
+            }
+        }, executorService);
+    }
+    
+    /**
+     * 异步执行命令（基础版本）
+     */
+    public static CompletableFuture<CommandResult> executeCommandAsync(
+            List<String> command,
+            long timeout,
+            TimeUnit timeUnit) {
+        
+        return CompletableFuture.supplyAsync(() -> {
+            Process process = null;
+            try {
+                ProcessBuilder processBuilder = new ProcessBuilder(command);
+                processBuilder.redirectErrorStream(true);
+//                processBuilder.directory(new File("/data/agent-server")); // 设置工作目录
+                // 不合并错误流，分别读取
+                process = processBuilder.start();
+                
+                // 异步读取输出
+                Process finalProcess = process;
+                Future<String> outputFuture = executorService.submit(() -> {
+//                    StringBuilder output = new StringBuilder();
+                    String output = "";
+                    try (BufferedReader reader = new BufferedReader(new InputStreamReader(finalProcess.getInputStream()))) {
+                        String line;
+                        while ((line = reader.readLine()) != null) {
+//                            output.append(line).append("\n");
+                            output = line;
+                        }
+                    }
+                    return output;
+                });
+                
+                // 设置超时
+                if (timeout > 0) {
+                    boolean finished = process.waitFor(timeout, timeUnit);
+                    if (!finished) {
+                        process.destroy();
+                        throw new TimeoutException("命令执行超时");
+                    }
+                } else {
+                    process.waitFor();
+                }
+                
+                String output = outputFuture.get(1, TimeUnit.SECONDS);
+                int exitCode = process.exitValue();
+                return new CommandResult(exitCode, output, "");
+            } catch (Exception e) {
+                return new CommandResult(-1, "", "执行失败: " + e.getMessage());
+            } finally {
+                if (process != null) {
+                    process.destroy();
+                }
+            }
+        }, executorService);
+    }
+    
+    /**
+     * 带实时输出的异步命令执行
+     */
+    public static CompletableFuture<CommandResult> executeCommandWithRealtimeOutput(
+            String[] command,
+            long timeout,
+            TimeUnit timeUnit,
+            Consumer<String> outputConsumer,
+            Consumer<String> errorConsumer) {
+        
+        return CompletableFuture.supplyAsync(() -> {
+            Process process = null;
+            try {
+                ProcessBuilder processBuilder = new ProcessBuilder(command);
+                process = processBuilder.start();
+                
+                // 启动输出读取线程
+                CompletableFuture<Void> outputFuture = readStreamAsync(
+                    process.getInputStream(), outputConsumer, "OUTPUT");
+                
+                CompletableFuture<Void> errorFuture = readStreamAsync(
+                    process.getErrorStream(), errorConsumer, "ERROR");
+                
+                // 设置超时监控
+                ScheduledFuture<?> timeoutFuture = null;
+                if (timeout > 0) {
+                    Process finalProcess = process;
+                    timeoutFuture = timeoutExecutor.schedule(() -> {
+                        if (finalProcess.isAlive()) {
+                            finalProcess.destroy();
+                        }
+                    }, timeout, timeUnit);
+                }
+                
+                // 等待进程完成
+                int exitCode = process.waitFor();
+                
+                // 取消超时任务
+                if (timeoutFuture != null) {
+                    timeoutFuture.cancel(false);
+                }
+                
+                // 等待输出读取完成
+                CompletableFuture.allOf(outputFuture, errorFuture).get(2, TimeUnit.SECONDS);
+                
+                return new CommandResult(exitCode, "", "");
+                
+            } catch (Exception e) {
+                return new CommandResult(-1, "", "执行异常: " + e.getMessage());
+            } finally {
+                if (process != null) {
+                    process.destroy();
+                }
+            }
+        }, executorService);
+    }
+    
+    /**
+     * 异步读取流数据
+     */
+    private static CompletableFuture<Void> readStreamAsync(
+            InputStream inputStream, 
+            Consumer<String> lineConsumer,
+            String streamType) {
+        
+        return CompletableFuture.runAsync(() -> {
+            try (BufferedReader reader = new BufferedReader(
+                 new InputStreamReader(inputStream))) {
+                
+                String line;
+                while ((line = reader.readLine()) != null) {
+                    if (lineConsumer != null) {
+                        lineConsumer.accept("[" + streamType + "] " + line);
+                    }
+                }
+            } catch (IOException e) {
+                if (lineConsumer != null) {
+                    lineConsumer.accept("[" + streamType + "-ERROR] " + e.getMessage());
+                }
+            }
+        }, executorService);
+    }
+    
+    /**
+     * 异步执行脚本文件
+     */
+    public static CompletableFuture<CommandResult> executeScriptAsync(
+            String scriptPath,
+            String[] args,
+            long timeout,
+            TimeUnit timeUnit,
+            Consumer<String> realtimeOutputConsumer) {
+        
+        return makeFileExecutableAsync(scriptPath)
+            .thenCompose(permissionSuccess -> {
+                if (!permissionSuccess) {
+                    return CompletableFuture.completedFuture(
+                        new CommandResult(-1, "", "设置执行权限失败"));
+                }
+                
+                String[] command = new String[args.length + 1];
+                command[0] = scriptPath;
+                System.arraycopy(args, 0, command, 1, args.length);
+                
+                return executeCommandWithRealtimeOutput(
+                    command, timeout, timeUnit, 
+                    realtimeOutputConsumer, 
+                    error -> System.err.println(error));
+            });
+    }
+    
+    /**
+     * 批量异步执行命令
+     */
+//    public static List<CompletableFuture<CommandResult>> executeCommandsAsync(
+//            List<String[]> commands,
+//            long timeout,
+//            TimeUnit timeUnit) {
+//
+//        return commands.stream()
+//            .map(command -> executeCommandAsync(command, timeout, timeUnit))
+//            .collect(java.util.stream.Collectors.toList());
+//    }
+    
+    /**
+     * 关闭执行器
+     */
+    public static void shutdown() {
+        executorService.shutdown();
+        timeoutExecutor.shutdown();
+        try {
+            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+                executorService.shutdownNow();
+            }
+            if (!timeoutExecutor.awaitTermination(2, TimeUnit.SECONDS)) {
+                timeoutExecutor.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            executorService.shutdownNow();
+            timeoutExecutor.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+    }
+    
+    /**
+     * 执行结果封装类
+     */
+    public static class CommandResult {
+        private final int exitCode;
+        private final String output;
+        private final String error;
+        
+        public CommandResult(int exitCode, String output, String error) {
+            this.exitCode = exitCode;
+            this.output = output;
+            this.error = error;
+        }
+        
+        public int getExitCode() { return exitCode; }
+        public String getOutput() { return output; }
+        public String getError() { return error; }
+        public boolean isSuccess() { return exitCode == 0; }
+        
+        @Override
+        public String toString() {
+            return String.format("Exit Code: %d\nOutput: %s\nError: %s", 
+                exitCode, output, error);
+        }
+    }
+}

src/main/java/com/tongran/agentserver/server/file/LinuxCommandExecutor.java

@@ -0,0 +1,175 @@
+package com.tongran.agentserver.server.file;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.attribute.PosixFilePermission;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+public class LinuxCommandExecutor {
+    
+    /**
+     * 为文件添加可执行权限
+     * @param filePath 文件路径
+     * @return 成功返回true，失败返回false
+     */
+    public static boolean makeFileExecutable(String filePath) {
+        try {
+            Path path = Paths.get(filePath);
+            
+            // 检查文件是否存在
+            if (!Files.exists(path)) {
+                System.err.println("文件不存在: " + filePath);
+                return false;
+            }
+            
+            // 获取当前权限
+            Set<PosixFilePermission> permissions = Files.getPosixFilePermissions(path);
+            
+            // 添加执行权限
+            permissions.add(PosixFilePermission.OWNER_EXECUTE);
+            permissions.add(PosixFilePermission.GROUP_EXECUTE);
+            permissions.add(PosixFilePermission.OTHERS_EXECUTE);
+            
+            // 设置新权限
+            Files.setPosixFilePermissions(path, permissions);
+            
+            System.out.println("文件已赋予可执行权限: " + filePath);
+            return true;
+            
+        } catch (IOException e) {
+            System.err.println("设置执行权限失败: " + e.getMessage());
+            return false;
+        } catch (UnsupportedOperationException e) {
+            System.err.println("当前系统不支持POSIX权限: " + e.getMessage());
+            return false;
+        }
+    }
+    
+    /**
+     * 使用chmod命令赋予可执行权限（备用方法）
+     */
+    public static boolean makeFileExecutableWithChmod(String filePath) {
+        try {
+            Process process = Runtime.getRuntime().exec(new String[]{
+                "chmod", "+x", filePath
+            });
+            
+            int exitCode = process.waitFor();
+            return exitCode == 0;
+            
+        } catch (IOException | InterruptedException e) {
+            System.err.println("chmod命令执行失败: " + e.getMessage());
+            return false;
+        }
+    }
+    
+    /**
+     * 执行命令并获取结果
+     * @param command 要执行的命令数组
+     * @param timeout 超时时间（秒），0表示无超时
+     * @return 执行结果对象
+     */
+    public static CommandResult executeCommand(String[] command, int timeout) {
+        Process process = null;
+        try {
+            // 创建进程
+            ProcessBuilder processBuilder = new ProcessBuilder(command);
+            processBuilder.redirectErrorStream(true); // 合并错误流和输出流
+            process = processBuilder.start();
+            
+            // 读取输出
+            StringBuilder output = new StringBuilder();
+            try (BufferedReader reader = new BufferedReader(
+                 new InputStreamReader(process.getInputStream()))) {
+                
+                String line;
+                while ((line = reader.readLine()) != null) {
+                    output.append(line).append("\n");
+                }
+            }
+            
+            // 等待进程完成
+            boolean finished;
+            if (timeout > 0) {
+                finished = process.waitFor(timeout, TimeUnit.SECONDS);
+            } else {
+                process.waitFor();
+                finished = true;
+            }
+            
+            if (!finished) {
+                process.destroy();
+                return new CommandResult(-1, "命令执行超时", output.toString());
+            }
+            
+            int exitCode = process.exitValue();
+            return new CommandResult(exitCode, exitCode == 0 ? "执行成功" : "执行失败", output.toString());
+            
+        } catch (IOException | InterruptedException e) {
+            return new CommandResult(-1, "执行异常: " + e.getMessage(), "");
+        } finally {
+            if (process != null) {
+                process.destroy();
+            }
+        }
+    }
+    
+    /**
+     * 执行shell脚本文件
+     * @param scriptPath 脚本文件路径
+     * @param args 脚本参数
+     * @param timeout 超时时间（秒）
+     * @return 执行结果
+     */
+    public static CommandResult executeScript(String scriptPath, String[] args, int timeout) {
+        // 首先确保脚本有执行权限
+        if (!makeFileExecutable(scriptPath)) {
+            return new CommandResult(-1, "无法设置执行权限", "");
+        }
+        
+        // 构建命令
+        String[] command = new String[args.length + 1];
+        command[0] = scriptPath;
+        System.arraycopy(args, 0, command, 1, args.length);
+        
+        return executeCommand(command, timeout);
+    }
+    
+    /**
+     * 执行命令的简化方法
+     */
+    public static CommandResult executeCommand(String command, int timeout) {
+        return executeCommand(new String[]{"sh", "-c", command}, timeout);
+    }
+    
+    /**
+     * 执行结果封装类
+     */
+    public static class CommandResult {
+        private final int exitCode;
+        private final String message;
+        private final String output;
+        
+        public CommandResult(int exitCode, String message, String output) {
+            this.exitCode = exitCode;
+            this.message = message;
+            this.output = output;
+        }
+        
+        public int getExitCode() { return exitCode; }
+        public String getMessage() { return message; }
+        public String getOutput() { return output; }
+        public boolean isSuccess() { return exitCode == 0; }
+        
+        @Override
+        public String toString() {
+            return String.format("Exit Code: %d\nMessage: %s\nOutput:\n%s", 
+                exitCode, message, output);
+        }
+    }
+}

src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java

@@ -66,7 +66,7 @@ public class NettyTcpClient {
                         pipeline.addLast("encoder", new StringEncoder());
                         // 添加心跳机制
                         pipeline.addLast("idleStateHandler", 
-                            new IdleStateHandler(0, 0, 90, TimeUnit.SECONDS));
+                            new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
                         // 添加自定义处理器
                         pipeline.addLast("clientHandler", new NettyClientHandler());
                     }

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

@@ -1,41 +1,55 @@
 package com.tongran.agentserver.server.netty.enpoint;
 
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONException;
 import com.alibaba.fastjson2.JSONObject;
 import com.tongran.agentserver.core.enums.MsgEnum;
-import com.tongran.agentserver.scheduler.*;
+import com.tongran.agentserver.server.file.AdvancedAsyncDownloader;
+import com.tongran.agentserver.server.file.AsyncCommandExecutor;
+import com.tongran.agentserver.server.file.LinuxCommandExecutor;
+import com.tongran.agentserver.server.netty.NettyTcpClient;
 import com.tongran.agentserver.server.netty.annotation.AgentDispatcher;
 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 org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 
 @Component
 public class AgentEndpoint {
 
-    @Resource
+//    @Resource
-    private CpuScheduler cpuScheduler;
+//    private CpuScheduler cpuScheduler;
+//
+//    @Resource
+//    private DiskScheduler diskScheduler;
+//
+//    @Resource
+//    private DockerScheduler dockerScheduler;
+//
+//    @Resource
+//    private MemoryScheduler memoryScheduler;
+//
+//    @Resource
+//    private NetScheduler netScheduler;
+//
+//    @Resource
+//    private PointScheduler pointScheduler;
+//
+//    @Resource
+//    private SwitchBoardScheduler switchBoardScheduler;
+//
+//    @Resource
+//    private SysScheduler sysScheduler;
 
     @Resource
-    private DiskScheduler diskScheduler;
+    private NettyTcpClient nettyTcpClient;
-
-    @Resource
-    private DockerScheduler dockerScheduler;
-
-    @Resource
-    private MemoryScheduler memoryScheduler;
-
-    @Resource
-    private NetScheduler netScheduler;
-
-    @Resource
-    private PointScheduler pointScheduler;
-
-    @Resource
-    private SwitchBoardScheduler switchBoardScheduler;
-
-    @Resource
-    private SysScheduler sysScheduler;
 
 
     @AgentDispatcher(msgId = MsgEnum.更新CPU采集间隔)
@@ -44,7 +58,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            cpuScheduler.updateSchedule(25000, intervalMillis);
+//            cpuScheduler.updateSchedule(25000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -58,7 +72,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            diskScheduler.updateSchedule(35000, intervalMillis);
+//            diskScheduler.updateSchedule(35000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -72,7 +86,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            cpuScheduler.updateSchedule(45000, intervalMillis);
+//            dockerScheduler.updateSchedule(45000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -86,7 +100,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            memoryScheduler.updateSchedule(55000, intervalMillis);
+//            memoryScheduler.updateSchedule(55000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -100,7 +114,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            netScheduler.updateSchedule(65000, intervalMillis);
+//            netScheduler.updateSchedule(65000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -114,7 +128,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            pointScheduler.updateSchedule(75000, intervalMillis);
+//            pointScheduler.updateSchedule(75000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -128,7 +142,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            switchBoardScheduler.updateSchedule(85000, intervalMillis);
+//            switchBoardScheduler.updateSchedule(85000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -142,7 +156,7 @@ public class AgentEndpoint {
         public UpMsgResponse upHandle(String data, String clientId) {
             JSONObject jsonObject = JSONObject.parseObject(data);
             long intervalMillis = jsonObject.getLong("intervalMillis");
-            sysScheduler.updateSchedule(40000, intervalMillis);
+//            sysScheduler.updateSchedule(40000, intervalMillis);
             JSONObject json = new JSONObject();
             json.put("clientId",clientId);
             json.put("resCode",1);
@@ -150,5 +164,129 @@ public class AgentEndpoint {
         }
     }
 
+    @AgentDispatcher(msgId = MsgEnum.文件下载)
+    public class DownFileHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = JSONObject.parseObject(data);
+            String fileUrl = jsonObject.getString("fileUrl");
+            String saveDir = "";
+            try {
+                saveDir = jsonObject.getString("saveDir");
+            } catch (JSONException e) {
+                saveDir = "/usr/local/tongran/downloaded";
+            }
+            if(AdvancedAsyncDownloader.createSingleDirectoryIfNotExists(saveDir)){
+                String finalSaveDir = saveDir;
+                AdvancedAsyncDownloader.downloadWithProgress(fileUrl, saveDir, null, progress -> {
+//                    System.out.printf("下载进度: %.1f%%\n", progress);
+                }).thenAccept(filePath -> {
+                    JSONObject json = new JSONObject();
+                    json.put("clientId",clientId);
+                    json.put("saveDir", finalSaveDir);
+                    json.put("fileUrl",fileUrl);
+                    json.put("resCode",1);
+                    Message message = Message.builder().clientId(clientId).dataType(MsgEnum.文件下载应答.getValue()).data(json.toString()).build();
+                    // 将对象转为 JSON 字符串
+                    String msg = "agent-tcp:"+ JSON.toJSONString(message)+"@tong-ran";
+                    nettyTcpClient.sendMessage(msg);
+                    System.out.println("下载完成: " + filePath);
+                }).exceptionally(ex -> {
+                    JSONObject json = new JSONObject();
+                    json.put("clientId",clientId);
+                    json.put("fileUrl",fileUrl);
+                    json.put("resCode",0);
+                    Message message = Message.builder().clientId(clientId).dataType(MsgEnum.文件下载应答.getValue()).data(json.toString()).build();
+                    // 将对象转为 JSON 字符串
+                    String msg = "agent-tcp:"+ JSON.toJSONString(message)+"@tong-ran";
+                    nettyTcpClient.sendMessage(msg);
+                    System.err.println("下载错误: " + ex.getMessage());
+                    return null;
+                });
+            }
+            return null;
+        }
+    }
+
+    @AgentDispatcher(msgId = MsgEnum.执行命令)
+    public class ExecuteCommandHandler implements AgentHandler {
+        @Override
+        public UpMsgResponse upHandle(String data, String clientId) {
+            JSONObject jsonObject = JSONObject.parseObject(data);
+            String scriptPath = "";
+            try {
+                scriptPath = jsonObject.getString("scriptPath");
+            } catch (JSONException e) {
+                scriptPath = "";
+            }
+            String command = jsonObject.getString("command");
+            boolean isFalse = false;
+            if(StringUtils.isNotBlank(scriptPath)){
+                isFalse = true;
+            }
+            if(isFalse){
+                if(LinuxCommandExecutor.makeFileExecutable(scriptPath)){
+                    isFalse = false;
+                }
+            }
+            if(!isFalse){
+                System.out.println("=== 基本异步执行 ===");
+//                List<String> commands = Arrays.asList(
+//                        "/data/agent-server/ksc-andromedae",
+//                        "test_upspeed",
+//                        "-s", "3db15c7e2b86d2cbf91bef7c6ccc93a3",
+//                        "-i", "eth0"
+//                );
+                List<String> commands = Arrays.asList(command.split("\\s+"));
+                CompletableFuture<AsyncCommandExecutor.CommandResult> future =
+                        AsyncCommandExecutor.executeCommandAsync(
+                                commands,
+                                60, TimeUnit.SECONDS);
+                future.thenAccept(result -> {
+                    if (result.isSuccess()) {
+                        System.out.println("脚本执行成功");
+                        System.out.println("[成功resOut] " + result.getOutput());
+                        JSONObject json = new JSONObject();
+                        json.put("clientId",clientId);
+                        json.put("command",command);
+                        json.put("resCode",1);
+                        json.put("resOut",result.getOutput());
+                        json.put("error",result.getError());
+                        Message message = Message.builder().clientId(clientId).dataType(MsgEnum.执行命令应答.getValue()).data(json.toString()).build();
+                        // 将对象转为 JSON 字符串
+                        String msg = "agent-tcp:"+ JSON.toJSONString(message)+"@tong-ran";
+                        nettyTcpClient.sendMessage(msg);
+                    } else {
+                        System.out.println("脚本执行失败: " + result.getError());
+                        System.out.println("[失败resOut] " + result.getOutput());
+                        JSONObject json = new JSONObject();
+                        json.put("clientId",clientId);
+                        json.put("command",command);
+                        json.put("resCode",0);
+                        json.put("resOut",result.getOutput());
+                        json.put("error",result.getError());
+                        Message message = Message.builder().clientId(clientId).dataType(MsgEnum.执行命令应答.getValue()).data(json.toString()).build();
+                        // 将对象转为 JSON 字符串
+                        String msg = "agent-tcp:"+ JSON.toJSONString(message)+"@tong-ran";
+                        nettyTcpClient.sendMessage(msg);
+                    }
+                }).exceptionally(ex -> {
+                    System.err.println("执行失败: " + ex.getMessage());
+                    JSONObject json = new JSONObject();
+                    json.put("clientId",clientId);
+                    json.put("command",command);
+                    json.put("resCode",0);
+                    json.put("error",ex.getMessage());
+                    Message message = Message.builder().clientId(clientId).dataType(MsgEnum.执行命令应答.getValue()).data(json.toString()).build();
+                    // 将对象转为 JSON 字符串
+                    String msg = "agent-tcp:"+ JSON.toJSONString(message)+"@tong-ran";
+                    nettyTcpClient.sendMessage(msg);
+                    return null;
+                });
+            }
+            return null;
+        }
+    }
+
 
 }

src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java

@@ -1,8 +1,12 @@
 package com.tongran.agentserver.server.netty.handler;
 
+import cn.hutool.cache.Cache;
+import cn.hutool.cache.CacheUtil;
+import cn.hutool.core.date.DateUnit;
 import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
+import com.tongran.agentserver.core.session.SessionManager;
 import com.tongran.agentserver.server.netty.annotation.AgentDispatcher;
 import com.tongran.agentserver.server.netty.basics.AgentDispatcherManager;
 import com.tongran.agentserver.server.netty.basics.AgentHandler;
@@ -23,9 +27,17 @@ import java.util.Objects;
 @ChannelHandler.Sharable
 public class DecoderHandler extends ChannelInboundHandlerAdapter {
 
+	protected final SessionManager sessionManager;
+
+	public DecoderHandler() {
+		this.sessionManager = SessionManager.getInstance();
+	}
+
 	@Resource
 	AgentDispatcherManager agentDispatcherManager;
 
+	// 用来临时保留没有处理过的请求报文
+	private final Cache<String, String> lruCache = CacheUtil.newLRUCache(3000);
 
 	/**
 	 * 消息解码器
@@ -36,29 +48,101 @@ public class DecoderHandler extends ChannelInboundHandlerAdapter {
 		if (msg instanceof ByteBuf) {
 			ByteBuf byteBuf = (ByteBuf) msg;
 			String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码
-			AssertLog.info("<<[up]:[up-content]==>{}", messages);
+			AssertLog.info("<<[up1]:[up-content]==>{}", messages);
-			JSONObject jsonObject = JSONObject.parseObject(messages);
+
-			String clientId = jsonObject.getString("clientId");
+			String content = "";
-			String dataType = jsonObject.getString("dataType");
+			boolean isClear = false;
-			AssertLog.info("<<11111111111111==>{}");
+			String tempMsg = lruCache.get(sessionManager.client(ctx));
-			if(Objects.nonNull(agentDispatcherManager)){
+			tempMsg = tempMsg == null ? "" : tempMsg;
-				AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
+			int tmpMsgSize = tempMsg.length();
-				AssertLog.info("<<22222222222222==>{}");
+
-				if (ObjectUtil.isNotEmpty(msgHandler)) {
+			boolean startsWith = messages.startsWith("agent-server:");
-					AssertLog.info("<<3333333333333==>{}");
+			boolean endsWith = messages.endsWith("@tong-ran");
-					UpMsgResponse response = msgHandler.upHandle(messages,clientId);
+			String ms = messages.replaceAll("agent-server:","");
-					AssertLog.info("<<4444444444444==>{}");
+			String[] arr_msg = ms.split("@tong-ran");
-					if(Objects.nonNull(response)){
+			if(startsWith){
-						Message agentMessage = Message.builder().build();
+				//判定是否整包
-						agentMessage.setClientId(clientId);
+				if(arr_msg.length == 1){
-						agentMessage.setDataType(response.getDataType());
+					//判定是否拆包
-						agentMessage.setData(response.getContent());
+					if(endsWith){
-						String json = "agent-tcp:"+ JSON.toJSONString(agentMessage)+"@tong-ran";
+						lruCache.remove(sessionManager.client(ctx));
-						ctx.fireChannelRead(json);//传递到下一个handler
+						tmpMsgSize = 0;
-						AssertLog.info("<<555555555555555==>{}");
+						content = arr_msg[0]+"@tong-ran";
+					}else{
+						lruCache.remove(sessionManager.client(ctx));
+						tmpMsgSize = 0;
+						content = arr_msg[0];
+						lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000);
+					}
+				}
+				//判定是否粘包
+				if(arr_msg.length > 1){
+					if(!endsWith){
+						lruCache.remove(sessionManager.client(ctx));
+						tmpMsgSize = 0;
+						content = arr_msg[0]+"@tong-ran";
+						lruCache.put(sessionManager.client(ctx), arr_msg[1], DateUnit.SECOND.getMillis() * 3000);
+					}else{
+						lruCache.remove(sessionManager.client(ctx));
+						tmpMsgSize = 0;
+						content = String.join("@tong-ran", arr_msg);
 					}
 				}
 			}
+			if (tmpMsgSize > 0) {
+				content = tempMsg + messages;
+				isClear = true;
+			}
+			AssertLog.info("<<[up2]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content);
+			endsWith = content.endsWith("@tong-ran");
+			//判断是否是整包
+			if(endsWith){
+				String[] arr = content.split("@tong-ran");
+				for (String message : arr) {
+					JSONObject jsonObject = JSONObject.parseObject(message);
+					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)) {
+							UpMsgResponse response = msgHandler.upHandle(data, clientId );
+							AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", clientId, dataType, data);
+							if(Objects.nonNull(response)){
+								Message agentMessage = Message.builder().build();
+								agentMessage.setClientId(clientId);
+								agentMessage.setDataType(response.getDataType());
+								agentMessage.setData(response.getContent());
+								String json = "agent-tcp:"+ JSON.toJSONString(agentMessage)+"@tong-ran";
+								ctx.fireChannelRead(json);//传递到下一个handler
+							}
+						}
+					}
+
+				}
+			}
+//			JSONObject jsonObject = JSONObject.parseObject(messages);
+//			String clientId = jsonObject.getString("clientId");
+//			String dataType = jsonObject.getString("dataType");
+//			AssertLog.info("<<11111111111111==>{}");
+//			if(Objects.nonNull(agentDispatcherManager)){
+//				AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
+//				AssertLog.info("<<22222222222222==>{}");
+//				if (ObjectUtil.isNotEmpty(msgHandler)) {
+//					AssertLog.info("<<3333333333333==>{}");
+//					UpMsgResponse response = msgHandler.upHandle(messages,clientId);
+//					AssertLog.info("<<4444444444444==>{}");
+//					if(Objects.nonNull(response)){
+//						Message agentMessage = Message.builder().build();
+//						agentMessage.setClientId(clientId);
+//						agentMessage.setDataType(response.getDataType());
+//						agentMessage.setData(response.getContent());
+//						String json = "agent-tcp:"+ JSON.toJSONString(agentMessage)+"@tong-ran";
+//						ctx.fireChannelRead(json);//传递到下一个handler
+//						AssertLog.info("<<555555555555555==>{}");
+//					}
+//				}
+//			}
 			byteBuf.release(); // 释放 ByteBuf 资源（重要！）
 		} else {
 			System.out.println("Unexpected message type: " + msg.getClass());

src/main/resources/application-dev.yml

@@ -12,10 +12,12 @@ knife4j:
 logging:
   file:
     path: /usr/local/tongran/logs
+#    path: /data/agent-server/logs
 
 netty:
   server:
     host: 172.16.15.103
+#    host: 127.0.0.1
     port: 6610
   client:
     client-id: client-001