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("轨迹点数据添加到列表缓存，当前列表数量：{}", listTrajectory.size());
|
}
|
if (type.equals("1")) {
|
// 任务状态数据
|
listTask.add(dataJson);
|
log.info("任务状态数据添加到列表缓存，当前列表数量：{}", listTask.size());
|
}
|
// jobid、status字段赋值
|
if (!listTask.isEmpty()) {
|
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);
|
}
|
}
|
if (listTrajectory.size() >= CHUNK_SIZE || listTask.size() >= CHUNK_SIZE) {
|
progress = Convert.toStr(pro + 1);
|
// 分片保存
|
save(sn, jobId);
|
log.info("当前任务保存成功，jobId:{},进度：{}",jobId , progress);
|
} else {
|
// 判断任务状态json中的status字段，如果为OK，则保存数据，一条完整的航线数据结束。
|
try {
|
if (isOK.equals("ok")) {
|
progress = "ok";
|
if (getWaylineTrajectoryByProgress(jobId, progress)) {
|
log.info("当前任务已完成，不再保存数据！jobId:{},progress:{}", jobId, progress);
|
return;
|
}
|
save(sn, jobId);
|
log.info("当前任务保存成功，jobId:{},进度：{}",jobId , progress);
|
}
|
} catch (Exception e) {
|
log.error("json解析异常！{}", 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);
|
|
if (trajectoryList.isEmpty() && taskList.isEmpty()) {
|
return;
|
}
|
try {
|
byte[] blobTrajectory = SerializeUtil.serialize(trajectoryList);
|
byte[] blobTask = SerializeUtil.serialize(taskList);
|
|
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");
|
}
|
}
|
|
} catch (Exception e) {
|
log.error("保存轨迹点列表数据失败！jobId:{}.{}", JobId, e.getMessage());
|
}
|
}
|
|
/**
|
* 返回航线轨迹完整的数据，拼接的轨迹点列表和任务状态数据。
|
*
|
* @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();
|
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));
|
}
|
|
return dto;
|
}
|
}
|
