优化ip公网探测降级逻辑。

优化网络流量重试个别文件发送失败逻辑。
This commit is contained in:
gaoyutao
2026-02-27 17:49:01 +08:00
parent 3ff73519f9
commit f453f17903
4 changed files with 83 additions and 33 deletions
@@ -65,4 +65,21 @@ public class SchedulerConfig {
executor.initialize(); executor.initialize();
return executor; 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;
}
} }
@@ -119,7 +119,7 @@ public class BusinessTasks {
/** /**
* 任务1:心跳上报任务 * 任务1:心跳上报任务
*/ */
@Async("taskExecutor") @Async("heartbeatExecutor")
public void heartbeatTask() { public void heartbeatTask() {
long timestamp = System.currentTimeMillis(); long timestamp = System.currentTimeMillis();
timestamp = Math.round(timestamp / 1000.0); timestamp = Math.round(timestamp / 1000.0);
@@ -2454,44 +2454,76 @@ public class AgentServiceImpl implements AgentService {
} }
private void processSingleFile(File file, boolean lastTraffic) { private void processSingleFile(File file, boolean lastTraffic) {
// 使用Properties类读取,并指定UTF-8编码 // 重试配置
Properties props = new Properties(); int maxRetry = 3; // 最大重试次数
try (InputStream input = Files.newInputStream(file.toPath()); for (int retryCount = 0; retryCount < maxRetry; retryCount++) {
InputStreamReader reader = new InputStreamReader(input, StandardCharsets.UTF_8)) { // 关键:指定UTF-8编码 try {
// 每次重试前都检查连接
props.load(reader); // 使用指定编码的Reader加载 boolean success = true;
String data = props.getProperty("traffic"); int activeConnect = client.getActiveConnections();
if (StringUtils.isNotBlank(data)) { if (activeConnect == 0) {
List<NetVO> list = JSONArray.parseArray(data, NetVO.class); success = connection();
if(list != null){ if (!success) {
for (NetVO netVO : list) { AssertLog.info("连接失败,第 {} 次重试,文件:{}",
netVO.setLastTrafficFlag(lastTraffic); retryCount + 1, file.getName());
continue;
} }
} }
data = JSONObject.toJSONString(list); // 使用Properties类读取,并指定UTF-8编码
Message message = Message.builder() Properties props = new Properties();
.clientId(GlobalConfig.CLIENT_ID) try (InputStream input = Files.newInputStream(file.toPath());
.dataType(MsgEnum.网络上报重试.getValue()) InputStreamReader reader = new InputStreamReader(input, StandardCharsets.UTF_8)) { // 关键:指定UTF-8编码
.data(data)
.build();
sessionManager.writeAndFlush( props.load(reader); // 使用指定编码的Reader加载
sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), String data = props.getProperty("traffic");
message if (StringUtils.isNotBlank(data)) {
); List<NetVO> 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) { sessionManager.writeAndFlush(
AssertLog.info("发送最后一条重试网络信息包={}", JSON.toJSONString(message)); sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(),
} else { message
AssertLog.info("发送重试recover网络信息包={}", JSON.toJSONString(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){ public void caseTypeBySystem(String type, int interval, boolean collect){
@@ -215,7 +215,7 @@ public class PublicIpFetcher {
return result; return result;
} catch (Exception e) { } catch (Exception e) {
AssertLog.error("HTTP主服务请求失败: {}", e.getMessage()); AssertLog.error("HTTP主服务请求失败: {}", e.getMessage());
return tryHttpFallback(); throw e; // 不在这里降级,只抛异常
} }
}); });
@@ -224,6 +224,7 @@ public class PublicIpFetcher {
} catch (TimeoutException e) { } catch (TimeoutException e) {
AssertLog.error("HTTP请求超时"); AssertLog.error("HTTP请求超时");
future.cancel(true); future.cancel(true);
return tryHttpFallback(); // 主线程执行降级
} catch (Exception e) { } catch (Exception e) {
AssertLog.error("HTTP请求异常: {}", e.getMessage()); AssertLog.error("HTTP请求异常: {}", e.getMessage());
} finally { } finally {