From 8b5fb8c401530746a47c25299999c3c712abb390 Mon Sep 17 00:00:00 2001 From: qiminbao Date: Wed, 3 Sep 2025 15:30:13 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=95=B0=E6=8D=AE=E9=87=87?= =?UTF-8?q?=E9=9B=86=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../agentserver/AgentServerApplication.java | 3 +- .../agentserver/scheduler/CpuScheduler.java | 102 ------ .../agentserver/scheduler/DiskScheduler.java | 104 ------ .../scheduler/DockerScheduler.java | 104 ------ .../agentserver/scheduler/HeartScheduler.java | 113 ------- .../scheduler/MemoryScheduler.java | 103 ------ .../agentserver/scheduler/NetScheduler.java | 105 ------ .../agentserver/scheduler/PointScheduler.java | 105 ------ .../scheduler/SwitchBoardScheduler.java | 105 ------ .../agentserver/scheduler/SysScheduler.java | 102 ------ .../scheduler/config/SchedulerConfig.java | 36 +++ .../scheduler/service/AppInitializer.java | 77 +++++ .../scheduler/service/BusinessTasks.java | 302 ++++++++++++++++++ .../scheduler/service/CompensatedTrigger.java | 56 ++++ .../scheduler/service/DynamicTaskService.java | 144 +++++++++ .../service/SchedulerController.java | 62 ++++ .../scheduler/task/SpecificTimeRequest.java | 52 +++ .../task/SpecificTimeTaskConfig.java | 114 +++++++ .../task/SpecificTimeTaskService.java | 120 +++++++ .../collect/cpu/impl/CpuServiceImpl.java | 4 +- .../server/collect/disk/DiskService.java | 4 +- .../collect/disk/impl/DiskServiceImpl.java | 6 +- .../server/collect/docker/DockerService.java | 2 +- .../docker/impl/DockerServiceImpl.java | 3 +- .../memory/impl/MemoryServiceImpl.java | 3 +- .../server/collect/net/NetService.java | 2 +- .../collect/net/impl/NetServiceImpl.java | 3 +- .../switchboard/SwitchBoardService.java | 2 +- .../impl/SwitchBoardServiceImpl.java | 7 +- .../system/impl/SystemServiceImpl.java | 3 +- .../server/netty/enpoint/AgentEndpoint.java | 208 ++++++------ .../server/netty/handler/DecoderHandler.java | 30 +- .../server/netty/model/ScriptPolicy.java | 4 + .../tongran/agentserver/utils/AgentUtil.java | 13 + .../utils/Java8FileDownloader.java | 182 +++++++++++ 35 files changed, 1313 insertions(+), 1072 deletions(-) delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/DiskScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/HeartScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/MemoryScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/PointScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java delete mode 100644 src/main/java/com/tongran/agentserver/scheduler/SysScheduler.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/config/SchedulerConfig.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/service/AppInitializer.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/service/BusinessTasks.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/service/CompensatedTrigger.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/service/DynamicTaskService.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/service/SchedulerController.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeRequest.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskConfig.java create mode 100644 src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskService.java create mode 100644 src/main/java/com/tongran/agentserver/utils/Java8FileDownloader.java diff --git a/src/main/java/com/tongran/agentserver/AgentServerApplication.java b/src/main/java/com/tongran/agentserver/AgentServerApplication.java index 33d9db6..d528985 100644 --- a/src/main/java/com/tongran/agentserver/AgentServerApplication.java +++ b/src/main/java/com/tongran/agentserver/AgentServerApplication.java @@ -3,9 +3,11 @@ package com.tongran.agentserver; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.ComponentScan; +import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @ComponentScan(basePackages = {"com.tongran.agentserver", "cn.hutool.extra.spring"}) +@EnableScheduling public class AgentServerApplication { public static void main(String[] args) { @@ -13,5 +15,4 @@ public class AgentServerApplication { } - } diff --git a/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java deleted file mode 100644 index 2bb9ea6..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java +++ /dev/null @@ -1,102 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.tongran.agentserver.server.collect.cpu.CpuService; -import com.tongran.agentserver.server.collect.vo.CpuVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class CpuScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:25000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private CpuService cpuService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public CpuScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动CPU信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("CPU信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送CPU信息包 - CpuVO cpuVO = cpuService.query(); - String data = JSON.toJSONString(cpuVO); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("CPU").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/DiskScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/DiskScheduler.java deleted file mode 100644 index 4c5d7dd..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/DiskScheduler.java +++ /dev/null @@ -1,104 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONArray; -import com.tongran.agentserver.server.collect.disk.DiskService; -import com.tongran.agentserver.server.collect.vo.DiskVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.List; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class DiskScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:35000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private DiskService diskService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public DiskScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动磁盘信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("磁盘信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送磁盘信息包 - List list = diskService.queryDiskList(); - String data = JSONArray.toJSONString(list); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("DISK").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java deleted file mode 100644 index 7a10722..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java +++ /dev/null @@ -1,104 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONArray; -import com.tongran.agentserver.server.collect.docker.DockerService; -import com.tongran.agentserver.server.collect.vo.DockerVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.List; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class DockerScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:45000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private DockerService dockerService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public DockerScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动容器信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("容器信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送容器包 - List list = dockerService.query(); - String data = JSONArray.toJSONString(list); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("DOCKER").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/HeartScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/HeartScheduler.java deleted file mode 100644 index 5dbd3ec..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/HeartScheduler.java +++ /dev/null @@ -1,113 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONObject; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class HeartScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:5000}") - private long initialDelay; - - @Value("${task.interval:30000}") - private long interval; - - @Resource - private NettyTcpClient nettyTcpClient; - - public HeartScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动心跳定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("心跳定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (!nettyTcpClient.isConnected()) { - try { - nettyTcpClient.connect(); - // 等待10秒 - try { - Thread.sleep(10000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - // 发送心跳包 - long timestamp = System.currentTimeMillis(); - JSONObject object = new JSONObject(); - object.put("strength","31"); - object.put("timestamp",timestamp); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("HEARTBEAT").data(object.toString()).build(); - // 将对象转为 JSON 字符串 标识 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - AssertLog.info("发送心跳包={}",json); - nettyTcpClient.sendMessage(json); - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } - -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/MemoryScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/MemoryScheduler.java deleted file mode 100644 index b931df1..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/MemoryScheduler.java +++ /dev/null @@ -1,103 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.tongran.agentserver.server.collect.memory.MemoryService; -import com.tongran.agentserver.server.collect.vo.MemoryVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class MemoryScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:55000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private MemoryService memoryService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public MemoryScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动内存信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("内存信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送内存信息包 - MemoryVO memoryVO = memoryService.query(); - String data = JSON.toJSONString(memoryVO); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("MEMORY").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - AssertLog.info("发送内存信息包={}",json); - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java deleted file mode 100644 index b3c9c96..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java +++ /dev/null @@ -1,105 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONArray; -import com.tongran.agentserver.server.collect.net.NetService; -import com.tongran.agentserver.server.collect.vo.NetVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.List; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class NetScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:65000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private NetService netService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public NetScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动网络信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("网络信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送网卡信息包 - List list = netService.query(); - String data = JSONArray.toJSONString(list); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("NET").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - AssertLog.info("发送网卡信息包={}",json); - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/PointScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/PointScheduler.java deleted file mode 100644 index fb43376..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/PointScheduler.java +++ /dev/null @@ -1,105 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONArray; -import com.tongran.agentserver.server.collect.disk.DiskService; -import com.tongran.agentserver.server.collect.vo.PointVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.List; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class PointScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:75000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private DiskService diskService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public PointScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动挂载信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("挂载信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送挂载点信息包 - List list = diskService.queryPointList(); - String data = JSONArray.toJSONString(list); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("POINT").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - AssertLog.info("发送挂载点信息包={}",json); - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java deleted file mode 100644 index 05d68d0..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java +++ /dev/null @@ -1,105 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONArray; -import com.tongran.agentserver.server.collect.switchboard.SwitchBoardService; -import com.tongran.agentserver.server.collect.vo.SwitchBoardVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.List; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class SwitchBoardScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:85000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private SwitchBoardService switchBoardService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public SwitchBoardScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动交换机信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("交换机信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送交换机信息包 - List list = switchBoardService.query(); - String data = JSONArray.toJSONString(list); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("SWITCHBOARD").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } - -} diff --git a/src/main/java/com/tongran/agentserver/scheduler/SysScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/SysScheduler.java deleted file mode 100644 index 3a62bfb..0000000 --- a/src/main/java/com/tongran/agentserver/scheduler/SysScheduler.java +++ /dev/null @@ -1,102 +0,0 @@ -package com.tongran.agentserver.scheduler; - -import com.alibaba.fastjson2.JSON; -import com.tongran.agentserver.server.collect.system.SystemService; -import com.tongran.agentserver.server.collect.vo.SystemVO; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import com.tongran.agentserver.server.netty.model.Message; -import com.tongran.agentserver.utils.AgentUtil; -import com.tongran.agentserver.utils.AssertLog; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.context.event.ApplicationReadyEvent; -import org.springframework.context.event.EventListener; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.stereotype.Component; - -import javax.annotation.PreDestroy; -import javax.annotation.Resource; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.concurrent.ScheduledFuture; - -@Component -@EnableScheduling -public class SysScheduler { - - private final TaskScheduler taskScheduler; - - private ScheduledFuture currentTask; - - private final Object lock = new Object(); - - @Value("${task.initial.delay:40000}") - private long initialDelay; - - @Value("${task.interval:300000}") - private long interval; - - @Resource - private SystemService systemService; - - @Resource - private NettyTcpClient nettyTcpClient; - - public SysScheduler(TaskScheduler taskScheduler) { - this.taskScheduler = taskScheduler; - } - - @EventListener(ApplicationReadyEvent.class) - public void startDynamicTask() { - AssertLog.info("启动系统信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); - scheduleTask(initialDelay, interval); - } - - public void scheduleTask(long delayMillis, long intervalMillis) { - synchronized (lock) { - // 取消现有任务 - if (currentTask != null) { - currentTask.cancel(false); - } - - // 安排新任务 - currentTask = taskScheduler.scheduleWithFixedDelay( - this::dynamicPeriodicTask, - Instant.now().plusMillis(delayMillis), - Duration.ofMillis(intervalMillis) - ); - } - } - - public void dynamicPeriodicTask() { - AssertLog.info("系统信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); - performDynamicTask(); - } - - private void performDynamicTask() { - // 判定客户端与服务端是否连接 - if (nettyTcpClient.isConnected()) { - // 发送系统信息包 - SystemVO systemVO = systemService.query(); - String data = JSON.toJSONString(systemVO); - String clientId = AgentUtil.getMotherboardUUID(); - Message message = Message.builder().clientId(clientId).dataType("SYSTEM").data(data).build(); - // 将对象转为 JSON 字符串 - String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; - nettyTcpClient.sendMessage(json); - } - } - - public void updateSchedule(long newDelay, long newInterval) { - AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); - scheduleTask(newDelay, newInterval); - } - - @PreDestroy - public void cleanup() { - if (currentTask != null) { - currentTask.cancel(false); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/config/SchedulerConfig.java b/src/main/java/com/tongran/agentserver/scheduler/config/SchedulerConfig.java new file mode 100644 index 0000000..4ed39cc --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/config/SchedulerConfig.java @@ -0,0 +1,36 @@ +package com.tongran.agentserver.scheduler.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import java.util.concurrent.ThreadPoolExecutor; + +@Configuration +@EnableScheduling +@EnableAsync +public class SchedulerConfig { + + @Bean + public ThreadPoolTaskScheduler taskScheduler() { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setPoolSize(10); // 核心线程数 + scheduler.setThreadNamePrefix("dynamic-scheduler-"); + scheduler.setAwaitTerminationSeconds(60); + scheduler.setWaitForTasksToCompleteOnShutdown(true); + scheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + scheduler.setRemoveOnCancelPolicy(true); + return scheduler; + } + + @Bean("taskExecutor") + public ThreadPoolTaskScheduler taskExecutor() { + ThreadPoolTaskScheduler executor = new ThreadPoolTaskScheduler(); + executor.setPoolSize(10); // 工作线程数 + executor.setThreadNamePrefix("task-worker-"); + executor.setAwaitTerminationSeconds(30); + executor.setWaitForTasksToCompleteOnShutdown(true); + return executor; + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/service/AppInitializer.java b/src/main/java/com/tongran/agentserver/scheduler/service/AppInitializer.java new file mode 100644 index 0000000..be16a48 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/service/AppInitializer.java @@ -0,0 +1,77 @@ +package com.tongran.agentserver.scheduler.service; + +import com.tongran.agentserver.utils.AssertLog; +import org.springframework.boot.CommandLineRunner; +import org.springframework.context.annotation.Lazy; +import org.springframework.stereotype.Component; + +@Component +public class AppInitializer implements CommandLineRunner { + + private final BusinessTasks businessTasks; + private final DynamicTaskService dynamicTaskService; + + // 使用构造函数注入，避免循环依赖 + public AppInitializer(DynamicTaskService dynamicTaskService, + @Lazy BusinessTasks businessTasks) { + this.dynamicTaskService = dynamicTaskService; + this.businessTasks = businessTasks; + } + + @Override + public void run(String... args) throws Exception { + System.out.println("应用启动完成，开始初始化定时任务..."); + // 初始化定时任务 + initScheduledTasks(); + System.out.println("定时任务初始化完成"); + } + + private void initScheduledTasks() { + // 每30秒执行心跳任务 + AssertLog.info("初始化启动心跳定时任务 - 延迟: {}ms, 间隔: {}ms", 5000, 30000); + dynamicTaskService.scheduleTask("heartbeat", + businessTasks::heartbeatTask, 5000, 30000); + + // 每300秒执行CPU信息采集任务 + AssertLog.info("初始化启动CPU信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 25000, 300000); + dynamicTaskService.scheduleTask("cpu", + businessTasks::cpuTask, 25000, 300000); // 25秒后开始，每300秒执行 + + // 每300秒执行磁盘信息采集任务 + AssertLog.info("初始化启动磁盘信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 35000, 300000); + dynamicTaskService.scheduleTask("disk", + businessTasks::diskTask, 35000, 300000); // 35秒后开始，每300秒执行 + + // 每300秒执行系统信息采集任务 + AssertLog.info("初始化启动系统信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 40000, 300000); + dynamicTaskService.scheduleTask("system", + businessTasks::systemTask, 40000, 300000); // 40秒后开始，每300秒执行 + + // 每300秒执行容器信息采集任务 + AssertLog.info("初始化启动容器信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 45000, 300000); + dynamicTaskService.scheduleTask("docker", + businessTasks::dockerTask, 45000, 300000); // 45秒后开始，每300秒执行 + + // 每300秒执行内存信息采集任务 + AssertLog.info("初始化启动内存信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 55000, 300000); + dynamicTaskService.scheduleTask("memory", + businessTasks::memoryTask, 55000, 300000); // 55秒后开始，每300秒执行 + + // 每300秒执行网络信息采集任务 + AssertLog.info("初始化启动网络信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 65000, 300000); + dynamicTaskService.scheduleTask("net", + businessTasks::netTask, 65000, 300000); // 65秒后开始，每300秒执行 + + // 每300秒执行挂载信息采集任务 + AssertLog.info("初始化启动挂载信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 75000, 300000); + dynamicTaskService.scheduleTask("point", + businessTasks::pointTask, 75000, 300000); // 75秒后开始，每300秒执行 + + // 每300秒执行交换机信息采集任务 + AssertLog.info("初始化启动交换机信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", 85000, 300000); + dynamicTaskService.scheduleTask("switchBoard", + businessTasks::switchBoardTask, 85000, 300000); // 85秒后开始，每300秒执行 + + + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/service/BusinessTasks.java b/src/main/java/com/tongran/agentserver/scheduler/service/BusinessTasks.java new file mode 100644 index 0000000..64c74ef --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/service/BusinessTasks.java @@ -0,0 +1,302 @@ +package com.tongran.agentserver.scheduler.service; + +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.tongran.agentserver.server.collect.cpu.CpuService; +import com.tongran.agentserver.server.collect.disk.DiskService; +import com.tongran.agentserver.server.collect.docker.DockerService; +import com.tongran.agentserver.server.collect.memory.MemoryService; +import com.tongran.agentserver.server.collect.net.NetService; +import com.tongran.agentserver.server.collect.switchboard.SwitchBoardService; +import com.tongran.agentserver.server.collect.system.SystemService; +import com.tongran.agentserver.server.collect.vo.*; +import com.tongran.agentserver.server.netty.NettyTcpClient; +import com.tongran.agentserver.server.netty.model.Message; +import com.tongran.agentserver.utils.AgentUtil; +import com.tongran.agentserver.utils.AssertLog; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; +import java.time.LocalDateTime; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +@Component +public class BusinessTasks { + + private final AtomicInteger task1Counter = new AtomicInteger(0); + private final AtomicInteger task2Counter = new AtomicInteger(0); + private final AtomicInteger task3Counter = new AtomicInteger(0); + private final AtomicInteger task4Counter = new AtomicInteger(0); + private final AtomicInteger task5Counter = new AtomicInteger(0); + private final AtomicInteger task6Counter = new AtomicInteger(0); + private final AtomicInteger task7Counter = new AtomicInteger(0); + private final AtomicInteger task8Counter = new AtomicInteger(0); + private final AtomicInteger task9Counter = new AtomicInteger(0); + + @Resource + private NettyTcpClient nettyTcpClient; + + @Resource + private CpuService cpuService; + + @Resource + private DiskService diskService; + + @Resource + private DockerService dockerService; + + @Resource + private MemoryService memoryService; + + @Resource + private NetService netService; + + @Resource + private SwitchBoardService switchBoardService; + + @Resource + private SystemService systemService; + + /** + * 任务1：心跳上报任务 + */ + @Async("taskExecutor") + public void heartbeatTask() { + long timestamp = System.currentTimeMillis(); + int count = task1Counter.incrementAndGet(); + AssertLog.info("心跳定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 业务处理 + // 判定客户端与服务端是否连接 + if (!nettyTcpClient.isConnected()) { + try { + nettyTcpClient.connect(); + // 等待10秒 + try { + Thread.sleep(10000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + // 发送心跳包 + JSONObject object = new JSONObject(); + object.put("strength","31"); + object.put("timestamp",timestamp); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("HEARTBEAT").data(object.toString()).build(); + // 将对象转为 JSON 字符串 标识 + String json = "agent-tcp:"+ JSON.toJSONString(message)+"@tong-ran"; + AssertLog.info("发送心跳包={}",json); + nettyTcpClient.sendMessage(json); + AssertLog.info("心跳定时任务执行 - task #{} completed", count); + } + + /** + * 任务2：cpu信息采集任务 + */ + @Async("taskExecutor") + public void cpuTask() { + long timestamp = System.currentTimeMillis(); + int count = task2Counter.incrementAndGet(); + AssertLog.info("CPU信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送CPU信息包 + CpuVO cpuVO = cpuService.query(); + cpuVO.setTimestamp(timestamp); + String data = JSON.toJSONString(cpuVO); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("CPU").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + nettyTcpClient.sendMessage(json); + } + AssertLog.info("CPU信息采集定时任务执行 - task #{} completed", count); + } + + /** + * 任务3：磁盘信息采集任务 + */ + @Async("taskExecutor") + public void diskTask() { + long timestamp = System.currentTimeMillis(); + int count = task3Counter.incrementAndGet(); + AssertLog.info("磁盘信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送磁盘信息包 + List list = diskService.queryDiskList(timestamp); + String data = JSONArray.toJSONString(list); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("DISK").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + nettyTcpClient.sendMessage(json); + } + AssertLog.info("磁盘信息采集定时任务执行 - task #{} completed", count); + } + + /** + * 任务4：容器信息采集任务 + */ + @Async("taskExecutor") + public void dockerTask() { + long timestamp = System.currentTimeMillis(); + int count = task4Counter.incrementAndGet(); + AssertLog.info("容器信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送容器包 + List list = dockerService.query(timestamp); + String data = JSONArray.toJSONString(list); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("DOCKER").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + nettyTcpClient.sendMessage(json); + } + AssertLog.info("容器信息采集定时任务执行 - task #{} completed", count); + } + + /** + * 任务5：内存信息采集任务 + */ + @Async("taskExecutor") + public void memoryTask() { + long timestamp = System.currentTimeMillis(); + int count = task5Counter.incrementAndGet(); + AssertLog.info("内存信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送内存信息包 + MemoryVO memoryVO = memoryService.query(); + memoryVO.setTimestamp(timestamp); + String data = JSON.toJSONString(memoryVO); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("MEMORY").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + AssertLog.info("发送内存信息包={}",json); + nettyTcpClient.sendMessage(json); + } + AssertLog.info("内存信息采集定时任务执行 - task #{} completed", count); + } + + /** + * 任务6：网络信息采集任务 + */ + @Async("taskExecutor") + public void netTask() { + long timestamp = System.currentTimeMillis(); + int count = task6Counter.incrementAndGet(); + AssertLog.info("网络信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送网卡信息包 + List list = netService.query(timestamp); + String data = JSONArray.toJSONString(list); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("NET").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + AssertLog.info("发送网卡信息包={}",json); + nettyTcpClient.sendMessage(json); + } + AssertLog.info("网络信息采集定时任务执行 - task #{} completed", count); + } + + /** + * 任务7：挂载信息采集任务 + */ + @Async("taskExecutor") + public void pointTask() { + long timestamp = System.currentTimeMillis(); + int count = task7Counter.incrementAndGet(); + AssertLog.info("挂载信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送挂载点信息包 + List list = diskService.queryPointList(timestamp); + String data = JSONArray.toJSONString(list); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("POINT").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + AssertLog.info("发送挂载点信息包={}",json); + nettyTcpClient.sendMessage(json); + } + AssertLog.info("挂载点信息采集定时任务执行 - task #{} completed", count); + } + + /** + * 任务8：交换机信息采集任务 + */ + @Async("taskExecutor") + public void switchBoardTask() { + long timestamp = System.currentTimeMillis(); + int count = task8Counter.incrementAndGet(); + AssertLog.info("交换机信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送交换机信息包 + List list = switchBoardService.query(timestamp); + String data = JSONArray.toJSONString(list); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("SWITCHBOARD").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + nettyTcpClient.sendMessage(json); + } + AssertLog.info("交换机信息采集定时任务执行 - task #{} completed", count); + } + + /** + * 任务9：系统机信息采集任务 + */ + @Async("taskExecutor") + public void systemTask() { + long timestamp = System.currentTimeMillis(); + int count = task9Counter.incrementAndGet(); + AssertLog.info("系统信息采集定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (nettyTcpClient.isConnected()) { + // 发送系统信息包 + SystemVO systemVO = systemService.query(); + systemVO.setTimestamp(timestamp); + String data = JSON.toJSONString(systemVO); + String clientId = AgentUtil.getMotherboardUUID(); + Message message = Message.builder().clientId(clientId).dataType("SYSTEM").data(data).build(); + // 将对象转为 JSON 字符串 + String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran"; + nettyTcpClient.sendMessage(json); + } + AssertLog.info("系统信息采集定时任务执行 - task #{} completed", count); + } + + + + + + /** + * 初始化定时任务 + */ +// @PostConstruct +// public void initScheduledTasks() { +// // 每10秒执行数据处理任务 +// dynamicTaskService.scheduleTask("heartbeat", +// this::dataProcessingTask, 5000, 10000); +// +// // 每15秒执行日志清理任务 +// dynamicTaskService.scheduleTask("cpu", +// this::logCleanupTask, 10000, 15000); // 10秒后开始，每60秒执行 +// +// // 每10秒执行监控任务 +// dynamicTaskService.scheduleTask("monitoring", +// this::monitoringTask, 5000, 10000); +// } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/service/CompensatedTrigger.java b/src/main/java/com/tongran/agentserver/scheduler/service/CompensatedTrigger.java new file mode 100644 index 0000000..9684fc6 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/service/CompensatedTrigger.java @@ -0,0 +1,56 @@ +package com.tongran.agentserver.scheduler.service; + +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import java.util.Date; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +public class CompensatedTrigger implements Trigger { + + private final long period; + private final AtomicLong nextExecutionTime; + private final TimeUnit timeUnit; + + public CompensatedTrigger(long initialDelay, long period) { + this(initialDelay, period, TimeUnit.MILLISECONDS); + } + + public CompensatedTrigger(long initialDelay, long period, TimeUnit timeUnit) { + this.period = period; + this.timeUnit = timeUnit; + this.nextExecutionTime = new AtomicLong( + System.currentTimeMillis() + timeUnit.toMillis(initialDelay) + ); + } + + @Override + public Date nextExecutionTime(TriggerContext triggerContext) { + long lastCompletionTime = triggerContext.lastCompletionTime() != null + ? triggerContext.lastCompletionTime().getTime() + : System.currentTimeMillis(); + + // 计算下一次执行时间（考虑补偿） + long currentNextTime = nextExecutionTime.get(); + long now = System.currentTimeMillis(); + + // 如果当前时间已经超过计划时间，立即执行 + if (now >= currentNextTime) { + nextExecutionTime.set(now + timeUnit.toMillis(period)); + return new Date(now); + } + + // 否则按计划时间执行 + nextExecutionTime.set(currentNextTime + timeUnit.toMillis(period)); + return new Date(currentNextTime); + } + + public void updatePeriod(long newPeriod) { + // 更新执行间隔 + long currentTime = System.currentTimeMillis(); + long remaining = nextExecutionTime.get() - currentTime; + if (remaining > 0) { + nextExecutionTime.set(currentTime + timeUnit.toMillis(newPeriod)); + } + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/service/DynamicTaskService.java b/src/main/java/com/tongran/agentserver/scheduler/service/DynamicTaskService.java new file mode 100644 index 0000000..d5a6261 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/service/DynamicTaskService.java @@ -0,0 +1,144 @@ +package com.tongran.agentserver.scheduler.service; + +import lombok.AllArgsConstructor; +import lombok.Data; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.stereotype.Service; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicLong; + +@Service +public class DynamicTaskService { + + @Autowired + private TaskScheduler taskScheduler; + + private final Map> taskFutures = new ConcurrentHashMap<>(); + private final Map taskConfigs = new ConcurrentHashMap<>(); + private final Map taskExecutions = new ConcurrentHashMap<>(); + private final Map lastExecutionTimes = new ConcurrentHashMap<>(); + + /** + * 添加或更新定时任务 + */ + public void scheduleTask(String taskId, Runnable task, long initialDelay, long period) { + // 取消现有任务 + cancelTask(taskId); + + TaskConfig config = new TaskConfig(initialDelay, period, System.currentTimeMillis()); + taskConfigs.put(taskId, config); + taskExecutions.put(taskId, new AtomicLong(0)); + + // 使用补偿机制调度任务 + ScheduledFuture future = taskScheduler.schedule( + createCompensatedTask(taskId, task), + new CompensatedTrigger(initialDelay, period) + ); + + taskFutures.put(taskId, future); + } + + /** + * 创建带补偿的任务 + */ + private Runnable createCompensatedTask(String taskId, Runnable originalTask) { + return () -> { + long startTime = System.currentTimeMillis(); + lastExecutionTimes.put(taskId, startTime); + + try { + originalTask.run(); + } catch (Exception e) { + System.err.println("Task " + taskId + " execution failed: " + e.getMessage()); + } finally { + long endTime = System.currentTimeMillis(); + long executionTime = endTime - startTime; + taskExecutions.get(taskId).incrementAndGet(); + + System.out.printf("Task %s executed in %dms, total executions: %d%n", + taskId, executionTime, taskExecutions.get(taskId).get()); + } + }; + } + + /** + * 取消任务 + */ + public boolean cancelTask(String taskId) { + ScheduledFuture future = taskFutures.remove(taskId); + if (future != null) { + future.cancel(false); + taskConfigs.remove(taskId); + taskExecutions.remove(taskId); + lastExecutionTimes.remove(taskId); + return true; + } + return false; + } + + /** + * 更新任务间隔 + */ + public void updateTaskInterval(String taskId, long newPeriod) { + TaskConfig config = taskConfigs.get(taskId); + if (config != null) { + Runnable task = () -> {}; // 这里需要根据实际情况获取原始任务 + scheduleTask(taskId, task, 0, newPeriod); // 立即重新调度 + } + } + + /** + * 获取任务状态 + */ + public TaskStatus getTaskStatus(String taskId) { + TaskConfig config = taskConfigs.get(taskId); + AtomicLong executions = taskExecutions.get(taskId); + Long lastTime = lastExecutionTimes.get(taskId); + + if (config != null && executions != null) { + return new TaskStatus( + taskId, + config.getPeriod(), + executions.get(), + lastTime, + taskFutures.get(taskId) != null && !taskFutures.get(taskId).isCancelled() + ); + } + return null; + } + + /** + * 获取所有任务状态 + */ + public Map getAllTaskStatus() { + Map statusMap = new ConcurrentHashMap<>(); + for (String taskId : taskConfigs.keySet()) { + statusMap.put(taskId, getTaskStatus(taskId)); + } + return statusMap; + } + + // 配置类 + @Data + @AllArgsConstructor + public static class TaskConfig { + private long initialDelay; + private long period; + private long createTime; + } + + // 状态类 + @Data + @AllArgsConstructor + public static class TaskStatus { + private String taskId; + private long interval; + private long executionCount; + private Long lastExecutionTime; + private boolean isRunning; + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/service/SchedulerController.java b/src/main/java/com/tongran/agentserver/scheduler/service/SchedulerController.java new file mode 100644 index 0000000..d96c8bd --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/service/SchedulerController.java @@ -0,0 +1,62 @@ +package com.tongran.agentserver.scheduler.service; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.*; +import java.util.Map; + +@RestController +@RequestMapping("/api/scheduler") +public class SchedulerController { + + @Autowired + private DynamicTaskService dynamicTaskService; + + /** + * 获取所有任务状态 + */ + @GetMapping("/tasks") + public Map getAllTasks() { + return dynamicTaskService.getAllTaskStatus(); + } + + /** + * 更新任务间隔 + */ + @PostMapping("/{taskId}/interval") + public String updateInterval(@PathVariable String taskId, + @RequestParam long intervalMs) { + dynamicTaskService.updateTaskInterval(taskId, intervalMs); + return "Task " + taskId + " interval updated to " + intervalMs + "ms"; + } + + /** + * 暂停任务 + */ + @PostMapping("/{taskId}/pause") + public String pauseTask(@PathVariable String taskId) { + boolean success = dynamicTaskService.cancelTask(taskId); + return success ? "Task paused" : "Task not found"; + } + + /** + * 恢复任务 + */ + @PostMapping("/{taskId}/resume") + public String resumeTask(@PathVariable String taskId, + @RequestParam(defaultValue = "30000") long intervalMs) { + // 这里需要根据taskId获取对应的Runnable任务 + // 实际项目中可能需要一个任务注册表 + Runnable task = getTaskById(taskId); + if (task != null) { + dynamicTaskService.scheduleTask(taskId, task, 0, intervalMs); + return "Task resumed with interval " + intervalMs + "ms"; + } + return "Task not found"; + } + + private Runnable getTaskById(String taskId) { + // 根据taskId返回对应的Runnable + // 实际项目中可以维护一个任务注册表 + return () -> {}; + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeRequest.java b/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeRequest.java new file mode 100644 index 0000000..a2fccb7 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeRequest.java @@ -0,0 +1,52 @@ +package com.tongran.agentserver.scheduler.task; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.time.LocalDateTime; +import java.util.List; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class SpecificTimeRequest { + private String taskId; + private String taskName; + private List taskData; + + private String clientId; + + // 多种时间指定方式 + private String cronExpression; // Cron表达式 + private LocalDateTime specificDateTime; // 具体日期时间 + private String time; // 每天固定时间（HH:mm:ss） + private Integer delaySeconds; // 延迟秒数 + +// // 构造方法、getter、setter +// public SpecificTimeRequest() {} +// +// // getter 和 setter 方法 +// public String getTaskId() { return taskId; } +// public void setTaskId(String taskId) { this.taskId = taskId; } +// +// public String getTaskName() { return taskName; } +// public void setTaskName(String taskName) { this.taskName = taskName; } +// +// public String getTaskData() { return taskData; } +// public void setTaskData(String taskData) { this.taskData = taskData; } +// +// public String getCronExpression() { return cronExpression; } +// public void setCronExpression(String cronExpression) { this.cronExpression = cronExpression; } +// +// public LocalDateTime getSpecificDateTime() { return specificDateTime; } +// public void setSpecificDateTime(LocalDateTime specificDateTime) { this.specificDateTime = specificDateTime; } +// +// public String getTime() { return time; } +// public void setTime(String time) { this.time = time; } +// +// public Integer getDelaySeconds() { return delaySeconds; } +// public void setDelaySeconds(Integer delaySeconds) { this.delaySeconds = delaySeconds; } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskConfig.java b/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskConfig.java new file mode 100644 index 0000000..aa83d15 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskConfig.java @@ -0,0 +1,114 @@ +package com.tongran.agentserver.scheduler.task; + +import lombok.Data; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.SchedulingConfigurer; +import org.springframework.scheduling.config.CronTask; +import org.springframework.scheduling.config.ScheduledTaskRegistrar; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.Date; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; + +@Data +@Configuration +public class SpecificTimeTaskConfig implements SchedulingConfigurer { + + private ScheduledTaskRegistrar taskRegistrar; + private final Map> taskFutures = new ConcurrentHashMap<>(); + private final Map taskMap = new ConcurrentHashMap<>(); + + @Override + public void configureTasks(ScheduledTaskRegistrar taskRegistrar) { + this.taskRegistrar = taskRegistrar; + } + + /** + * 添加Cron表达式任务 + */ + public void addCronTask(String taskId, Runnable task, String cronExpression) { + removeTaskIfExists(taskId); + + CronTask cronTask = new CronTask(task, cronExpression); + ScheduledFuture future = taskRegistrar.getScheduler().schedule( + cronTask.getRunnable(), + cronTask.getTrigger() + ); + + taskMap.put(taskId, cronTask); + taskFutures.put(taskId, future); + } + + /** + * 添加指定日期时间任务（只执行一次） + */ + public void addSpecificDateTimeTask(String taskId, Runnable task, LocalDateTime dateTime) { + removeTaskIfExists(taskId); + + Date startTime = Date.from(dateTime.atZone(ZoneId.systemDefault()).toInstant()); + long delay = startTime.getTime() - System.currentTimeMillis(); + + if (delay < 0) { + throw new IllegalArgumentException("指定时间不能是过去时间"); + } + + ScheduledFuture future = taskRegistrar.getScheduler().schedule( + task, + new Date(System.currentTimeMillis() + delay) + ); + + taskMap.put(taskId, "ONCE_" + dateTime.toString()); + taskFutures.put(taskId, future); + } + + /** + * 添加每天固定时间任务 + */ + public void addDailyTimeTask(String taskId, Runnable task, String time) { + String[] timeParts = time.split(":"); + if (timeParts.length < 2) { + throw new IllegalArgumentException("时间格式应为 HH:mm 或 HH:mm:ss"); + } + + int hour = Integer.parseInt(timeParts[0]); + int minute = Integer.parseInt(timeParts[1]); + int second = timeParts.length > 2 ? Integer.parseInt(timeParts[2]) : 0; + + String cronExpression = String.format("%d %d %d * * ?", second, minute, hour); + addCronTask(taskId, task, cronExpression); + } + + /** + * 添加延迟任务 + */ + public void addDelayTask(String taskId, Runnable task, int delaySeconds) { + removeTaskIfExists(taskId); + + ScheduledFuture future = taskRegistrar.getScheduler().schedule( + task, + new Date(System.currentTimeMillis() + delaySeconds * 1000L) + ); + + taskMap.put(taskId, "DELAY_" + delaySeconds); + taskFutures.put(taskId, future); + } + + private void removeTaskIfExists(String taskId) { + if (taskFutures.containsKey(taskId)) { + taskFutures.get(taskId).cancel(true); + taskFutures.remove(taskId); + taskMap.remove(taskId); + } + } + + public void removeTask(String taskId) { + removeTaskIfExists(taskId); + } + + public boolean containsTask(String taskId) { + return taskMap.containsKey(taskId); + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskService.java b/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskService.java new file mode 100644 index 0000000..81b1cf8 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/scheduler/task/SpecificTimeTaskService.java @@ -0,0 +1,120 @@ +package com.tongran.agentserver.scheduler.task; + +import cn.hutool.core.collection.CollectionUtil; +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; +import com.tongran.agentserver.core.enums.MsgEnum; +import com.tongran.agentserver.server.file.AsyncCommandExecutor; +import com.tongran.agentserver.server.netty.NettyTcpClient; +import com.tongran.agentserver.server.netty.model.Message; +import com.tongran.agentserver.utils.AgentUtil; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.time.LocalDateTime; +import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +@Service +public class SpecificTimeTaskService { + + @Resource + private SpecificTimeTaskConfig taskConfig; + + @Resource + private NettyTcpClient nettyTcpClient; + + /** + * 创建指定时间任务 + */ + public boolean createSpecificTimeTask(SpecificTimeRequest request) { + String taskId = request.getTaskId() != null ? + request.getTaskId() : "task-" + System.currentTimeMillis(); + + Runnable task = createTaskRunnable(request); + + try { + if (request.getCronExpression() != null) { + taskConfig.addCronTask(taskId, task, request.getCronExpression()); + } else if (request.getSpecificDateTime() != null) { + taskConfig.addSpecificDateTimeTask(taskId, task, request.getSpecificDateTime()); + } else if (request.getTime() != null) { + taskConfig.addDailyTimeTask(taskId, task, request.getTime()); + } else if (request.getDelaySeconds() != null) { + taskConfig.addDelayTask(taskId, task, request.getDelaySeconds()); + } else { + throw new IllegalArgumentException("必须指定一种时间方式"); + } + return true; + } catch (Exception e) { + throw new RuntimeException("创建任务失败: " + e.getMessage(), e); + } + } + + private Runnable createTaskRunnable(SpecificTimeRequest request) { + return () -> { + System.out.println("执行定时任务: " + request.getTaskName()); + System.out.println("任务数据: " + request.getTaskData()); + System.out.println("执行时间: " + LocalDateTime.now()); + System.out.println("-----------------------------------"); + + // 具体的业务逻辑 + executeBusinessLogic(request); + }; + } + + private void executeBusinessLogic(SpecificTimeRequest request) { + // 实现你的业务逻辑 + try { + System.out.println("处理业务: " + request.getTaskData()); + + List> list = new ArrayList<>(); + for (String command : request.getTaskData()) { + Map map = new HashMap<>(); + List cmd = Arrays.asList(command.split("\\s+")); + CompletableFuture future = + AsyncCommandExecutor.executeCommandAsync( + cmd, + 60, TimeUnit.SECONDS); + future.thenAccept(result -> { + if (result.isSuccess()) { + System.out.println("脚本执行成功"); + System.out.println("[成功resOut] " + result.getOutput()); + map.put("command",command); + map.put("resOut",result.getOutput()); + } else { + System.out.println("脚本执行失败"); + System.out.println("[失败resOut] " + result.getOutput()); + map.put("command",command); + map.put("resOut",result.getOutput()); + } + }).exceptionally(ex -> { + System.err.println("执行失败: " + ex.getMessage()); + map.put("command",command); + map.put("resOut","Policy execute filed"); + return null; + }); + list.add(map); + } + if(CollectionUtil.isNotEmpty(list)){ + JSONObject temp = new JSONObject(); + temp.put("policyId",request.getTaskId()); + temp.put("result", AgentUtil.toJsonString(list)); + JSONObject json = new JSONObject(); + json.put("clientId",request.getClientId()); + json.put("resCode",1); + json.put("resData", temp.toString()); + json.put("error",""); + Message message = Message.builder().clientId(request.getClientId()).dataType(MsgEnum.执行脚本策略应答.getValue()).data(json.toString()).build(); + // 将对象转为 JSON 字符串 + String msg = "agent-tcp:"+ JSON.toJSONString(message)+"@tong-ran"; + nettyTcpClient.sendMessage(msg); + } + + // 调用其他服务等 + } catch (Exception e) { + System.err.println("任务执行异常: " + e.getMessage()); + } + } +} \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/server/collect/cpu/impl/CpuServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/cpu/impl/CpuServiceImpl.java index bf8ce53..7931d3e 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/cpu/impl/CpuServiceImpl.java +++ b/src/main/java/com/tongran/agentserver/server/collect/cpu/impl/CpuServiceImpl.java @@ -19,9 +19,7 @@ public class CpuServiceImpl implements CpuService { @Override public CpuVO query() { - // 获取当前时间戳（毫秒级） - long timestamp = System.currentTimeMillis(); - CpuVO cpuVO = CpuVO.builder().timestamp(timestamp).build(); + CpuVO cpuVO = CpuVO.builder().build(); try { SystemInfo si = new SystemInfo(); HardwareAbstractionLayer hal = si.getHardware(); diff --git a/src/main/java/com/tongran/agentserver/server/collect/disk/DiskService.java b/src/main/java/com/tongran/agentserver/server/collect/disk/DiskService.java index 26658b6..e65877b 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/disk/DiskService.java +++ b/src/main/java/com/tongran/agentserver/server/collect/disk/DiskService.java @@ -6,7 +6,7 @@ import com.tongran.agentserver.server.collect.vo.PointVO; import java.util.List; public interface DiskService { - List queryDiskList(); + List queryDiskList(long timestamp); - List queryPointList(); + List queryPointList(long timestamp); } diff --git a/src/main/java/com/tongran/agentserver/server/collect/disk/impl/DiskServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/disk/impl/DiskServiceImpl.java index 84c84d9..30e8c2a 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/disk/impl/DiskServiceImpl.java +++ b/src/main/java/com/tongran/agentserver/server/collect/disk/impl/DiskServiceImpl.java @@ -16,11 +16,10 @@ import java.util.Objects; @Service public class DiskServiceImpl implements DiskService { @Override - public List queryDiskList() { + public List queryDiskList(long timestamp) { List tempList = new ArrayList<>(); List resultList = new ArrayList<>(); SystemInfo si = new SystemInfo(); - long timestamp = System.currentTimeMillis(); System.out.println("========================================================="); // 获取磁盘IO信息 System.out.println("\n=== 磁盘IO信息 ==="); @@ -93,8 +92,7 @@ public class DiskServiceImpl implements DiskService { } @Override - public List queryPointList() { - long timestamp = System.currentTimeMillis(); + public List queryPointList(long timestamp) { List list = new ArrayList<>(); SystemInfo si = new SystemInfo(); // 获取文件系统信息 diff --git a/src/main/java/com/tongran/agentserver/server/collect/docker/DockerService.java b/src/main/java/com/tongran/agentserver/server/collect/docker/DockerService.java index bd29941..1c996fb 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/docker/DockerService.java +++ b/src/main/java/com/tongran/agentserver/server/collect/docker/DockerService.java @@ -5,5 +5,5 @@ import com.tongran.agentserver.server.collect.vo.DockerVO; import java.util.List; public interface DockerService { - List query(); + List query(long timestamp); } diff --git a/src/main/java/com/tongran/agentserver/server/collect/docker/impl/DockerServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/docker/impl/DockerServiceImpl.java index 6a3e13c..56e2d91 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/docker/impl/DockerServiceImpl.java +++ b/src/main/java/com/tongran/agentserver/server/collect/docker/impl/DockerServiceImpl.java @@ -21,7 +21,7 @@ import java.util.List; @Service public class DockerServiceImpl implements DockerService { @Override - public List query(){ + public List query(long timestamp){ List list = new ArrayList<>(); // 配置Docker客户端 DefaultDockerClientConfig config = DefaultDockerClientConfig.createDefaultConfigBuilder() @@ -47,7 +47,6 @@ public class DockerServiceImpl implements DockerService { // 打印容器信息 System.out.println("运行中的Docker容器:"); System.out.println("容器ID\t\t镜像\t\t状态\t\t名称"); - long timestamp = System.currentTimeMillis(); for (Container container : containers) { DockerVO dockerVO = DockerVO.builder().timestamp(timestamp).build(); String id = container.getId().substring(0, 12); // 只显示短ID diff --git a/src/main/java/com/tongran/agentserver/server/collect/memory/impl/MemoryServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/memory/impl/MemoryServiceImpl.java index 402c2ad..20cf516 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/memory/impl/MemoryServiceImpl.java +++ b/src/main/java/com/tongran/agentserver/server/collect/memory/impl/MemoryServiceImpl.java @@ -14,8 +14,7 @@ import java.util.Map; public class MemoryServiceImpl implements MemoryService { @Override public MemoryVO query() { - long timestamp = System.currentTimeMillis(); - MemoryVO memoryVO = MemoryVO.builder().timestamp(timestamp).build(); + MemoryVO memoryVO = MemoryVO.builder().build(); try { Map memInfo = parseMemInfo(); diff --git a/src/main/java/com/tongran/agentserver/server/collect/net/NetService.java b/src/main/java/com/tongran/agentserver/server/collect/net/NetService.java index 4417392..a808760 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/net/NetService.java +++ b/src/main/java/com/tongran/agentserver/server/collect/net/NetService.java @@ -5,5 +5,5 @@ import com.tongran.agentserver.server.collect.vo.NetVO; import java.util.List; public interface NetService { - List query(); + List query(long timestamp); } 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 1a52dd5..05fe5b8 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 @@ -17,8 +17,7 @@ import java.util.concurrent.TimeUnit; @Service public class NetServiceImpl implements NetService { @Override - public List query() { - long timestamp = System.currentTimeMillis(); + public List query(long timestamp) { List list = new ArrayList<>(); try { SystemInfo si = new SystemInfo(); diff --git a/src/main/java/com/tongran/agentserver/server/collect/switchboard/SwitchBoardService.java b/src/main/java/com/tongran/agentserver/server/collect/switchboard/SwitchBoardService.java index 5b2aaa0..c8ab125 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/switchboard/SwitchBoardService.java +++ b/src/main/java/com/tongran/agentserver/server/collect/switchboard/SwitchBoardService.java @@ -5,5 +5,5 @@ import com.tongran.agentserver.server.collect.vo.SwitchBoardVO; import java.util.List; public interface SwitchBoardService { - List query(); + List query(long timestamp); } diff --git a/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java index 408ebfc..425a72d 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java +++ b/src/main/java/com/tongran/agentserver/server/collect/switchboard/impl/SwitchBoardServiceImpl.java @@ -20,7 +20,7 @@ import java.util.List; @Service public class SwitchBoardServiceImpl implements SwitchBoardService { @Override - public List query() { + public List query(long timestamp) { List list = new ArrayList<>(); System.out.println("==================== 交换机流量信息 ===================="); try { @@ -41,7 +41,7 @@ public class SwitchBoardServiceImpl implements SwitchBoardService { getSystemInfo(snmp, target); // 4. 获取接口信息 - list = getInterfaceInfo(snmp, target); + list = getInterfaceInfo(snmp, target, timestamp); snmp.close(); } catch (Exception e) { @@ -77,7 +77,7 @@ public class SwitchBoardServiceImpl implements SwitchBoardService { } } - private static List getInterfaceInfo(Snmp snmp, Target target) throws IOException { + private static List getInterfaceInfo(Snmp snmp, Target target, long timestamp) throws IOException { // 获取接口数量 String ifNumberOID = "1.3.6.1.2.1.2.1.0"; PDU pdu = new PDU(); @@ -106,7 +106,6 @@ public class SwitchBoardServiceImpl implements SwitchBoardService { System.out.printf("%-5s %-15s %-10s %-15s %-8s %-12s %-12s%n", "Index", "Name", "Type", "Speed", "Status", "InBytes", "OutBytes"); List list = new ArrayList<>(); - long timestamp = System.currentTimeMillis(); for (int i = 1; i <= ifNumber; i++) { pdu = new PDU(); for (String baseOID : ifOIDs) { diff --git a/src/main/java/com/tongran/agentserver/server/collect/system/impl/SystemServiceImpl.java b/src/main/java/com/tongran/agentserver/server/collect/system/impl/SystemServiceImpl.java index 4cfe40d..3372fc8 100644 --- a/src/main/java/com/tongran/agentserver/server/collect/system/impl/SystemServiceImpl.java +++ b/src/main/java/com/tongran/agentserver/server/collect/system/impl/SystemServiceImpl.java @@ -18,8 +18,7 @@ import java.time.format.DateTimeFormatter; public class SystemServiceImpl implements SystemService { @Override public SystemVO query() { - long timestamp = System.currentTimeMillis(); - SystemVO systemVO = SystemVO.builder().timestamp(timestamp).build(); + SystemVO systemVO = SystemVO.builder().build(); try { SystemInfo si = new SystemInfo(); HardwareAbstractionLayer hal = si.getHardware(); 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 eccdc17..cd0f589 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 @@ -5,7 +5,10 @@ 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.scheduler.service.BusinessTasks; +import com.tongran.agentserver.scheduler.service.DynamicTaskService; +import com.tongran.agentserver.scheduler.task.SpecificTimeRequest; +import com.tongran.agentserver.scheduler.task.SpecificTimeTaskService; import com.tongran.agentserver.server.file.AdvancedAsyncDownloader; import com.tongran.agentserver.server.file.AsyncCommandExecutor; import com.tongran.agentserver.server.file.LinuxCommandExecutor; @@ -16,11 +19,13 @@ import com.tongran.agentserver.server.netty.model.Message; import com.tongran.agentserver.server.netty.model.ScriptPolicy; import com.tongran.agentserver.server.netty.model.UpMsgResponse; import com.tongran.agentserver.utils.AgentUtil; +import com.tongran.agentserver.utils.Java8FileDownloader; import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; import javax.annotation.Resource; -import java.io.FileOutputStream; import java.io.IOException; import java.util.*; import java.util.concurrent.CompletableFuture; @@ -30,36 +35,23 @@ import java.util.concurrent.atomic.AtomicBoolean; @Component public class AgentEndpoint { - @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 NettyTcpClient nettyTcpClient; -// @Resource -// private SpecificTimeTaskService taskService; + @Resource + private SpecificTimeTaskService taskService; + private final DynamicTaskService dynamicTaskService; + + private final BusinessTasks businessTasks; + + // 在注入点使用@Lazy + @Autowired + public AgentEndpoint(DynamicTaskService dynamicTaskService, + @Lazy BusinessTasks businessTasks) { + this.dynamicTaskService = dynamicTaskService; + this.businessTasks = businessTasks; + } @AgentDispatcher(msgId = MsgEnum.更新CPU采集间隔) public class TimeCpuHandler implements AgentHandler { @@ -67,7 +59,8 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - cpuScheduler.updateSchedule(25000, intervalMillis); + dynamicTaskService.scheduleTask("cpu", + businessTasks::cpuTask, 25000, intervalMillis); // 25秒后开始，每300秒执行 JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -81,7 +74,8 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - diskScheduler.updateSchedule(35000, intervalMillis); + dynamicTaskService.scheduleTask("disk", + businessTasks::cpuTask, 35000, intervalMillis); // 35秒后开始，每300秒执行 JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -89,13 +83,29 @@ public class AgentEndpoint { } } + @AgentDispatcher(msgId = MsgEnum.更新系统采集间隔) + public class TimeSysHandler implements AgentHandler { + @Override + public UpMsgResponse upHandle(String data, String clientId) { + JSONObject jsonObject = JSONObject.parseObject(data); + long intervalMillis = jsonObject.getLong("intervalMillis"); + dynamicTaskService.scheduleTask("system", + businessTasks::cpuTask, 40000, intervalMillis); // 40秒后开始，每300秒执行 + JSONObject json = new JSONObject(); + json.put("clientId",clientId); + json.put("resCode",1); + return UpMsgResponse.builder().dataType(MsgEnum.更新系统采集间隔应答.getValue()).content(json.toString()).build(); + } + } + @AgentDispatcher(msgId = MsgEnum.更新容器采集间隔) public class TimeDockerHandler implements AgentHandler { @Override public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - dockerScheduler.updateSchedule(45000, intervalMillis); + dynamicTaskService.scheduleTask("docker", + businessTasks::cpuTask, 45000, intervalMillis); // 45秒后开始，每300秒执行 JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -109,7 +119,8 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - memoryScheduler.updateSchedule(55000, intervalMillis); + dynamicTaskService.scheduleTask("memory", + businessTasks::cpuTask, 55000, intervalMillis); // 55秒后开始，每300秒执行 JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -123,7 +134,8 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - netScheduler.updateSchedule(65000, intervalMillis); + dynamicTaskService.scheduleTask("net", + businessTasks::cpuTask, 65000, intervalMillis); // 65秒后开始，每300秒执行 JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -137,7 +149,8 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - pointScheduler.updateSchedule(75000, intervalMillis); + dynamicTaskService.scheduleTask("point", + businessTasks::cpuTask, 75000, intervalMillis); // 75秒后开始，每300秒执行 JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -151,7 +164,8 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - switchBoardScheduler.updateSchedule(85000, intervalMillis); + dynamicTaskService.scheduleTask("switchBoard", + businessTasks::cpuTask, 85000, intervalMillis); // 85秒后开始，每300秒执行 JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -159,20 +173,6 @@ public class AgentEndpoint { } } - @AgentDispatcher(msgId = MsgEnum.更新系统采集间隔) - public class TimeSysHandler implements AgentHandler { - @Override - public UpMsgResponse upHandle(String data, String clientId) { - JSONObject jsonObject = JSONObject.parseObject(data); - long intervalMillis = jsonObject.getLong("intervalMillis"); - sysScheduler.updateSchedule(40000, intervalMillis); - JSONObject json = new JSONObject(); - json.put("clientId",clientId); - json.put("resCode",1); - return UpMsgResponse.builder().dataType(MsgEnum.更新系统采集间隔应答.getValue()).content(json.toString()).build(); - } - } - @AgentDispatcher(msgId = MsgEnum.文件下载) public class DownFileHandler implements AgentHandler { @Override @@ -313,28 +313,40 @@ public class AgentEndpoint { if(f.getFileType() == 0){ String filePath = policy.getFilePath() + "/" + f.getFileName(); // 移除Base64 URL前缀（如果存在） - if (f.getFileData().contains(",")) { - f.setFileData(f.getFileData().split(",")[1]); - } - byte[] dataByte = Base64.getDecoder().decode(f.getFileData()); - try (FileOutputStream stream = new FileOutputStream(filePath)) { - stream.write(dataByte); - isFalse.set(false); +// if (f.getFileData().contains(",")) { +// f.setFileData(f.getFileData().split(",")[1]); +// } + try { + AgentUtil.base64ToFile(f.getFileData(), filePath); System.out.println("文件保存成功: " + filePath); + isFalse.set(false); } catch (IOException e) { - e.printStackTrace(); + System.err.println("文件保存失败: " + e.getMessage()); } }else{ - AdvancedAsyncDownloader.downloadWithProgress(f.getFileUrl(), finalSaveDir, null, progress -> { -// System.out.printf("下载进度: %.1f%%\n", progress); - }).thenAccept(filePath -> { - isFalse.set(false); - System.out.println("下载完成: " + filePath); - }).exceptionally(ex -> { - isFalse.set(true); - System.err.println("下载错误: " + ex.getMessage()); - return null; - }); + Java8FileDownloader.DownloadResult result = Java8FileDownloader.downloadFile(f.getFileUrl(), finalSaveDir, f.getFileName()); + // 验证下载是否完成 + if (result.isSuccess()) { + boolean isComplete = Java8FileDownloader.verifyFileCompletion( + result.getFilePath(), result.getExpectedSize()); + if(isComplete){ + isFalse.set(false); + System.out.println("下载完成: " + finalSaveDir+"/"+f.getFileName()); + }else{ + isFalse.set(true); + System.err.println("下载失败"); + } + } +// AdvancedAsyncDownloader.downloadWithProgress(f.getFileUrl(), finalSaveDir, null, progress -> { +//// System.out.printf("下载进度: %.1f%%\n", progress); +// }).thenAccept(filePath -> { +// isFalse.set(false); +// System.out.println("下载完成: " + filePath); +// }).exceptionally(ex -> { +// isFalse.set(true); +// System.err.println("下载错误: " + ex.getMessage()); +// return null; +// }); } } } @@ -345,34 +357,35 @@ public class AgentEndpoint { if(CollectionUtil.isNotEmpty(policy.getCommands())){ //执行方式：0、立即执行；1、定时执行； if(policy.getMethod() == 1){ -// SpecificTimeRequest request = SpecificTimeRequest.builder() -// .taskId("") -// .taskName("") -// .taskData(policy.getCommands()) -// .specificDateTime(policy.getPolicyTime()) -// .build(); -// try { -//// taskService.createSpecificTimeTask(request); -// JSONObject json = new JSONObject(); -// json.put("clientId",clientId); -// json.put("resCode",1); -// json.put("result", ""); -// json.put("error",""); -// 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); -// } catch (Exception e) { -// JSONObject json = new JSONObject(); -// json.put("clientId",clientId); -// json.put("resCode",0); -// json.put("result", ""); -// json.put("error","policy save failed"); -// 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); -// } + SpecificTimeRequest request = SpecificTimeRequest.builder() + .taskId(policy.getPolicyId()) + .taskName(policy.getPolicyName()) + .taskData(policy.getCommands()) + .specificDateTime(policy.getPolicyTime()) + .clientId(clientId) + .build(); + try { + taskService.createSpecificTimeTask(request); + JSONObject json = new JSONObject(); + json.put("clientId",clientId); + json.put("resCode",1); + json.put("result", ""); + json.put("error",""); + 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); + } catch (Exception e) { + JSONObject json = new JSONObject(); + json.put("clientId",clientId); + json.put("resCode",0); + json.put("result", ""); + json.put("error","policy save failed"); + 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{ List> list = new ArrayList<>(); for (String command : policy.getCommands()) { @@ -403,10 +416,13 @@ public class AgentEndpoint { list.add(map); } if(CollectionUtil.isNotEmpty(list)){ + JSONObject temp = new JSONObject(); + temp.put("policyId",policy.getPolicyId()); + temp.put("result", AgentUtil.toJsonString(list)); JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); - json.put("result", AgentUtil.toJsonString(list)); + json.put("result", temp.toString()); json.put("error",""); Message message = Message.builder().clientId(clientId).dataType(MsgEnum.执行脚本策略应答.getValue()).data(json.toString()).build(); // 将对象转为 JSON 字符串 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 d8d0b99..65d525a 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 @@ -37,7 +37,7 @@ public class DecoderHandler extends ChannelInboundHandlerAdapter { AgentDispatcherManager agentDispatcherManager; // 用来临时保留没有处理过的请求报文 - private final Cache lruCache = CacheUtil.newLRUCache(3000); + private final Cache lruCache = CacheUtil.newLRUCache(5000); /** * 消息解码器 @@ -51,6 +51,7 @@ public class DecoderHandler extends ChannelInboundHandlerAdapter { AssertLog.info("<<[up1]:[up-content]==>{}", messages); String content = ""; + StringBuilder sb = new StringBuilder(); boolean isClear = false; String tempMsg = lruCache.get(sessionManager.client(ctx)); tempMsg = tempMsg == null ? "" : tempMsg; @@ -67,12 +68,14 @@ public class DecoderHandler extends ChannelInboundHandlerAdapter { if(endsWith){ lruCache.remove(sessionManager.client(ctx)); tmpMsgSize = 0; - content = arr_msg[0]+"@tong-ran"; + sb.append(arr_msg[0]+"@tong-ran"); +// 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); + sb.append(arr_msg[0]); +// content = arr_msg[0]; + lruCache.put(sessionManager.client(ctx), sb.toString(), DateUnit.SECOND.getMillis() * 5000); } } //判定是否粘包 @@ -80,18 +83,29 @@ public class DecoderHandler extends ChannelInboundHandlerAdapter { 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); + sb.append(arr_msg[0]+"@tong-ran"); +// content = arr_msg[0]+"@tong-ran"; + lruCache.put(sessionManager.client(ctx), arr_msg[1], DateUnit.SECOND.getMillis() * 5000); }else{ lruCache.remove(sessionManager.client(ctx)); tmpMsgSize = 0; - content = String.join("@tong-ran", arr_msg); + sb.append(String.join("@tong-ran", arr_msg)); +// content = String.join("@tong-ran", arr_msg); } } } + content = sb.toString(); if (tmpMsgSize > 0) { content = tempMsg + messages; - isClear = true; + 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); + } + } AssertLog.info("<<[up2]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content); endsWith = content.endsWith("@tong-ran"); diff --git a/src/main/java/com/tongran/agentserver/server/netty/model/ScriptPolicy.java b/src/main/java/com/tongran/agentserver/server/netty/model/ScriptPolicy.java index 659d56c..05cb92b 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/model/ScriptPolicy.java +++ b/src/main/java/com/tongran/agentserver/server/netty/model/ScriptPolicy.java @@ -17,6 +17,10 @@ public class ScriptPolicy implements Serializable { private static final long serialVersionUID = -1267013167162440612L; + private String policyId; + + private String policyName; + /** * 文件 */ diff --git a/src/main/java/com/tongran/agentserver/utils/AgentUtil.java b/src/main/java/com/tongran/agentserver/utils/AgentUtil.java index d2eaf88..181e261 100644 --- a/src/main/java/com/tongran/agentserver/utils/AgentUtil.java +++ b/src/main/java/com/tongran/agentserver/utils/AgentUtil.java @@ -6,6 +6,9 @@ import oshi.SystemInfo; import oshi.hardware.Baseboard; import oshi.hardware.HardwareAbstractionLayer; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.Base64; import java.util.List; import java.util.Map; @@ -30,4 +33,14 @@ public class AgentUtil { } } + public static void base64ToFile(String base64String, String filePath) throws IOException { + // 解码Base64字符串 + byte[] decodedBytes = Base64.getDecoder().decode(base64String); + + // 写入文件 + try (FileOutputStream fos = new FileOutputStream(filePath)) { + fos.write(decodedBytes); + } + } + } diff --git a/src/main/java/com/tongran/agentserver/utils/Java8FileDownloader.java b/src/main/java/com/tongran/agentserver/utils/Java8FileDownloader.java new file mode 100644 index 0000000..8ed4e94 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/utils/Java8FileDownloader.java @@ -0,0 +1,182 @@ +package com.tongran.agentserver.utils; + +import java.io.BufferedInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.net.URL; +import java.net.URLConnection; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +public class Java8FileDownloader { + + /** + * 下载结果类 + */ + public static class DownloadResult { + private boolean success; + private String message; + private long expectedSize; + private long downloadedSize; + private String filePath; + + public DownloadResult() { + this.success = false; + this.message = ""; + } + + // Getter和Setter方法 + public boolean isSuccess() { return success; } + public void setSuccess(boolean success) { this.success = success; } + + public String getMessage() { return message; } + public void setMessage(String message) { this.message = message; } + + public long getExpectedSize() { return expectedSize; } + public void setExpectedSize(long expectedSize) { this.expectedSize = expectedSize; } + + public long getDownloadedSize() { return downloadedSize; } + public void setDownloadedSize(long downloadedSize) { this.downloadedSize = downloadedSize; } + + public String getFilePath() { return filePath; } + public void setFilePath(String filePath) { this.filePath = filePath; } + + @Override + public String toString() { + if (success) { + return String.format("下载成功: %s (%d/%d 字节)", + filePath, downloadedSize, expectedSize); + } else { + return String.format("下载失败: %s", message); + } + } + } + + /** + * 同步下载文件 + * @param fileURL 文件URL地址 + * @param saveDir 保存目录 + * @param fileName 文件名（如果为null则从URL获取） + * @return 下载结果 + */ + public static DownloadResult downloadFile(String fileURL, String saveDir, String fileName) { + DownloadResult result = new DownloadResult(); + + try { + // 创建保存目录 + Path dirPath = Paths.get(saveDir); + if (!Files.exists(dirPath)) { + Files.createDirectories(dirPath); + } + + // 获取文件名 + String actualFileName = getFileName(fileURL, fileName); + String savePath = Paths.get(saveDir, actualFileName).toString(); + result.setFilePath(savePath); + + URL url = new URL(fileURL); + URLConnection connection = url.openConnection(); + connection.setConnectTimeout(10000); // 10秒连接超时 + connection.setReadTimeout(30000); // 30秒读取超时 + + // 获取文件信息 + long expectedSize = connection.getContentLengthLong(); + String contentType = connection.getContentType(); + result.setExpectedSize(expectedSize); + + System.out.println("开始下载: " + actualFileName); + System.out.println("文件大小: " + formatFileSize(expectedSize)); + System.out.println("文件类型: " + contentType); + + // 下载文件 + try (BufferedInputStream in = new BufferedInputStream(connection.getInputStream()); + FileOutputStream out = new FileOutputStream(savePath)) { + + byte[] buffer = new byte[8192]; // 8KB缓冲区 + int bytesRead; + long totalRead = 0; + + while ((bytesRead = in.read(buffer)) != -1) { + out.write(buffer, 0, bytesRead); + totalRead += bytesRead; + + // 显示进度（可选） + if (expectedSize > 0) { + int progress = (int) ((totalRead * 100) / expectedSize); + if (progress % 10 == 0) { // 每10%显示一次进度 + System.out.printf("下载进度: %d%%\n", progress); + } + } + } + + result.setDownloadedSize(totalRead); + result.setSuccess(true); + + // 验证文件完整性 + if (expectedSize > 0 && totalRead != expectedSize) { + result.setSuccess(false); + result.setMessage(String.format("文件不完整: 期望 %d 字节, 实际 %d 字节", + expectedSize, totalRead)); + } else { + result.setMessage("下载完成，文件完整性验证通过"); + } + + } + + } catch (IOException e) { + result.setSuccess(false); + result.setMessage("下载失败: " + e.getClass().getSimpleName() + " - " + e.getMessage()); + } + + return result; + } + + /** + * 从URL中提取文件名 + */ + private static String getFileName(String fileURL, String customFileName) { + if (customFileName != null && !customFileName.trim().isEmpty()) { + return customFileName; + } + + try { + URL url = new URL(fileURL); + String path = url.getPath(); + if (path.contains("/")) { + return path.substring(path.lastIndexOf("/") + 1); + } + return "downloaded_file"; + } catch (Exception e) { + return "downloaded_file"; + } + } + + /** + * 格式化文件大小 + */ + private static String formatFileSize(long size) { + if (size <= 0) return "0 B"; + + final String[] units = new String[]{"B", "KB", "MB", "GB", "TB"}; + int digitGroups = (int) (Math.log10(size) / Math.log10(1024)); + + return String.format("%.1f %s", size / Math.pow(1024, digitGroups), units[digitGroups]); + } + + /** + * 验证文件是否完整下载 + */ + public static boolean verifyFileCompletion(String filePath, long expectedSize) { + try { + Path path = Paths.get(filePath); + if (Files.exists(path)) { + long actualSize = Files.size(path); + return actualSize == expectedSize; + } + return false; + } catch (IOException e) { + return false; + } + } +} \ No newline at end of file