forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/manage/service/impl/WaylineTrajectoryServiceImpl.java
@@ -49,7 +49,7 @@ /** * 给数据加个进度 */ private final ConcurrentHashMap<String, String> cacheProgress = new ConcurrentHashMap<>(); // private final ConcurrentHashMap<String, String> cacheProgress = new ConcurrentHashMap<>(); /** * 分片保存阈值条数 */ @@ -65,95 +65,139 @@ public void setList(String sn, String type, String dataJson) { List<String> listTrajectory = cacheTrajectoryMap.computeIfAbsent(sn, k -> new ArrayList<>()); List<String> listTask = cacheTaskMap.computeIfAbsent(sn, k -> new ArrayList<>()); String progress = cacheProgress.computeIfAbsent(sn, k -> "0"); // String progress = cacheProgress.computeIfAbsent(sn, k -> "0"); // 取进度值，就为了判断ok一个任务的结束 Integer pro = Convert.toInt(cacheProgress.get(sn)); // Integer pro = Convert.toInt(cacheProgress.get(sn)); // 取jobid、status字段值 String jobId = ""; String isOK = ""; 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()); } // jobid、status字段赋值 if (!listTask.isEmpty()) { String json = listTask.get(listTask.size() - 1); EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = mapper.convertValue(json, log.info("SN:{} 当前任务记录最后一条数据:{}", sn, 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); log.info("SN:{} 当前任务jobId:{},任务执行进度isOK:{}", sn, jobId, isOK); } catch (Exception e) { log.error("SN:{} 解析任务状态JSON失败: {}", sn, e.getMessage()); log.error("SN:{} JSON内容: {}", sn, json); } } if (listTrajectory.size() >= CHUNK_SIZE || listTask.size() >= CHUNK_SIZE) { progress = Convert.toStr(pro + 1); else { // 当前任务状态数据为空时，清空缓存 log.info("SN:{} 任务状态数据为空，当前无任务清空缓存", sn); // 清空缓存 cacheTrajectoryMap.put(sn, new ArrayList<>()); cacheTaskMap.put(sn, new ArrayList<>()); return; } // listTrajectory.size() >= CHUNK_SIZE || if (listTask.size() >= CHUNK_SIZE) { // 分片保存 save(sn, jobId); log.info("当前任务保存成功，jobId:{},进度：{}",jobId , progress); save(sn, jobId, "0"); log.info("SN:{} 当前任务保存成功，jobId:{}, listTrajectory.size():{}，listTask.size():{}", sn, jobId, listTrajectory.size(), listTask.size()); // log.info("当前任务保存成功，jobId:{},进度：{}",jobId , progress); } else { // 判断任务状态json中的status字段，如果为OK，则保存数据，一条完整的航线数据结束。 try { if (isOK.equals("ok")) { progress = "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); log.info("当前任务保存成功，jobId:{},进度：{}",jobId , progress); save(sn, 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) { List<String> trajectoryList = cacheTrajectoryMap.get(sn); List<String> taskList = cacheTaskMap.get(sn); String progress = cacheProgress.get(sn); public void save(String sn, String JobId, String progress) { // 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("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(); TrajectoryEntity.setJobId(JobId); TrajectoryEntity.setName(waylineJobDTO.get().getJobName()); TrajectoryEntity.setFileId(waylineJobDTO.get().getFileId()); TrajectoryEntity.setFileName(waylineJobDTO.get().getFileName()); TrajectoryEntity.setDockSn(sn); TrajectoryEntity.setProgress(progress); TrajectoryEntity.setCreateTime(System.currentTimeMillis()); TrajectoryEntity.setUpdateTime(System.currentTimeMillis()); TrajectoryEntity.setTrajectoryList(blobTrajectory); TrajectoryEntity.setTaskList(blobTask); // 保存轨迹点列表数据 int i = _waylineTrajectoryMapper.insert(TrajectoryEntity); if (i > 0) { log.info("保存轨迹点列表数据成功！jobId:{}", JobId); // 清空缓存 cacheTrajectoryMap.put(sn, new ArrayList<>()); cacheTaskMap.put(sn, new ArrayList<>()); if (progress.equals("ok")) { cacheProgress.put(sn, "0"); if(waylineJobDTO.isPresent()) { TrajectoryEntity.setJobId(JobId); TrajectoryEntity.setName(waylineJobDTO.get().getJobName()); TrajectoryEntity.setFileId(waylineJobDTO.get().getFileId()); TrajectoryEntity.setFileName(waylineJobDTO.get().getFileName()); TrajectoryEntity.setDockSn(sn); TrajectoryEntity.setProgress(progress); TrajectoryEntity.setCreateTime(System.currentTimeMillis()); TrajectoryEntity.setUpdateTime(System.currentTimeMillis()); TrajectoryEntity.setTrajectoryList(blobTrajectory); TrajectoryEntity.setTaskList(blobTask); // 保存轨迹点列表数据 int i = _waylineTrajectoryMapper.insert(TrajectoryEntity); if (i > 0) { log.info("保存轨迹点列表数据成功！jobId:{}", JobId); // 清空缓存 cacheTrajectoryMap.put(sn, new ArrayList<>()); cacheTaskMap.put(sn, new ArrayList<>()); // if (progress.equals("ok")) { // cacheProgress.put(sn, "0"); // } } } } 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); } } } @@ -189,23 +233,29 @@ return !waylineTrajectoryEntityList.isEmpty(); } public WaylineTrajectoryDTO TrajectoryEntityConvertToDTO(List<WaylineTrajectoryEntity> entity) { if (entity == null) { return null; } WaylineTrajectoryDTO dto = new WaylineTrajectoryDTO(); dto.setId(entity.get(0).getId()); dto.setJobId(entity.get(0).getJobId()); dto.setName(entity.get(0).getName()); dto.setFileId(entity.get(0).getFileId()); dto.setFileName(entity.get(0).getFileName()); dto.setDockSn(entity.get(0).getDockSn()); dto.setProgress(entity.get(0).getProgress()); dto.setCreateTime(entity.get(0).getCreateTime()); dto.setUpdateTime(entity.get(0).getUpdateTime()); for (WaylineTrajectoryEntity e : entity) { dto.getTrajectoryList().addAll(SerializeUtil.deserialize(e.getTrajectoryList(), List.class)); dto.getTaskList().addAll(SerializeUtil.deserialize(e.getTaskList(), List.class)); try { dto.setId(entity.get(0).getId()); dto.setJobId(entity.get(0).getJobId()); dto.setName(entity.get(0).getName()); dto.setFileId(entity.get(0).getFileId()); dto.setFileName(entity.get(0).getFileName()); dto.setDockSn(entity.get(0).getDockSn()); dto.setProgress(entity.get(0).getProgress()); dto.setCreateTime(entity.get(0).getCreateTime()); dto.setUpdateTime(entity.get(0).getUpdateTime()); for (WaylineTrajectoryEntity e : entity) { dto.getTrajectoryList().addAll(SerializeUtil.deserialize(e.getTrajectoryList(), List.class)); dto.getTaskList().addAll(SerializeUtil.deserialize(e.getTaskList(), List.class)); } } catch (Exception e) { log.error("获取任务id轨迹点列表数据失败！{}", e.getMessage()); } return dto;
