From 64d307134bfd5dfcc96a6bf363d385004ee33263 Mon Sep 17 00:00:00 2001
From: zhaowenyu <35589726@qq.com>
Date: Sat, 17 Jan 2026 16:18:47 +0800
Subject: [PATCH] 2026-01-17 zwy 优化
---
ard-sample/src/main/java/com/ard/sample/manage/service/impl/WaylineTrajectoryServiceImpl.java | 80 ++++++++++++++++++++++++----------------
1 files changed, 48 insertions(+), 32 deletions(-)
diff --git a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/WaylineTrajectoryServiceImpl.java b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/WaylineTrajectoryServiceImpl.java
index 5b2c5a1..b3b9d8e 100644
--- a/ard-sample/src/main/java/com/ard/sample/manage/service/impl/WaylineTrajectoryServiceImpl.java
+++ b/ard-sample/src/main/java/com/ard/sample/manage/service/impl/WaylineTrajectoryServiceImpl.java
@@ -74,18 +74,18 @@
if (type.equals("0")) {
// 飞行轨迹数据
listTrajectory.add(dataJson);
- log.info("轨迹点数据添加到列表缓存，当前列表数量：{}", listTrajectory.size());
+ log.info("SN:{} 轨迹点数据添加到列表缓存，当前列表数量：{}", sn, listTrajectory.size());
+// log.info("SN:{} 轨迹点数据：{}", sn, dataJson);
}
if (type.equals("1")) {
// 任务状态数据
listTask.add(dataJson);
- log.info("任务状态数据添加到列表缓存，当前列表数量：{}", listTask.size());
+ log.info("SN:{} 任务状态数据添加到列表缓存，当前列表数量：{}", sn, listTask.size());
}
- //region 先弃用
// jobid、status字段赋值
if (!listTask.isEmpty()) {
String json = listTask.get(listTask.size() - 1);
- log.info("当前任务记录最后一条数据:{}", json);
+ log.info("SN:{} 当前任务记录最后一条数据:{}", sn, json);
try {
EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = mapper.readValue(json,
new TypeReference<EventsReceiver<WaylineTaskProgressReceiver>>() {
@@ -93,35 +93,27 @@
jobId = eventsReceiver.getOutput().getExt().getFlightId();
isOK = eventsReceiver.getOutput().getStatus();
- log.info("当前任务jobId:{},任务执行进度isOK:{}", jobId, isOK);
+ log.info("SN:{} 当前任务jobId:{},任务执行进度isOK:{}", sn, jobId, isOK);
} catch (Exception e) {
- log.error("解析任务状态JSON失败: {}", e.getMessage());
- log.error("JSON内容: {}", json);
+ log.error("SN:{} 解析任务状态JSON失败: {}", sn, e.getMessage());
+ log.error("SN:{} JSON内容: {}", sn, json);
}
}
+ else {
+ // 当前任务状态数据为空时，清空缓存
+ log.info("SN:{} 任务状态数据为空，当前无任务清空缓存", sn);
+ // 清空缓存
+ cacheTrajectoryMap.put(sn, new ArrayList<>());
+ cacheTaskMap.put(sn, new ArrayList<>());
+ return;
+ }
// listTrajectory.size() >= CHUNK_SIZE ||
- // endregion
if (listTask.size() >= CHUNK_SIZE) {
-// String json = listTask.get(listTask.size() - 1);
-// log.info("当前任务记录最后一条数据:{}", json);
-// try {
-// EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = mapper.readValue(json,
-// new TypeReference<EventsReceiver<WaylineTaskProgressReceiver>>() {
-// });
-// jobId = eventsReceiver.getOutput().getExt().getFlightId();
-// isOK = eventsReceiver.getOutput().getStatus();
-//
-// log.info("当前任务jobId:{},任务执行进度isOK:{}", jobId, isOK);
-//
-// } catch (Exception e) {
-// log.error("解析任务状态JSON失败: {}", e.getMessage());
-// log.error("JSON内容: {}", json);
-// }
-// progress = Convert.toStr(pro + 1);
// 分片保存
save(sn, jobId, "0");
- log.info("当前任务保存成功，jobId:{}, listTrajectory.size():{}，listTask.size():{}", jobId, listTrajectory.size(), listTask.size());
+ log.info("SN:{} 当前任务保存成功，jobId:{}, listTrajectory.size():{}，listTask.size():{}",
+ sn, jobId, listTrajectory.size(), listTask.size());
// log.info("当前任务保存成功，jobId:{},进度：{}",jobId , progress);
} else {
// 判断任务状态json中的status字段，如果为OK，则保存数据，一条完整的航线数据结束。
@@ -129,31 +121,48 @@
if (isOK.equals("ok")) {
String progress = "ok";
if (getWaylineTrajectoryByProgress(jobId, progress)) {
- log.info("当前任务已完成，不再保存数据！jobId:{},progress:{}", jobId, progress);
+ log.info("SN:{} 当前任务已完成，不再保存数据！jobId:{},progress:{}", sn, jobId, progress);
+ // 清空缓存
+ cacheTrajectoryMap.put(sn, new ArrayList<>());
+ cacheTaskMap.put(sn, new ArrayList<>());
return;
}
save(sn, jobId, progress);
- log.info("当前任务保存成功，jobId:{},进度：{}",jobId , progress);
+ log.info("SN:{} 当前任务保存成功，jobId:{},进度：{}", sn,jobId , progress);
}
} catch (Exception e) {
- log.error("json解析异常！{}", e.getMessage());
+ log.error("SN:{} json解析异常！{}", sn, e.getMessage());
}
}
}
@Async
public void save(String sn, String JobId, String progress) {
- List<String> trajectoryList = cacheTrajectoryMap.get(sn);
- List<String> taskList = cacheTaskMap.get(sn);
+// List<String> trajectoryList = cacheTrajectoryMap.get(sn);
+// List<String> taskList = cacheTaskMap.get(sn);
// String progress = cacheProgress.get(sn);
+ // 关键修复：原子性remove获取缓存（拿到当前所有数据）+ 清空缓存，避免异步竞态
+ List<String> trajectoryList = cacheTrajectoryMap.remove(sn);
+ List<String> taskList = cacheTaskMap.remove(sn);
+ // 处理remove返回null的情况（缓存中无该sn）
+ trajectoryList = trajectoryList == null ? new ArrayList<>() : trajectoryList;
+ taskList = taskList == null ? new ArrayList<>() : taskList;
+ // 序列化前打印关键日志（确认拿到的数据）
+ log.info("SN:{} save方法原子性获取缓存 - 轨迹数据size：{}，任务数据size：{}",
+ sn, trajectoryList.size(), taskList.size());
+ // 空数据直接返回
if (trajectoryList.isEmpty() && taskList.isEmpty()) {
+ log.warn("SN:{} 原子性获取的缓存数据为空，无需保存！jobId:{}", sn, JobId);
return;
}
try {
- log.info("开始保存轨迹点列表数据！jobId:{},轨迹数量：{},任务状态数量：{}", JobId, trajectoryList.size(), taskList.size());
+ log.info("SN:{} 开始保存轨迹点列表数据！jobId:{},轨迹数量：{},任务状态数量：{}",
+ sn, JobId, trajectoryList.size(), taskList.size());
byte[] blobTrajectory = SerializeUtil.serialize(trajectoryList);
byte[] blobTask = SerializeUtil.serialize(taskList);
+ log.info("SN:{} 序列化结果 - 轨迹字节长度：{}，任务字节长度：{}",
+ sn, blobTrajectory.length, blobTask.length);
Optional<WaylineJobDTO> waylineJobDTO = _waylineJobService.getJobByJobId(JobId);
WaylineTrajectoryEntity TrajectoryEntity = new WaylineTrajectoryEntity();
@@ -181,7 +190,14 @@
}
}
} catch (Exception e) {
- log.error("保存轨迹点列表数据失败！jobId:{}.{}", JobId, e.getMessage());
+ log.error("SN:{} 保存数据异常！jobId:{}, 异常信息：{}", sn, JobId, e.getMessage(), e);
+ // 异常时把数据放回缓存（避免数据丢失）
+ if (!trajectoryList.isEmpty()) {
+ cacheTrajectoryMap.computeIfAbsent(sn, k -> new ArrayList<>()).addAll(trajectoryList);
+ }
+ if (!taskList.isEmpty()) {
+ cacheTaskMap.computeIfAbsent(sn, k -> new ArrayList<>()).addAll(taskList);
+ }
}
}
--
Gitblit v1.9.3