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