package com.ard.sample.manage.service.impl;
|
|
import cn.hutool.core.util.SerializeUtil;
|
import com.ard.common.core.text.Convert;
|
import com.ard.sample.component.mqtt.model.EventsReceiver;
|
import com.ard.sample.manage.mapper.IWaylineTrajectoryMapper;
|
import com.ard.sample.manage.model.dto.WaylineTrajectoryDTO;
|
import com.ard.sample.manage.model.entity.WaylineTrajectoryEntity;
|
import com.ard.sample.manage.service.IWaylineTrajectoryService;
|
import com.ard.sample.wayline.model.dto.WaylineJobDTO;
|
import com.ard.sample.wayline.model.dto.WaylineTaskProgressReceiver;
|
import com.ard.sample.wayline.service.IWaylineJobService;
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
import com.fasterxml.jackson.core.type.TypeReference;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.scheduling.annotation.Async;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.Resource;
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.Optional;
|
import java.util.concurrent.ConcurrentHashMap;
|
|
@Service
|
@Slf4j
|
public class WaylineTrajectoryServiceImpl implements IWaylineTrajectoryService {
|
|
@Resource
|
private IWaylineTrajectoryMapper _waylineTrajectoryMapper;
|
|
@Autowired
|
private IWaylineJobService _waylineJobService;
|
|
@Autowired
|
protected ObjectMapper mapper;
|
/**
|
* 轨迹点列表缓存
|
*/
|
private final ConcurrentHashMap<String, List<String>> cacheTrajectoryMap = new ConcurrentHashMap<>();
|
|
/**
|
* 任务状态数据缓存
|
*/
|
private final ConcurrentHashMap<String, List<String>> cacheTaskMap = new ConcurrentHashMap<>();
|
|
/**
|
* 给数据加个进度
|
*/
|
// private final ConcurrentHashMap<String, String> cacheProgress = new ConcurrentHashMap<>();
|
/**
|
* 分片保存阈值条数
|
*/
|
private static final int CHUNK_SIZE = 300;
|
|
/**
|
* 增加list数据，type为：0时，表示轨迹点列表；type为：1时，表示任务状态数据。
|
* @param sn 设备序列号
|
* @param type 类型
|
* @param dataJson json数据
|
*/
|
@Override
|
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");
|
// 取进度值，就为了判断ok一个任务的结束
|
// Integer pro = Convert.toInt(cacheProgress.get(sn));
|
// 取jobid、status字段值
|
String jobId = "";
|
String isOK = "";
|
if (type.equals("0")) {
|
// 飞行轨迹数据
|
listTrajectory.add(dataJson);
|
log.info("SN:{} 轨迹点数据添加到列表缓存，当前列表数量：{}", sn, listTrajectory.size());
|
// log.info("SN:{} 轨迹点数据：{}", sn, dataJson);
|
}
|
if (type.equals("1")) {
|
// 任务状态数据
|
listTask.add(dataJson);
|
log.info("SN:{} 任务状态数据添加到列表缓存，当前列表数量：{}", sn, listTask.size());
|
}
|
// jobid、status字段赋值
|
if (!listTask.isEmpty()) {
|
String json = listTask.get(listTask.size() - 1);
|
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("SN:{} 当前任务jobId:{},任务执行进度isOK:{}", sn, jobId, isOK);
|
|
} catch (Exception e) {
|
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 ||
|
if (listTask.size() >= CHUNK_SIZE) {
|
// 分片保存
|
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")) {
|
String progress = "ok";
|
if (getWaylineTrajectoryByProgress(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("SN:{} 当前任务保存成功，jobId:{},进度：{}", sn,jobId , progress);
|
}
|
} catch (Exception e) {
|
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);
|
// 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();
|
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("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);
|
}
|
}
|
}
|
|
/**
|
* 返回航线轨迹完整的数据，拼接的轨迹点列表和任务状态数据。
|
*
|
* @param jobId 任务ID
|
* @return 航线轨迹完整的数据
|
*/
|
@Override
|
public WaylineTrajectoryDTO getWayLineTrajectorybyJobID(String jobId) {
|
|
List<WaylineTrajectoryEntity> waylineTrajectoryEntityList = new ArrayList<>(_waylineTrajectoryMapper.selectList(new LambdaQueryWrapper<WaylineTrajectoryEntity>()
|
.eq(jobId != null, WaylineTrajectoryEntity::getJobId, jobId)
|
.orderByAsc(WaylineTrajectoryEntity::getId)));
|
|
return TrajectoryEntityConvertToDTO(waylineTrajectoryEntityList);
|
}
|
|
/**
|
* 根据作业ID和进度获取航线轨迹
|
*
|
* @param jobId 作业ID，可为null
|
* @param progress 进度值，可为null
|
* @return 当存在符合条件的航线轨迹时返回true，否则返回false
|
*/
|
public Boolean getWaylineTrajectoryByProgress(String jobId, String progress) {
|
List<WaylineTrajectoryEntity> waylineTrajectoryEntityList = new ArrayList<>(_waylineTrajectoryMapper.selectList(new LambdaQueryWrapper<WaylineTrajectoryEntity>()
|
.eq(jobId != null, WaylineTrajectoryEntity::getJobId, jobId)
|
.eq(progress != null, WaylineTrajectoryEntity::getProgress, progress)
|
.orderByAsc(WaylineTrajectoryEntity::getId)));
|
|
return !waylineTrajectoryEntityList.isEmpty();
|
}
|
|
|
public WaylineTrajectoryDTO TrajectoryEntityConvertToDTO(List<WaylineTrajectoryEntity> entity) {
|
if (entity == null) {
|
return null;
|
}
|
WaylineTrajectoryDTO dto = new WaylineTrajectoryDTO();
|
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;
|
}
|
}
|
