Agent/agent-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

接收服务端指令-更新采集时间

Browse Source
This commit is contained in:
qiminbao
2025-08-26 13:00:30 +08:00
parent bddbf10503
commit 1351585ed5
19 changed files with 790 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
+19
View File
@@ -91,6 +91,25 @@
<version>3.3.0</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>com.github.docker-java</groupId>-->
<!-- <artifactId>docker-java-transport-okhttp</artifactId>-->
<!-- <version>3.3.0</version>-->
<!-- </dependency>-->
<!-- Docker Java API -->
<dependency>
<groupId>com.github.docker-java</groupId>
<artifactId>docker-java-api</artifactId>
<version>3.3.0</version>
</dependency>
<!-- 使用 Apache HttpClient 5 实现 -->
<dependency>
<groupId>com.github.docker-java</groupId>
<artifactId>docker-java-transport-httpclient5</artifactId>
<version>3.3.0</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
src/main/java/com/tongran/agentserver/AgentServerApplication.java
+4
View File
@@ -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);
}
}
src/main/java/com/tongran/agentserver/controller/TcpController.java
+20 -27
View File
@@ -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连接未建立，消息发送失败";
// }
}
src/main/java/com/tongran/agentserver/scheduler/CpuScheduler.java
+65 -41
View File
@@ -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);
}
}
}
src/main/java/com/tongran/agentserver/scheduler/DiskScheduler.java
+64 -41
View File
@@ -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<DiskVO> 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<DiskVO> 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);
}
}
}
src/main/java/com/tongran/agentserver/scheduler/DockerScheduler.java
+64 -42
View File
@@ -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<DockerVO> 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<DockerVO> 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);
}
}
}
src/main/java/com/tongran/agentserver/scheduler/HeartScheduler.java
+74 -42
View File
@@ -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);
}
}
}
src/main/java/com/tongran/agentserver/scheduler/MemoryScheduler.java
+65 -41
View File
@@ -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);
}
}
}
src/main/java/com/tongran/agentserver/scheduler/NetScheduler.java
+65 -41
View File
@@ -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<NetVO> 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<NetVO> 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);
}
}
}
src/main/java/com/tongran/agentserver/scheduler/PointScheduler.java
+65 -41
View File
@@ -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<PointVO> 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<PointVO> 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);
}
}
}
src/main/java/com/tongran/agentserver/scheduler/SwitchBoardScheduler.java
+65 -40
View File
@@ -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<SwitchBoardVO> 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<SwitchBoardVO> 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);
}
}
src/main/java/com/tongran/agentserver/scheduler/SysScheduler.java
+65 -41
View File
@@ -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);
}
}
}
src/main/java/com/tongran/agentserver/server/collect/docker/DockerClientManager.java
+62
View File
@@ -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) {
// 记录日志
}
}
}
}
src/main/java/com/tongran/agentserver/server/collect/docker/impl/DockerServiceImpl.java
+2 -8
View File
@@ -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<DockerVO> query(){
List<DockerVO> 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<Container> containers = dockerClient.listContainersCmd()
src/main/java/com/tongran/agentserver/server/netty/NettyTcpClient.java
+6 -3
View File
@@ -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",
src/main/java/com/tongran/agentserver/server/netty/annotation/AgentDispatcher.java
+1 -1
View File
@@ -19,7 +19,7 @@ public @interface AgentDispatcher {
@Getter
@AllArgsConstructor
enum VersionEnum {
V1("2025");
V1("2026");
public final String value;
}
src/main/java/com/tongran/agentserver/server/netty/basics/AgentDispatcherManager.java
+2
View File
@@ -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 {
/**
src/main/java/com/tongran/agentserver/server/netty/enpoint/AgentEndpoint.java
+50 -67
View File
@@ -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 {
}
}
src/main/java/com/tongran/agentserver/server/netty/handler/DecoderHandler.java
+32 -14
View File
@@ -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());