package com.ard.work.service.impl;
|
|
import cn.hutool.core.util.SerializeUtil;
|
import com.ard.common.utils.DateUtils;
|
import com.ard.work.domain.cache.UavStateCache;
|
import com.ard.work.domain.cache.UavTelemetryDataCache;
|
import com.ard.work.domain.dto.UavFlightStateDTO;
|
import com.ard.work.domain.dto.state.UavState;
|
import com.ard.work.domain.dto.telemetry.TelemetryData;
|
import com.ard.work.domain.entity.ArdFlightTaskPart;
|
import com.ard.work.mapper.ArdFlightTaskPartMapper;
|
import com.ard.work.service.IArdFlightTaskPartService;
|
import lombok.RequiredArgsConstructor;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.scheduling.annotation.Async;
|
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.stereotype.Service;
|
|
import java.util.ArrayList;
|
import java.util.Comparator;
|
import java.util.Date;
|
import java.util.List;
|
import java.util.concurrent.ConcurrentHashMap;
|
|
/**
|
* 轨迹分片处理业务
|
*/
|
@Slf4j
|
@Service
|
@RequiredArgsConstructor
|
public class ArdFlightTaskPartServiceImpl implements IArdFlightTaskPartService {
|
|
private final ArdFlightTaskPartMapper ardflightTaskPartMapper;
|
// 上一次飞行状态缓存
|
private final ConcurrentHashMap<String, Integer> lastFlightStatusMap = new ConcurrentHashMap<>();
|
/**
|
* 缓存：boxSn -> 当前分片数据
|
*/
|
private final ConcurrentHashMap<String, List<UavFlightStateDTO>> cacheMap = new ConcurrentHashMap<>();
|
|
/**
|
* 分片保存阈值条数
|
*/
|
private static final int CHUNK_SIZE = 300;
|
|
/**
|
* 定时任务：每秒读取一次设备状态数据
|
*/
|
@Scheduled(fixedRate = 1000)
|
public void readTelemetryTask() {
|
log.debug("定时任务：读取遥测数据");
|
try {
|
long now = System.currentTimeMillis();
|
UavTelemetryDataCache.forEach((boxSn, telemetryData) -> {
|
// 1️⃣ 从状态缓存里取当前状态
|
UavState uavState = UavStateCache.get(boxSn);
|
if (uavState == null) {
|
log.warn("状态不存在，不记录");
|
return; // 状态不存在，不记录
|
}
|
|
// 2️⃣ 使用遥测自身时间戳判断是否还活着
|
long dataTs = telemetryData.getTimestamp(); // 设备时间 / 服务器接收时间
|
if (now - dataTs > 3000) {
|
//log.warn("超过 3 秒没有新遥测 → 认为已离线或冻结，不记录：{}", new Date(dataTs));
|
return; // 超过 3 秒没有新遥测 → 认为已离线或冻结，不记录
|
}
|
|
int currentStatus = uavState.getFlightControllerState().getFlightStatus();
|
Integer lastStatus = lastFlightStatusMap.get(boxSn);
|
|
// 仅在“飞行中 → 非飞行中”状态变化时 flush
|
if (lastStatus != null && lastStatus == 2 && currentStatus != 2) {
|
flushRemaining(boxSn);
|
}
|
lastFlightStatusMap.put(boxSn, currentStatus);
|
log.debug("设备 {} 当前状态为：{}", boxSn, currentStatus);
|
// 只有飞行中才添加缓存
|
if (currentStatus == 2) {
|
log.warn("设备 {} 当前状态为飞行中，添加遥测数据到缓存", boxSn);
|
UavFlightStateDTO dto = new UavFlightStateDTO(uavState, telemetryData, dataTs);
|
addTelemetryData(dto);
|
}
|
});
|
}
|
catch (Exception e) {
|
log.error("定时任务：读取遥测数据异常", e);
|
}
|
}
|
|
/**
|
* 添加一条遥测数据到缓存
|
*/
|
@Override
|
public void addTelemetryData(UavFlightStateDTO uavFlightStateDTO) {
|
String boxSn = uavFlightStateDTO.getTelemetryData().getBoxSn();
|
List<UavFlightStateDTO> list = cacheMap.computeIfAbsent(boxSn, k -> new ArrayList<>());
|
list.add(uavFlightStateDTO);
|
|
// 当缓存条数达到阈值 或 时间间隔超过 CHUNK_INTERVAL 时，分片保存
|
if (list.size() >= CHUNK_SIZE) {
|
saveChunk(boxSn); // taskId 可根据你的业务传入
|
}
|
}
|
|
/**
|
* 分片保存缓存数据到数据库
|
*/
|
@Async
|
public void saveChunk(String boxSn) {
|
List<UavFlightStateDTO> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>());
|
if (dataList.isEmpty()) {
|
return;
|
}
|
|
try {
|
byte[] blobData = SerializeUtil.serialize(dataList);
|
// 取第一个和最后一个状态时间
|
long startTs = dataList.get(0).getTimestamp();
|
long endTs = dataList.get(dataList.size() - 1).getTimestamp();
|
ArdFlightTaskPart part = ArdFlightTaskPart.builder()
|
.boxSn(boxSn)
|
.count(dataList.size())
|
.dataPart(blobData)
|
.startTime(new Date(startTs))
|
.endTime(new Date(endTs))
|
.build();
|
part.setCreateTime(new Date());
|
ardflightTaskPartMapper.insert(part);
|
|
log.info("设备 {} 分片保存 {} 条飞行数据到数据库成功", boxSn, dataList.size());
|
|
// 清空缓存，更新上次保存时间
|
cacheMap.put(boxSn, new ArrayList<>());
|
|
} catch (Exception e) {
|
log.error("设备 {} 分片保存飞行数据失败", boxSn, e);
|
}
|
}
|
|
|
/**
|
* 将剩余缓存分片保存
|
*/
|
@Override
|
public void flushRemaining(String boxSn) {
|
List<UavFlightStateDTO> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>());
|
if (!dataList.isEmpty()) {
|
saveChunk(boxSn);
|
}
|
}
|
|
/**
|
* 获取完整航线轨迹（合并所有分片）
|
*/
|
@Override
|
public List<UavFlightStateDTO> getCompleteFlightData(ArdFlightTaskPart ardFlightTaskPart) {
|
List<UavFlightStateDTO> allData = new ArrayList<>();
|
String boxSn = ardFlightTaskPart.getBoxSn();
|
List<ArdFlightTaskPart> parts = ardflightTaskPartMapper.selectByTimeRange(ardFlightTaskPart);
|
for (ArdFlightTaskPart part : parts) {
|
try {
|
List<UavFlightStateDTO> partData = SerializeUtil.deserialize(part.getDataPart());
|
allData.addAll(partData);
|
} catch (Exception e) {
|
log.error("反序列化 boxSn={} 分片数据失败", boxSn, e);
|
}
|
}
|
// 按 telemetryData.timestamp 升序排序
|
allData.sort(Comparator.comparingLong(dto -> dto.getTelemetryData().getTimestamp()));
|
|
return allData;
|
}
|
}
|
