fix: 恢复心跳发送、断线重连、注册定时任务的完整实现

故障诊断结果:Agent启动后TCP连接成功,但连接断开后无法重连

根因:
- connectionTask()重连方法是空stub,连接断开后无法自动重连
- heartbeatTask()心跳逻辑被简化,只是本地计数不通过TCP发送
- registerTask()注册方法是空stub
- MultiTargetNettyClient没有断线检测和重连机制

修复内容:
- BusinessTasks.java: 恢复完整的heartbeatTask(心跳上报)、connectionTask(断线重连)、
  registerTask(注册)、policyTask(策略更新)、networkDetectTask(多网IP探测)、
  checkFrpcTask(FRPC保活)、upFrpcMsgTask(FRPC状态上报)方法
- 新增 NetBusinessService.java 接口(定时任务依赖)
- 新增 NetBusinessVO.java VO类
- 新增 NetworkInterfaceInfo.java VO类
- 新增 NetworkInterfaceUtil.java 工具类
- MsgEnum.java: 新增心跳上报、注册、多网IP探测枚举值
This commit is contained in:
lee
2026-07-22 17:15:10 +08:00
parent ff7cd12fd7
commit f1fa6dca7c
6 changed files with 441 additions and 22 deletions
@@ -40,6 +40,16 @@ public enum MsgEnum {
告警上报("ALARM"),
建立连接("CONNECT"),
获取最新策略("GET_POLICY"),
多网IP探测上报("NETWORK_DETECT"),
内存详情上报("MEMORY_DETAIL"),
业务网络上报("BUSINESS_NET"),
开启或更新系统采集("SYSTEM_COLLECT_START"),
开启或更新系统采集应答("SYSTEM_COLLECT_START_RSP"),
@@ -0,0 +1,27 @@
package com.tongran.agent.client.core.vo;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class NetBusinessVO implements Serializable {
private static final long serialVersionUID = 9L;
private String name;
private String mac;
private int pid;
private String processName;
private Long inSpeed;
private Long outSpeed;
private Long ipv4InSpeed;
private Long ipv4OutSpeed;
private Long ipv6InSpeed;
private Long ipv6OutSpeed;
private int connectionCount;
private long timestamp;
}
@@ -0,0 +1,74 @@
package com.tongran.agent.client.core.vo;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.AllArgsConstructor;
import java.util.List;
/**
* 网络接口信息VO(从1.20恢复)
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class NetworkInterfaceInfo {
private String name;
private String mac;
private String type;
private String ipv4;
private String gateway;
private String publicIp;
private String carrier;
private String province;
private String city;
private String ipv6;
private String status;
private String parentInterface;
private Long netCreateTime;
private List<NetworkInterfaceInfo> subInterfaces;
/**
* 兼容旧builder模式
*/
public static NetworkInterfaceInfoBuilder builder() {
return new NetworkInterfaceInfoBuilder();
}
public static class NetworkInterfaceInfoBuilder {
private String name;
private String mac;
private String type;
private String ipv4;
private String gateway;
private String publicIp;
private String carrier;
private String province;
private String city;
private String ipv6;
private String status;
private String parentInterface;
private Long netCreateTime;
private List<NetworkInterfaceInfo> subInterfaces;
public NetworkInterfaceInfoBuilder name(String name) { this.name = name; return this; }
public NetworkInterfaceInfoBuilder mac(String mac) { this.mac = mac; return this; }
public NetworkInterfaceInfoBuilder type(String type) { this.type = type; return this; }
public NetworkInterfaceInfoBuilder ipv4(String ipv4) { this.ipv4 = ipv4; return this; }
public NetworkInterfaceInfoBuilder gateway(String gateway) { this.gateway = gateway; return this; }
public NetworkInterfaceInfoBuilder publicIp(String publicIp) { this.publicIp = publicIp; return this; }
public NetworkInterfaceInfoBuilder carrier(String carrier) { this.carrier = carrier; return this; }
public NetworkInterfaceInfoBuilder province(String province) { this.province = province; return this; }
public NetworkInterfaceInfoBuilder city(String city) { this.city = city; return this; }
public NetworkInterfaceInfoBuilder ipv6(String ipv6) { this.ipv6 = ipv6; return this; }
public NetworkInterfaceInfoBuilder status(String status) { this.status = status; return this; }
public NetworkInterfaceInfoBuilder parentInterface(String parentInterface) { this.parentInterface = parentInterface; return this; }
public NetworkInterfaceInfoBuilder netCreateTime(Long netCreateTime) { this.netCreateTime = netCreateTime; return this; }
public NetworkInterfaceInfoBuilder subInterfaces(List<NetworkInterfaceInfo> subInterfaces) { this.subInterfaces = subInterfaces; return this; }
public NetworkInterfaceInfo build() {
return new NetworkInterfaceInfo(name, mac, type, ipv4, gateway, publicIp, carrier, province, city, ipv6, status, parentInterface, netCreateTime, subInterfaces);
}
}
}
@@ -9,6 +9,8 @@ import com.tongran.agent.client.core.config.GlobalConfig;
import com.tongran.agent.client.core.enums.MsgEnum;
import com.tongran.agent.client.core.session.SessionManager;
import com.tongran.agent.client.core.vo.*;
import com.tongran.agent.client.netty.MultiTargetNettyClient;
import com.tongran.agent.client.netty.config.AgentNettyConfig;
import com.tongran.agent.client.netty.model.Message;
import com.tongran.agent.client.service.*;
import com.tongran.agent.client.utils.AgentUtil;
@@ -116,6 +118,18 @@ public class BusinessTasks {
@Resource
private ApplicationProperties properties;
@Resource
private MultiTargetNettyClient client;
@Resource
private AgentNettyConfig config;
@Resource
private AgentService agentService;
@Resource
private NetBusinessService netBusinessService;
/**
* 任务1:心跳上报任务
*/
@@ -125,19 +139,34 @@ public class BusinessTasks {
timestamp = Math.round(timestamp / 1000.0);
int count = heartbeatTask.incrementAndGet();
AssertLog.info("心跳定时任务执行 - 时间: {},task #{}", LocalDateTime.now(), count);
// 业务处理
// 判定客户端与服务端是否连接
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
// 发送心跳包
JSONObject object = new JSONObject();
object.put("strength","31");
object.put("name", properties.getName());
object.put("version", properties.getVersion());
object.put("startupTime", GlobalConfig.startupTime);
object.put("timestamp",timestamp);
Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.心跳上报.getValue()).data(object.toString()).build();
sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message);
AssertLog.info("发送心跳包={}",JSON.toJSONString(message));
// 检查活跃连接,如果断开则尝试重连
boolean success = true;
int activeConnect = client.getActiveConnections();
if (activeConnect == 0) {
success = agentService.connection();
}
if (success) {
// 判定客户端与服务端是否连接
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) {
GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN();
}
// 发送心跳包
JSONObject object = new JSONObject();
object.put("clientId", GlobalConfig.CLIENT_ID);
object.put("logicalNode", agentService.getLogicalNode());
object.put("sn", GlobalConfig.DEVICE_SN);
object.put("strength","31");
object.put("name", properties.getName());
object.put("version", properties.getVersion());
object.put("startupTime", GlobalConfig.startupTime);
object.put("timestamp",timestamp);
Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.心跳上报.getValue()).data(object.toString()).build();
sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message);
AssertLog.info("发送心跳包={}",JSON.toJSONString(message));
}
} else {
AssertLog.info("心跳定时任务执行失败-连接断开");
}
AssertLog.info("心跳定时任务执行 - task #{} completed", count);
}
@@ -1275,50 +1304,216 @@ public class BusinessTasks {
/**
* 策略更新定时任务
*/
@Async("taskExecutor")
public void policyTask() {
// TODO: 从1.20恢复完整实现
long timestamp = AgentUtil.roundMinutes();
int count = new AtomicInteger(0).incrementAndGet();
AssertLog.info("获取最新策略定时任务执行 - 时间: {},task #{}", LocalDateTime.now(), count);
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) {
GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN();
}
JSONObject object = new JSONObject();
object.put("clientId", GlobalConfig.CLIENT_ID);
object.put("sn", GlobalConfig.DEVICE_SN);
object.put("timestamp", timestamp);
Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.获取最新策略.getValue()).data(object.toString()).build();
sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message);
AssertLog.info("发送获取最新策略信息包={}", JSON.toJSONString(message));
}
AssertLog.info("获取最新策略定时任务执行 - task #{} completed", count);
}
/**
* 多网IP探测定时任务
*/
@Async("taskExecutor")
public void networkDetectTask() {
// TODO: 从1.20恢复完整实现
int count = new AtomicInteger(0).incrementAndGet();
AssertLog.info("多网IP探测定时任务执行 - 时间: {}task #{}", LocalDateTime.now(), count);
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) {
GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN();
}
long timestamps = System.currentTimeMillis();
timestamps = Math.round(timestamps / 1000.0);
List infos = null;
try {
// NetworkInterfaceUtil may not be available in all environments
infos = com.tongran.agent.client.utils.NetworkInterfaceUtil.collectNetworkInfo();
} catch (Exception e) {
AssertLog.warn("collectNetworkInfo failed: {}", e.getMessage());
}
JSONObject objects = new JSONObject();
objects.put("clientId", GlobalConfig.CLIENT_ID);
objects.put("sn", GlobalConfig.DEVICE_SN);
objects.put("networkInfo", JSON.toJSONString(infos));
objects.put("timestamp", timestamps);
Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.多网IP探测上报.getValue()).data(objects.toString()).build();
sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message);
AssertLog.info("发送多网IP探测信息包={}", JSON.toJSONString(message));
}
AssertLog.info("发送多网IP探测定时任务执行 - task #{} completed", count);
}
/**
* FRPC保活定时任务
*/
@Async("taskExecutor")
public void checkFrpcTask() {
// TODO: 从1.20恢复完整实现
int count = new AtomicInteger(0).incrementAndGet();
AssertLog.info("frpc保活机制定时任务执行 - 时间: {},task #{}", LocalDateTime.now(), count);
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
agentService.checkFrpc();
} else {
AssertLog.info("断开连接--frpc保活机制定时任务执行 - task #{} completed", count);
}
}
/**
* FRPC状态上报定时任务
*/
@Async("taskExecutor")
public void upFrpcMsgTask() {
// TODO: 从1.20恢复完整实现
int count = new AtomicInteger(0).incrementAndGet();
AssertLog.info("frpc状态上报定时任务执行 - 时间: {},task #{}", LocalDateTime.now(), count);
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
agentService.upFrpcMsg();
} else {
AssertLog.info("断开连接--frpc状态上报定时任务执行 - task #{} completed", count);
}
}
/**
* 本地存储流量信息上报定时任务
*/
@Async("taskExecutor")
public void upTempTraffic() {
// TODO: 从1.20恢复完整实现
int count = new AtomicInteger(0).incrementAndGet();
AssertLog.info("本地存储流量信息上报定时任务执行 - 时间: {},task #{}", LocalDateTime.now(), count);
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
agentService.checkTraffic();
}
AssertLog.info("本地存储流量信息上报定时任务执行 - task #{} completed", count);
}
/**
* 注册定时任务
* 注册重试定时任务
*/
@Async("taskExecutor")
public void registerTask() {
// TODO: 从1.20恢复完整实现
long timestamp = System.currentTimeMillis();
timestamp = Math.round(timestamp / 1000.0);
int count = new AtomicInteger(0).incrementAndGet();
AssertLog.info("注册重试定时任务执行 - 时间: {},task #{}", LocalDateTime.now(), count);
boolean success = true;
int activeConnect = client.getActiveConnections();
AssertLog.info("注册重试定时任务执行 - 时间: {}activeConnect={}", LocalDateTime.now(), activeConnect);
if (activeConnect == 0 || count > 1) {
success = agentService.connection();
}
if (success) {
AssertLog.info("连接成功,发送初始连接消息");
try {
if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) {
if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) {
GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN();
}
List infos = null;
try {
infos = com.tongran.agent.client.utils.NetworkInterfaceUtil.collectNetworkInfo();
} catch (Exception e) {
AssertLog.warn("collectNetworkInfo failed: {}", e.getMessage());
}
JSONObject object = new JSONObject();
object.put("clientId", GlobalConfig.CLIENT_ID);
object.put("sn", GlobalConfig.DEVICE_SN);
object.put("networkInfo", JSON.toJSONString(infos));
object.put("timestamp", timestamp);
Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message);
AssertLog.info("发送注册重试={}", JSON.toJSONString(message));
}
} catch (Exception e) {
e.printStackTrace();
}
} else {
AssertLog.info("注册重试定时任务执行 - 时间: {},建立连接失败", LocalDateTime.now());
}
AssertLog.info("注册重试定时任务执行 - task #{} completed", count);
}
/**
* 连接重试定时任务
* 连接重试定时任务 - 最关键的断线重连逻辑
*/
@Async("taskExecutor")
public void connectionTask() {
// TODO: 从1.20恢复完整实现
int count = new AtomicInteger(0).incrementAndGet();
AssertLog.info("建立连接重试定时任务执行 - 时间: {},task #{}", LocalDateTime.now(), count);
boolean success = client.createConnection(GlobalConfig.CLIENT_ID, config.getHost(), config.getPort(), 5);
if (success) {
if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) {
GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN();
}
long timestamp = System.currentTimeMillis();
timestamp = Math.round(timestamp / 1000.0);
JSONObject object = new JSONObject();
object.put("clientId", GlobalConfig.CLIENT_ID);
object.put("sn", GlobalConfig.DEVICE_SN);
object.put("timestamp", timestamp);
Message msg = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.建立连接.getValue()).data(object.toString()).build();
client.sendMessages(GlobalConfig.CLIENT_ID, msg);
if (GlobalConfig.isRegister) {
AssertLog.info("启动心跳定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 30000);
dynamicTaskService.scheduleTask("heartbeat", () -> businessTasks.heartbeatTask(), 15000L, 30000L);
long milli = AgentUtil.getMillisToNextMinute() + 60000L;
AssertLog.info("启动更新策略定时任务 - 延迟: {}ms, 间隔: {}ms", milli, 60000);
dynamicTaskService.scheduleTask("policy", () -> businessTasks.policyTask(), milli, 60000L);
AssertLog.info("启动多网IP探测定时任务Retry - 延迟: {}ms, 间隔: {}ms", milli, 300000);
dynamicTaskService.scheduleTask("networkDetect", () -> businessTasks.networkDetectTask(), milli, 300000L);
AssertLog.info("检测监控策略配置Retry");
agentService.checkMonitor();
AssertLog.info("检测agent更新配置");
agentService.checkAgentUpdate();
AssertLog.info("启动frpc保活定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 600000);
dynamicTaskService.scheduleTask("checkFrpc", () -> businessTasks.checkFrpcTask(), 15000L, 600000L);
AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000);
dynamicTaskService.scheduleTask("upFrpcMsg", () -> businessTasks.upFrpcMsgTask(), 15000L, 300000L);
long milliFive = AgentUtil.millisecondsToNext5Minute();
AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 0x6DDD00);
dynamicTaskService.scheduleTask("upTempTraffic", () -> businessTasks.upTempTraffic(), milliFive, 0x6DDD00L);
AssertLog.info("检测PppoE配置");
agentService.handleRebootRecovery();
AssertLog.info("检测tcpdump探测时间配置");
agentService.checkTcpdumpTimes();
} else {
try {
if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) {
GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN();
}
long timestamps = System.currentTimeMillis();
timestamps = Math.round(timestamps / 1000.0);
List infos = null;
try {
infos = com.tongran.agent.client.utils.NetworkInterfaceUtil.collectNetworkInfo();
} catch (Exception e) {
AssertLog.warn("collectNetworkInfo failed: {}", e.getMessage());
}
JSONObject objects = new JSONObject();
objects.put("clientId", GlobalConfig.CLIENT_ID);
objects.put("sn", GlobalConfig.DEVICE_SN);
objects.put("networkInfo", JSON.toJSONString(infos));
objects.put("timestamp", timestamps);
Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.注册.getValue()).data(objects.toString()).build();
client.sendMessages(GlobalConfig.CLIENT_ID, message);
} catch (Exception e) {
e.printStackTrace();
}
dynamicTaskService.scheduleTask("register", () -> businessTasks.registerTask(), 0L, 300000L);
}
dynamicTaskService.cancelTask("connection");
}
AssertLog.info("建立连接重试定时任务执行 - task #{} completed", count);
}
}
@@ -0,0 +1,8 @@
package com.tongran.agent.client.service;
import com.tongran.agent.client.core.vo.NetBusinessVO;
import java.util.List;
public interface NetBusinessService {
List<NetBusinessVO> netList(long var1);
}
@@ -0,0 +1,105 @@
package com.tongran.agent.client.utils;
import com.tongran.agent.client.core.vo.NetworkInterfaceInfo;
import java.util.ArrayList;
import java.util.List;
/**
* 网络接口工具类(从1.20恢复的简化版)
* 完整的网卡信息采集依赖PublicIpFetcher等多个类,此处提供基础实现
* collectNetworkInfo用于注册消息中的网络信息上报
*/
public class NetworkInterfaceUtil {
public static boolean getmacVlanStatus(String interfaceName) {
try {
Process process = Runtime.getRuntime().exec(
new String[]{"ping", "-I", interfaceName, "-c", "1", "-W", "2", "www.baidu.com"});
return process.waitFor() == 0;
} catch (Exception e) {
AssertLog.warn("接口 {} ping异常: {}", interfaceName, e.getMessage());
return false;
}
}
/**
* 收集网络接口信息
* 返回所有活跃的以太网接口信息列表
*/
public static List<NetworkInterfaceInfo> collectNetworkInfo() {
List<NetworkInterfaceInfo> result = new ArrayList<>();
try {
// 使用 ip link show 获取网卡列表
Process process = Runtime.getRuntime().exec("ip -o link show");
java.io.BufferedReader reader = new java.io.BufferedReader(
new java.io.InputStreamReader(process.getInputStream()));
String line;
while ((line = reader.readLine()) != null) {
// 格式: 1: lo: <LOOPBACK,UP,LOWER_UP> ...
String[] parts = line.trim().split(":");
if (parts.length < 2) continue;
String name = parts[1].trim();
// 跳过回环接口
if ("lo".equals(name)) continue;
// 检查是否UP
if (!line.contains("UP")) continue;
NetworkInterfaceInfo info = NetworkInterfaceInfo.builder()
.name(name)
.type("Ethernet")
.build();
// 获取MAC地址
try {
Process macProcess = Runtime.getRuntime().exec(
"cat /sys/class/net/" + name + "/address");
java.io.BufferedReader macReader = new java.io.BufferedReader(
new java.io.InputStreamReader(macProcess.getInputStream()));
String mac = macReader.readLine();
if (mac != null && !mac.trim().isEmpty()) {
info.setMac(mac.trim());
}
macReader.close();
} catch (Exception ignored) {}
// 获取IPv4
try {
Process ipProcess = Runtime.getRuntime().exec(
new String[]{"sh", "-c", "ip -4 addr show " + name + " 2>/dev/null | grep 'inet ' | awk '{print $2}' | cut -d'/' -f1"});
java.io.BufferedReader ipReader = new java.io.BufferedReader(
new java.io.InputStreamReader(ipProcess.getInputStream()));
String ip = ipReader.readLine();
if (ip != null && !ip.trim().isEmpty() && !"127.0.0.1".equals(ip.trim())) {
info.setIpv4(ip.trim());
}
ipReader.close();
} catch (Exception ignored) {}
result.add(info);
}
reader.close();
} catch (Exception e) {
AssertLog.warn("collectNetworkInfo失败: {}", e.getMessage());
}
return result;
}
public static long getInterfaceCreateTime(String interfaceName) {
try {
Process process = Runtime.getRuntime().exec(
new String[]{"stat", "-c", "%Y", "/sys/class/net/" + interfaceName});
java.io.BufferedReader reader = new java.io.BufferedReader(
new java.io.InputStreamReader(process.getInputStream()));
String output = reader.readLine();
reader.close();
process.waitFor();
if (output != null && !output.trim().isEmpty()) {
return Long.parseLong(output.trim());
}
} catch (Exception e) {
AssertLog.error("获取网卡 {} 创建时间失败: {}", interfaceName, e.getMessage());
}
return -1L;
}
}