From dfda888d9a4adbcf43d458df7bcda17538eaaa90 Mon Sep 17 00:00:00 2001 From: qiminbao Date: Fri, 29 Aug 2025 11:21:07 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=87=E9=9B=86=E7=BD=91=E5=8F=A3=E4=BF=A1?= =?UTF-8?q?=E6=81=AF=E5=A2=9E=E5=8A=A0=20=E5=8D=8F=E5=95=86=E9=80=9F?= =?UTF-8?q?=E5=BA=A6=E3=80=81=E5=B7=A5=E4=BD=9C=E6=A8=A1=E5=BC=8F=20?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=AE=A2=E6=88=B7=E7=AB=AF=E5=A4=84=E7=90=86?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E7=AB=AF=E4=B8=8B=E5=8F=91-=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E4=B8=8B=E8=BD=BD=20=E5=A2=9E=E5=8A=A0=E5=AE=A2?= =?UTF-8?q?=E6=88=B7=E7=AB=AF=E5=A4=84=E7=90=86=E6=9C=8D=E5=8A=A1=E7=AB=AF?= =?UTF-8?q?=E4=B8=8B=E5=8F=91-=E6=89=A7=E8=A1=8C=E5=91=BD=E4=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../agentserver/core/enums/MsgEnum.java | 10 +- .../collect/net/impl/NetServiceImpl.java | 43 ++- .../agentserver/server/collect/vo/NetVO.java | 7 +- .../server/file/AdvancedAsyncDownloader.java | 121 ++++++++ .../server/file/AsyncCommandExecutor.java | 270 ++++++++++++++++++ .../server/file/LinuxCommandExecutor.java | 175 ++++++++++++ .../server/netty/NettyTcpClient.java | 2 +- .../server/netty/enpoint/AgentEndpoint.java | 198 +++++++++++-- .../server/netty/handler/DecoderHandler.java | 124 ++++++-- src/main/resources/application-dev.yml | 2 + 10 files changed, 895 insertions(+), 57 deletions(-) create mode 100644 src/main/java/com/tongran/agentserver/server/file/AdvancedAsyncDownloader.java create mode 100644 src/main/java/com/tongran/agentserver/server/file/AsyncCommandExecutor.java create mode 100644 src/main/java/com/tongran/agentserver/server/file/LinuxCommandExecutor.java diff --git a/src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java b/src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java index 62b85c3..582d567 100644 --- a/src/main/java/com/tongran/agentserver/core/enums/MsgEnum.java +++ b/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; diff --git a/src/main/java/com/tongran/agentserver/server/collect/net/impl/NetServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/net/impl/NetServiceImpl.java index 32717d3..1a52dd5 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/net/impl/NetServiceImpl.java +++ b/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.util.List; -import java.util.Objects; +import java.io.BufferedReader; +import java.io.InputStreamReader; +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 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 getNetworkMode(String interfaceName) { + Map 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; + } + } diff --git a/src/main/java/com/tongran/agentserver/server/collect/vo/NetVO.java b/src/main/java/com/tongran/agentserver/server/collect/vo/NetVO.java index bfdbc89..6c90219 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/vo/NetVO.java +++ b/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 = "接收速度") -// private String speed; + @Schema(description = "协商速度") + private String speed; + + @Schema(description = "工作模式") + private String duplex; @Schema(description = "时间戳") private long timestamp; diff --git a/src/main/java/com/tongran/agentserver/server/file/AdvancedAsyncDownloader.java b/src/main/java/com/tongran/agentserver/server/file/AdvancedAsyncDownloader.java new file mode 100644 index 0000000..7a0d4e1 --- /dev/null +++ b/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 downloadWithProgress( + String fileUrl, + String saveDir, + String fileName, + Consumer 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("异步下载已启动..."); + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/server/file/AsyncCommandExecutor.java b/src/main/java/com/tongran/agentserver/server/file/AsyncCommandExecutor.java new file mode 100644 index 0000000..fb3f374 --- /dev/null +++ b/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 makeFileExecutableAsync(String filePath) { + return CompletableFuture.supplyAsync(() -> { + try { + Path path = Paths.get(filePath); + + if (!Files.exists(path)) { + throw new RuntimeException("文件不存在: " + filePath); + } + + Set 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 executeCommandAsync( + List 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 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 executeCommandWithRealtimeOutput( + String[] command, + long timeout, + TimeUnit timeUnit, + Consumer outputConsumer, + Consumer errorConsumer) { + + return CompletableFuture.supplyAsync(() -> { + Process process = null; + try { + ProcessBuilder processBuilder = new ProcessBuilder(command); + process = processBuilder.start(); + + // 启动输出读取线程 + CompletableFuture outputFuture = readStreamAsync( + process.getInputStream(), outputConsumer, "OUTPUT"); + + CompletableFuture 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 readStreamAsync( + InputStream inputStream, + Consumer 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 executeScriptAsync( + String scriptPath, + String[] args, + long timeout, + TimeUnit timeUnit, + Consumer 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> executeCommandsAsync( +// List 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); + } + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/server/file/LinuxCommandExecutor.java b/src/main/java/com/tongran/agentserver/server/file/LinuxCommandExecutor.java new file mode 100644 index 0000000..40fce42 --- /dev/null +++ b/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 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); + } + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java b/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java index 9881352..db42f90 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java +++ b/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()); } diff --git a/src/main/java/com/tongran/agentserver/server/netty/enpoint/AgentEndpoint.java b/src/main/java/com/tongran/agentserver/server/netty/enpoint/AgentEndpoint.java index cf96005..d8e9ece 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/enpoint/AgentEndpoint.java +++ b/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 - private CpuScheduler cpuScheduler; +// @Resource +// 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; - - @Resource - private DockerScheduler dockerScheduler; - - @Resource - private MemoryScheduler memoryScheduler; - - @Resource - private NetScheduler netScheduler; - - @Resource - private PointScheduler pointScheduler; - - @Resource - private SwitchBoardScheduler switchBoardScheduler; - - @Resource - private SysScheduler sysScheduler; + private NettyTcpClient nettyTcpClient; @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 commands = Arrays.asList( +// "/data/agent-server/ksc-andromedae", +// "test_upspeed", +// "-s", "3db15c7e2b86d2cbf91bef7c6ccc93a3", +// "-i", "eth0" +// ); + List commands = Arrays.asList(command.split("\\s+")); + CompletableFuture 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; + } + } + } \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java b/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java index 29e2e89..d8d0b99 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java +++ b/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java @@ -1,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 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); - 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==>{}"); + AssertLog.info("<<[up1]:[up-content]==>{}", messages); + + String content = ""; + boolean isClear = false; + String tempMsg = lruCache.get(sessionManager.client(ctx)); + tempMsg = tempMsg == null ? "" : tempMsg; + int tmpMsgSize = tempMsg.length(); + + boolean startsWith = messages.startsWith("agent-server:"); + boolean endsWith = messages.endsWith("@tong-ran"); + String ms = messages.replaceAll("agent-server:",""); + String[] arr_msg = ms.split("@tong-ran"); + if(startsWith){ + //判定是否整包 + if(arr_msg.length == 1){ + //判定是否拆包 + if(endsWith){ + lruCache.remove(sessionManager.client(ctx)); + tmpMsgSize = 0; + 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()); diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index 8813dbb..e1e24ef 100644 --- a/src/main/resources/application-dev.yml +++ b/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