diff --git a/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java b/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java index 928dde6..f1879da 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java +++ b/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java @@ -65,4 +65,21 @@ public class SchedulerConfig { executor.initialize(); return executor; } + + // 新增：心跳专用线程池 + @Bean("heartbeatExecutor") + public ThreadPoolTaskExecutor heartbeatExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(2); + executor.setMaxPoolSize(4); + executor.setQueueCapacity(0); // 重要：设为0，不要队列！ + executor.setKeepAliveSeconds(30); + executor.setThreadNamePrefix("heartbeat-"); + + // 重要：使用CallerRunsPolicy，确保心跳一定执行 + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + + executor.initialize(); + return executor; + } } \ No newline at end of file diff --git a/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java b/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java index 9f373fb..7312400 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java +++ b/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java @@ -119,7 +119,7 @@ public class BusinessTasks { /** * 任务1：心跳上报任务 */ - @Async("taskExecutor") + @Async("heartbeatExecutor") public void heartbeatTask() { long timestamp = System.currentTimeMillis(); timestamp = Math.round(timestamp / 1000.0); diff --git a/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java b/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java index def0038..fe2e4d5 100644 --- a/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java +++ b/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java @@ -2454,44 +2454,76 @@ public class AgentServiceImpl implements AgentService { } private void processSingleFile(File file, boolean lastTraffic) { - // 使用Properties类读取，并指定UTF-8编码 - Properties props = new Properties(); - try (InputStream input = Files.newInputStream(file.toPath()); - InputStreamReader reader = new InputStreamReader(input, StandardCharsets.UTF_8)) { // 关键：指定UTF-8编码 - - props.load(reader); // 使用指定编码的Reader加载 - String data = props.getProperty("traffic"); - if (StringUtils.isNotBlank(data)) { - List list = JSONArray.parseArray(data, NetVO.class); - if(list != null){ - for (NetVO netVO : list) { - netVO.setLastTrafficFlag(lastTraffic); + // 重试配置 + int maxRetry = 3; // 最大重试次数 + for (int retryCount = 0; retryCount < maxRetry; retryCount++) { + try { + // 每次重试前都检查连接 + boolean success = true; + int activeConnect = client.getActiveConnections(); + if (activeConnect == 0) { + success = connection(); + if (!success) { + AssertLog.info("连接失败，第 {} 次重试，文件：{}", + retryCount + 1, file.getName()); + continue; } } - data = JSONObject.toJSONString(list); - Message message = Message.builder() - .clientId(GlobalConfig.CLIENT_ID) - .dataType(MsgEnum.网络上报重试.getValue()) - .data(data) - .build(); + // 使用Properties类读取，并指定UTF-8编码 + Properties props = new Properties(); + try (InputStream input = Files.newInputStream(file.toPath()); + InputStreamReader reader = new InputStreamReader(input, StandardCharsets.UTF_8)) { // 关键：指定UTF-8编码 - sessionManager.writeAndFlush( - sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), - message - ); + props.load(reader); // 使用指定编码的Reader加载 + String data = props.getProperty("traffic"); + if (StringUtils.isNotBlank(data)) { + List list = JSONArray.parseArray(data, NetVO.class); + if(list != null){ + for (NetVO netVO : list) { + netVO.setLastTrafficFlag(lastTraffic); + } + } + data = JSONObject.toJSONString(list); + Message message = Message.builder() + .clientId(GlobalConfig.CLIENT_ID) + .dataType(MsgEnum.网络上报重试.getValue()) + .data(data) + .build(); - if (lastTraffic) { - AssertLog.info("发送最后一条重试网络信息包={}", JSON.toJSONString(message)); - } else { - AssertLog.info("发送重试recover网络信息包={}", JSON.toJSONString(message)); + sessionManager.writeAndFlush( + sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), + message + ); + + if (lastTraffic) { + AssertLog.info("发送最后一条重试网络信息包={}", JSON.toJSONString(message)); + } else { + AssertLog.info("发送重试recover网络信息包={}", JSON.toJSONString(message)); + } + return; // 成功返回 + } else { + AssertLog.info("文件中没有有效数据，删除文件：{}", file.getName()); + file.delete(); + return; + } + } + }catch (Exception e) { + AssertLog.error("loadNetFile异常，文件：{}, 重试次数：{}, 异常信息:{}", + file.getName(), retryCount + 1, e.getMessage()); + + // 如果不是最后一次重试，等待2秒再试 + if (retryCount < maxRetry - 1) { + try { + Thread.sleep(2000); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + return; + } } - } else { - AssertLog.info("文件中没有有效数据，删除文件：{}", file.getName()); - file.delete(); } - } catch (Exception e) { - AssertLog.error("loadNetFile异常，文件：{},异常信息:{}", file.getName(), e.getMessage()); } + // 所有重试都失败 + AssertLog.error("文件处理失败，已达最大重试次数({})，文件：{}", maxRetry, file.getName()); } public void caseTypeBySystem(String type, int interval, boolean collect){ diff --git a/src/main/java/com/tongran/agent/client/utils/PublicIpFetcher.java b/src/main/java/com/tongran/agent/client/utils/PublicIpFetcher.java index b57cb46..fe2a025 100644 --- a/src/main/java/com/tongran/agent/client/utils/PublicIpFetcher.java +++ b/src/main/java/com/tongran/agent/client/utils/PublicIpFetcher.java @@ -215,7 +215,7 @@ public class PublicIpFetcher { return result; } catch (Exception e) { AssertLog.error("HTTP主服务请求失败: {}", e.getMessage()); - return tryHttpFallback(); + throw e; // 不在这里降级，只抛异常 } }); @@ -224,6 +224,7 @@ public class PublicIpFetcher { } catch (TimeoutException e) { AssertLog.error("HTTP请求超时"); future.cancel(true); + return tryHttpFallback(); // 主线程执行降级 } catch (Exception e) { AssertLog.error("HTTP请求异常: {}", e.getMessage()); } finally {