From 1351585ed59bfaff68dd1dc3b3fd63152537ec38 Mon Sep 17 00:00:00 2001 From: qiminbao Date: Tue, 26 Aug 2025 13:00:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E6=94=B6=E6=9C=8D=E5=8A=A1=E7=AB=AF?= =?UTF-8?q?=E6=8C=87=E4=BB=A4-=E6=9B=B4=E6=96=B0=E9=87=87=E9=9B=86?= =?UTF-8?q?=E6=97=B6=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pom.xml | 19 +++ .../agentserver/AgentServerApplication.java | 4 + .../agentserver/controller/TcpController.java | 47 +++---- .../agentserver/scheduler/CpuScheduler.java | 106 ++++++++++------ .../agentserver/scheduler/DiskScheduler.java | 105 ++++++++++------ .../scheduler/DockerScheduler.java | 106 +++++++++------- .../agentserver/scheduler/HeartScheduler.java | 116 ++++++++++------- .../scheduler/MemoryScheduler.java | 106 ++++++++++------ .../agentserver/scheduler/NetScheduler.java | 106 ++++++++++------ .../agentserver/scheduler/PointScheduler.java | 106 ++++++++++------ .../scheduler/SwitchBoardScheduler.java | 105 ++++++++++------ .../agentserver/scheduler/SysScheduler.java | 106 ++++++++++------ .../collect/docker/DockerClientManager.java | 62 ++++++++++ .../docker/impl/DockerServiceImpl.java | 10 +- .../server/netty/NettyTcpClient.java | 9 +- .../netty/annotation/AgentDispatcher.java | 2 +- .../netty/basics/AgentDispatcherManager.java | 2 + .../server/netty/enpoint/AgentEndpoint.java | 117 ++++++++---------- .../server/netty/handler/DecoderHandler.java | 46 ++++--- 19 files changed, 790 insertions(+), 490 deletions(-) create mode 100644 src/main/java/com/tongran/agentserver/server/collect/docker/DockerClientManager.java diff --git a/pom.xml b/pom.xml index 28c1da3..16a737a 100644 --- a/pom.xml +++ b/pom.xml @@ -91,6 +91,25 @@ 3.3.0 + + + + + + + + com.github.docker-java + docker-java-api + 3.3.0 + + + + + com.github.docker-java + docker-java-transport-httpclient5 + 3.3.0 + + org.projectlombok lombok diff --git a/src/main/java/com/tongran/agentserver/AgentServerApplication.java b/src/main/java/com/tongran/agentserver/AgentServerApplication.java index 4e1097f..33d9db6 100644 --- a/src/main/java/com/tongran/agentserver/AgentServerApplication.java +++ b/src/main/java/com/tongran/agentserver/AgentServerApplication.java @@ -2,12 +2,16 @@ package com.tongran.agentserver; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.ComponentScan; @SpringBootApplication +@ComponentScan(basePackages = {"com.tongran.agentserver", "cn.hutool.extra.spring"}) public class AgentServerApplication { public static void main(String[] args) { SpringApplication.run(AgentServerApplication.class, args); } + + } diff --git a/src/main/java/com/tongran/agentserver/controller/TcpController.java b/src/main/java/com/tongran/agentserver/controller/TcpController.java index 5fd858b..ffe144a 100644 --- a/src/main/java/com/tongran/agentserver/controller/TcpController.java +++ b/src/main/java/com/tongran/agentserver/controller/TcpController.java @@ -1,35 +1,28 @@ package com.tongran.agentserver.controller; -import com.tongran.agentserver.scheduler.HeartScheduler; -import com.tongran.agentserver.server.netty.NettyTcpClient; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.web.bind.annotation.GetMapping; -import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; -import javax.annotation.Resource; - @RestController public class TcpController { - @Autowired - private NettyTcpClient nettyTcpClient; - - @Resource - private HeartScheduler heartScheduler; - - @GetMapping("/send") - public String sendMessage(@RequestParam String message) { - if (nettyTcpClient.isConnected()) { - nettyTcpClient.sendMessage(message); - return "消息已发送: " + message; - } - return "TCP连接未建立，消息发送失败"; - } - - @GetMapping("/task") - public String task(@RequestParam long intervalMillis) { - heartScheduler.updateInterval(intervalMillis); - return "TCP连接未建立，消息发送失败"; - } +// @Autowired +// private NettyTcpClient nettyTcpClient; +// +// @Resource +// private HeartScheduler heartScheduler; +// +// @GetMapping("/send") +// public String sendMessage(@RequestParam String message) { +// if (nettyTcpClient.isConnected()) { +// nettyTcpClient.sendMessage(message); +// return "消息已发送: " + message; +// } +// return "TCP连接未建立，消息发送失败"; +// } +// +// @GetMapping("/task") +// public String task(@RequestParam long intervalMillis) { +// heartScheduler.updateInterval(intervalMillis); +// return "TCP连接未建立，消息发送失败"; +// } } \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java b/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java index 1866fc9..2bb9ea6 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java @@ -7,72 +7,96 @@ 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.PostConstruct; +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 { - - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待12秒 - try { - Thread.sleep(12000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认3分钟间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public CpuScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } - Runnable task = () -> { - AssertLog.info("CPU信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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"; - AssertLog.info("发送CPU信息包={}",json); - nettyTcpClient.sendMessage(json); + @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); } - }; - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("CPU信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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 index cf25ee5..4c5d7dd 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/DiskScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/DiskScheduler.java @@ -8,12 +8,17 @@ 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.PostConstruct; +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; @@ -21,61 +26,79 @@ import java.util.concurrent.ScheduledFuture; @Component @EnableScheduling public class DiskScheduler { - - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待27秒 - try { - Thread.sleep(27000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认300秒间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public DiskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } + @EventListener(ApplicationReadyEvent.class) + public void startDynamicTask() { + AssertLog.info("启动磁盘信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); + scheduleTask(initialDelay, interval); + } - Runnable task = () -> { - AssertLog.info("磁盘信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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"; - AssertLog.info("发送磁盘信息包={}",json); - nettyTcpClient.sendMessage(json); + public void scheduleTask(long delayMillis, long intervalMillis) { + synchronized (lock) { + // 取消现有任务 + if (currentTask != null) { + currentTask.cancel(false); } - }; - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("磁盘信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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 index f686982..7a10722 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java @@ -8,12 +8,17 @@ 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.PostConstruct; +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; @@ -21,62 +26,79 @@ import java.util.concurrent.ScheduledFuture; @Component @EnableScheduling public class DockerScheduler { - - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待42秒 - try { - Thread.sleep(42000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认180秒间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public DockerScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } + @EventListener(ApplicationReadyEvent.class) + public void startDynamicTask() { + AssertLog.info("启动容器信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); + scheduleTask(initialDelay, interval); + } - Runnable task = () -> { - AssertLog.info("容器信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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"; - AssertLog.info("发送容器信息包={}",json); - nettyTcpClient.sendMessage(json); + public void scheduleTask(long delayMillis, long intervalMillis) { + synchronized (lock) { + // 取消现有任务 + if (currentTask != null) { + currentTask.cancel(false); } - }; - - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("容器信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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 index a84ed50..f7972fc 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/HeartScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/HeartScheduler.java @@ -6,74 +6,106 @@ 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.PostConstruct; +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 ScheduledFuture scheduledTask; + 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 TaskScheduler taskScheduler; + private NettyTcpClient nettyTcpClient; - @Resource - private NettyTcpClient nettyTcpClient; - - - @PostConstruct - public void init() { - // 等待5秒 - try { - Thread.sleep(5000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(30000); // 默认30秒间隔启动 + public HeartScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } + @EventListener(ApplicationReadyEvent.class) + public void startDynamicTask() { + AssertLog.info("启动心跳定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); + scheduleTask(initialDelay, interval); + } - Runnable task = () -> { - // 业务逻辑 - if (!nettyTcpClient.isConnected()) { + 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 { - nettyTcpClient.connect(); + Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } + } catch (InterruptedException e) { + e.printStackTrace(); } - // 发送心跳包 - JSONObject object = new JSONObject(); - object.put("strength","31"); - 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); - }; - - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + } + // 发送心跳包 + JSONObject object = new JSONObject(); + object.put("strength","31"); + 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 updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void updateSchedule(long newDelay, long newInterval) { + AssertLog.info("更新调度配置 - 延迟: {}ms, 间隔: {}ms", newDelay, newInterval); + scheduleTask(newDelay, newInterval); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + @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 index 40e15c4..b931df1 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/MemoryScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/MemoryScheduler.java @@ -7,73 +7,97 @@ 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.PostConstruct; +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 { - - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待47秒 - try { - Thread.sleep(47000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认300秒间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public MemoryScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } + @EventListener(ApplicationReadyEvent.class) + public void startDynamicTask() { + AssertLog.info("启动内存信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); + scheduleTask(initialDelay, interval); + } - Runnable task = () -> { - AssertLog.info("内存信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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 scheduleTask(long delayMillis, long intervalMillis) { + synchronized (lock) { + // 取消现有任务 + if (currentTask != null) { + currentTask.cancel(false); } - }; - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("内存信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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 index 5b2f81b..b3c9c96 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java @@ -8,12 +8,17 @@ 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.PostConstruct; +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; @@ -21,61 +26,80 @@ import java.util.concurrent.ScheduledFuture; @Component @EnableScheduling public class NetScheduler { - - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待72秒 - try { - Thread.sleep(72000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认180秒间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public NetScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } + @EventListener(ApplicationReadyEvent.class) + public void startDynamicTask() { + AssertLog.info("启动网络信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); + scheduleTask(initialDelay, interval); + } - Runnable task = () -> { - AssertLog.info("网卡信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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 scheduleTask(long delayMillis, long intervalMillis) { + synchronized (lock) { + // 取消现有任务 + if (currentTask != null) { + currentTask.cancel(false); } - }; - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("网络信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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 index 66a284a..fb43376 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/PointScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/PointScheduler.java @@ -8,12 +8,17 @@ 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.PostConstruct; +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; @@ -21,61 +26,80 @@ import java.util.concurrent.ScheduledFuture; @Component @EnableScheduling public class PointScheduler { - - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待97秒 - try { - Thread.sleep(97000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认300秒间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public PointScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } + @EventListener(ApplicationReadyEvent.class) + public void startDynamicTask() { + AssertLog.info("启动挂载信息采集定时任务 - 延迟: {}ms, 间隔: {}ms", initialDelay, interval); + scheduleTask(initialDelay, interval); + } - Runnable task = () -> { - AssertLog.info("挂载点信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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 scheduleTask(long delayMillis, long intervalMillis) { + synchronized (lock) { + // 取消现有任务 + if (currentTask != null) { + currentTask.cancel(false); } - }; - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("挂载信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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 index d02418f..05d68d0 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java @@ -8,12 +8,17 @@ 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.PostConstruct; +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; @@ -21,59 +26,79 @@ import java.util.concurrent.ScheduledFuture; @Component @EnableScheduling public class SwitchBoardScheduler { - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待33秒 - try { - Thread.sleep(33000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认3分钟间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public SwitchBoardScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } - Runnable task = () -> { - AssertLog.info("交换机信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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"; - AssertLog.info("发送交换机信息包={}",json); - nettyTcpClient.sendMessage(json); + @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); } - }; - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("交换机信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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 index c609525..3a62bfb 100644 --- a/src/main/java/com/tongran/agentserver/scheduler/SysScheduler.java +++ b/src/main/java/com/tongran/agentserver/scheduler/SysScheduler.java @@ -7,72 +7,96 @@ 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.PostConstruct; +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 { - - @Resource - private TaskScheduler taskScheduler; - private ScheduledFuture scheduledTask; + private final TaskScheduler taskScheduler; - @Resource - private NettyTcpClient nettyTcpClient; + 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; - @PostConstruct - public void init() { - // 等待50秒 - try { - Thread.sleep(50000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - startTask(300000); // 默认300秒间隔启动 + @Resource + private NettyTcpClient nettyTcpClient; + + public SysScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; } - public void startTask(long intervalMillis) { - if (scheduledTask != null) { - scheduledTask.cancel(false); - } - Runnable task = () -> { - AssertLog.info("系统信息包准备={}", LocalDateTime.now()); - // 业务逻辑 - 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"; - AssertLog.info("发送系统信息包={}",json); - nettyTcpClient.sendMessage(json); + @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); } - }; - scheduledTask = taskScheduler.scheduleAtFixedRate(task, intervalMillis); + // 安排新任务 + currentTask = taskScheduler.scheduleWithFixedDelay( + this::dynamicPeriodicTask, + Instant.now().plusMillis(delayMillis), + Duration.ofMillis(intervalMillis) + ); + } } - public void updateInterval(long newIntervalMillis) { - startTask(newIntervalMillis); + public void dynamicPeriodicTask() { + AssertLog.info("系统信息采集定时任务执行 - 时间: {}", LocalDateTime.now()); + performDynamicTask(); } - public void stopTask() { - if (scheduledTask != null) { - scheduledTask.cancel(false); - scheduledTask = null; + 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/server/collect/docker/DockerClientManager.java b/src/main/java/com/tongran/agentserver/server/collect/docker/DockerClientManager.java new file mode 100644 index 0000000..97e6047 --- /dev/null +++ b/src/main/java/com/tongran/agentserver/server/collect/docker/DockerClientManager.java @@ -0,0 +1,62 @@ +package com.tongran.agentserver.server.collect.docker; + +import com.github.dockerjava.api.DockerClient; +import com.github.dockerjava.core.DefaultDockerClientConfig; +import com.github.dockerjava.core.DockerClientImpl; +import com.github.dockerjava.httpclient5.ApacheDockerHttpClient; +import com.github.dockerjava.transport.DockerHttpClient; + +import java.time.Duration; + +public class DockerClientManager { + + private static DockerClient dockerClient; + + public static synchronized DockerClient getDockerClient() { + if (dockerClient == null) { + dockerClient = createDockerClient(); + } + return dockerClient; + } + + private static DockerClient createDockerClient() { + try { + DefaultDockerClientConfig config = DefaultDockerClientConfig.createDefaultConfigBuilder() + .withDockerHost("unix:///var/run/docker.sock") + .withDockerTlsVerify(false) // 根据你的配置调整 + .build(); + + DockerHttpClient httpClient = new ApacheDockerHttpClient.Builder() + .dockerHost(config.getDockerHost()) + .sslConfig(config.getSSLConfig()) + .maxConnections(100) + .connectionTimeout(Duration.ofSeconds(30)) + .responseTimeout(Duration.ofSeconds(45)) + .build(); + + return DockerClientImpl.getInstance(config, httpClient); + + } catch (Exception e) { + throw new RuntimeException("Failed to create Docker client", e); + } + } + + private static String getDockerHost() { + // 可以从环境变量、配置文件等获取 + String host = System.getenv("DOCKER_HOST"); + if (host == null || host.trim().isEmpty()) { + host = "unix:///var/run/docker.sock"; // 默认值 + } + return host; + } + + public static void close() { + if (dockerClient != null) { + try { + dockerClient.close(); + } catch (Exception e) { + // 记录日志 + } + } + } +} \ No newline at end of file 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 f69f9a6..79bb268 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 @@ -2,8 +2,7 @@ package com.tongran.agentserver.server.collect.docker.impl; import com.github.dockerjava.api.DockerClient; import com.github.dockerjava.api.model.Container; -import com.github.dockerjava.core.DefaultDockerClientConfig; -import com.github.dockerjava.core.DockerClientBuilder; +import com.tongran.agentserver.server.collect.docker.DockerClientManager; import com.tongran.agentserver.server.collect.docker.DockerService; import com.tongran.agentserver.server.collect.vo.DockerVO; import org.apache.commons.lang3.StringUtils; @@ -21,12 +20,7 @@ public class DockerServiceImpl implements DockerService { public List query(){ List list = new ArrayList<>(); // 配置Docker客户端 - DefaultDockerClientConfig config = DefaultDockerClientConfig.createDefaultConfigBuilder() - .withDockerHost("unix:///var/run/docker.sock") // 本地Docker - // 或远程Docker: .withDockerHost("tcp://your-docker-host:2375") - .build(); - DockerClient dockerClient = DockerClientBuilder.getInstance(config).build(); - + DockerClient dockerClient = DockerClientManager.getDockerClient(); try { // 获取正在运行的容器列表 List containers = dockerClient.listContainersCmd() diff --git a/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java b/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java index 9342cae..9881352 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java +++ b/src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java @@ -14,7 +14,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; -import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import javax.annotation.Resource; import java.util.concurrent.TimeUnit; @@ -31,6 +30,10 @@ public class NettyTcpClient { @Resource private AgentNettyConfig config; + @Resource + private DecoderHandler decoderHandler; + + // 目标服务器IP和端口 // private final String host = "www.tz2.xyz"; // 替换为实际IP // private final int port = 6610; // 替换为实际端口 @@ -38,7 +41,7 @@ public class NettyTcpClient { /** * 启动客户端连接 */ - @PostConstruct +// @PostConstruct public void start() { try { connect(); @@ -59,7 +62,7 @@ public class NettyTcpClient { protected void initChannel(SocketChannel ch) { ChannelPipeline pipeline = ch.pipeline(); // 添加编解码器 - pipeline.addLast("decoder", new DecoderHandler()); + pipeline.addLast("decoder", decoderHandler); pipeline.addLast("encoder", new StringEncoder()); // 添加心跳机制 pipeline.addLast("idleStateHandler", diff --git a/src/main/java/com/tongran/agentserver/server/netty/annotation/AgentDispatcher.java b/src/main/java/com/tongran/agentserver/server/netty/annotation/AgentDispatcher.java index 0827b85..bb2379b 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/annotation/AgentDispatcher.java +++ b/src/main/java/com/tongran/agentserver/server/netty/annotation/AgentDispatcher.java @@ -19,7 +19,7 @@ public @interface AgentDispatcher { @Getter @AllArgsConstructor enum VersionEnum { - V1("2025"); + V1("2026"); public final String value; } diff --git a/src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java b/src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java index 58cc4ab..1d22741 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java +++ b/src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java @@ -4,6 +4,7 @@ import com.tongran.agentserver.server.netty.annotation.AgentDispatcher; import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; +import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; @@ -11,6 +12,7 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @Component +@Order(1) public class AgentDispatcherManager implements ApplicationContextAware { /** 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 58b1b4c..cf96005 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 @@ -44,7 +44,7 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - cpuScheduler.updateInterval(intervalMillis); + cpuScheduler.updateSchedule(25000, intervalMillis); JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -52,69 +52,13 @@ public class AgentEndpoint { } } - @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"); - cpuScheduler.updateInterval(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.更新CPU采集间隔) - public class CpuHandler implements AgentHandler { - @Override - public UpMsgResponse upHandle(String data, String clientId) { - JSONObject jsonObject = JSONObject.parseObject(data); - long intervalMillis = jsonObject.getLong("intervalMillis"); - dockerScheduler.updateInterval(intervalMillis); - JSONObject json = new JSONObject(); - json.put("clientId",clientId); - json.put("resCode",1); - return UpMsgResponse.builder().dataType(MsgEnum.更新CPU采集间隔应答.getValue()).content(json.toString()).build(); - } - } - - @AgentDispatcher(msgId = MsgEnum.更新网卡采集间隔) - public class TimeNetHandler implements AgentHandler { - @Override - public UpMsgResponse upHandle(String data, String clientId) { - JSONObject jsonObject = JSONObject.parseObject(data); - long intervalMillis = jsonObject.getLong("intervalMillis"); - netScheduler.updateInterval(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 TimeSwitchHandler implements AgentHandler { - @Override - public UpMsgResponse upHandle(String data, String clientId) { - JSONObject jsonObject = JSONObject.parseObject(data); - long intervalMillis = jsonObject.getLong("intervalMillis"); - switchBoardScheduler.updateInterval(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 TimeDiskHandler implements AgentHandler { @Override public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - diskScheduler.updateInterval(intervalMillis); + diskScheduler.updateSchedule(35000, intervalMillis); JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -122,17 +66,17 @@ public class AgentEndpoint { } } - @AgentDispatcher(msgId = MsgEnum.更新挂载采集间隔) - public class TimePointHandler implements AgentHandler { + @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"); - pointScheduler.updateInterval(intervalMillis); + cpuScheduler.updateSchedule(45000, intervalMillis); JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); - return UpMsgResponse.builder().dataType(MsgEnum.更新挂载采集间隔应答.getValue()).content(json.toString()).build(); + return UpMsgResponse.builder().dataType(MsgEnum.更新容器采集间隔应答.getValue()).content(json.toString()).build(); } } @@ -142,7 +86,7 @@ public class AgentEndpoint { public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - memoryScheduler.updateInterval(intervalMillis); + memoryScheduler.updateSchedule(55000, intervalMillis); JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -150,13 +94,55 @@ public class AgentEndpoint { } } + @AgentDispatcher(msgId = MsgEnum.更新网卡采集间隔) + public class TimeNetHandler implements AgentHandler { + @Override + public UpMsgResponse upHandle(String data, String clientId) { + JSONObject jsonObject = JSONObject.parseObject(data); + long intervalMillis = jsonObject.getLong("intervalMillis"); + netScheduler.updateSchedule(65000, 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 TimePointHandler implements AgentHandler { + @Override + public UpMsgResponse upHandle(String data, String clientId) { + JSONObject jsonObject = JSONObject.parseObject(data); + long intervalMillis = jsonObject.getLong("intervalMillis"); + pointScheduler.updateSchedule(75000, 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 TimeSwitchHandler implements AgentHandler { + @Override + public UpMsgResponse upHandle(String data, String clientId) { + JSONObject jsonObject = JSONObject.parseObject(data); + long intervalMillis = jsonObject.getLong("intervalMillis"); + switchBoardScheduler.updateSchedule(85000, 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 TimeSysHandler implements AgentHandler { @Override public UpMsgResponse upHandle(String data, String clientId) { JSONObject jsonObject = JSONObject.parseObject(data); long intervalMillis = jsonObject.getLong("intervalMillis"); - sysScheduler.updateInterval(intervalMillis); + sysScheduler.updateSchedule(40000, intervalMillis); JSONObject json = new JSONObject(); json.put("clientId",clientId); json.put("resCode",1); @@ -165,7 +151,4 @@ public class AgentEndpoint { } - - - } \ No newline at end of file diff --git a/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java b/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java index 95e95d8..29e2e89 100644 --- a/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java +++ b/src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java @@ -1,6 +1,13 @@ package com.tongran.agentserver.server.netty.handler; +import cn.hutool.core.util.ObjectUtil; +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; +import com.tongran.agentserver.server.netty.annotation.AgentDispatcher; import com.tongran.agentserver.server.netty.basics.AgentDispatcherManager; +import com.tongran.agentserver.server.netty.basics.AgentHandler; +import com.tongran.agentserver.server.netty.model.Message; +import com.tongran.agentserver.server.netty.model.UpMsgResponse; import com.tongran.agentserver.utils.AssertLog; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandler; @@ -10,13 +17,15 @@ import io.netty.util.CharsetUtil; import org.springframework.stereotype.Component; import javax.annotation.Resource; +import java.util.Objects; @Component @ChannelHandler.Sharable public class DecoderHandler extends ChannelInboundHandlerAdapter { @Resource - private AgentDispatcherManager agentDispatcherManager; + AgentDispatcherManager agentDispatcherManager; + /** * 消息解码器 @@ -28,19 +37,28 @@ public class DecoderHandler extends ChannelInboundHandlerAdapter { ByteBuf byteBuf = (ByteBuf) msg; String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码 AssertLog.info("<<[up]:[up-content]==>{}", messages); -// JSONObject jsonObject = JSONObject.parseObject(messages); -// String clientId = jsonObject.getString("clientId"); -// String dataType = jsonObject.getString("dataType"); -// AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value); -// UpMsgResponse response = msgHandler.upHandle(messages,clientId); -// if(Objects.nonNull(response)){ -// Message agentMessage = Message.builder().build(); -// agentMessage.setClientId(clientId); -// agentMessage.setDataType(response.getDataType()); -// agentMessage.setData(response.getContent()); -// String json = "agent-tcp:"+ JSON.toJSONString(agentMessage)+"@tong-ran"; -// ctx.fireChannelRead(json);//传递到下一个handler -// } + JSONObject jsonObject = JSONObject.parseObject(messages); + String clientId = jsonObject.getString("clientId"); + String dataType = jsonObject.getString("dataType"); + AssertLog.info("<<11111111111111==>{}"); + if(Objects.nonNull(agentDispatcherManager)){ + AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value); + AssertLog.info("<<22222222222222==>{}"); + if (ObjectUtil.isNotEmpty(msgHandler)) { + AssertLog.info("<<3333333333333==>{}"); + UpMsgResponse response = msgHandler.upHandle(messages,clientId); + AssertLog.info("<<4444444444444==>{}"); + if(Objects.nonNull(response)){ + Message agentMessage = Message.builder().build(); + agentMessage.setClientId(clientId); + agentMessage.setDataType(response.getDataType()); + agentMessage.setData(response.getContent()); + String json = "agent-tcp:"+ JSON.toJSONString(agentMessage)+"@tong-ran"; + ctx.fireChannelRead(json);//传递到下一个handler + AssertLog.info("<<555555555555555==>{}"); + } + } + } byteBuf.release(); // 释放 ByteBuf 资源（重要！） } else { System.out.println("Unexpected message type: " + msg.getClass());