package com.ard.work.service.impl;
|
|
import cn.hutool.core.util.SerializeUtil;
|
import com.ard.work.domain.cache.UavTelemetryDataCache;
|
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.time.LocalDateTime;
|
import java.time.temporal.ChronoUnit;
|
import java.util.ArrayList;
|
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;
|
|
/**
|
* 缓存：boxSn -> 当前分片数据
|
*/
|
private final ConcurrentHashMap<String, List<TelemetryData>> cacheMap = new ConcurrentHashMap<>();
|
|
/**
|
* 分片保存阈值条数
|
*/
|
private static final int CHUNK_SIZE = 300;
|
|
/**
|
* 定时任务：每秒读取一次设备状态数据
|
* 注意：这里模拟读取状态，实际应该调用设备接口或缓存
|
*/
|
@Scheduled(fixedRate = 1000)
|
public void readTelemetryTask() {
|
UavTelemetryDataCache.forEach((boxSn, telemetryData) -> {
|
addTelemetryData(telemetryData);
|
});
|
}
|
|
/**
|
* 添加一条遥测数据到缓存
|
*/
|
@Override
|
public void addTelemetryData(TelemetryData telemetryData) {
|
String boxSn = telemetryData.getBoxSn();
|
List<TelemetryData> list = cacheMap.computeIfAbsent(boxSn, k -> new ArrayList<>());
|
list.add(telemetryData);
|
|
// 当缓存条数达到阈值 或 时间间隔超过 CHUNK_INTERVAL 时，分片保存
|
if (list.size() >= CHUNK_SIZE) {
|
saveChunk(boxSn); // taskId 可根据你的业务传入
|
}
|
}
|
|
/**
|
* 分片保存缓存数据到数据库
|
*/
|
@Async
|
public void saveChunk(String boxSn) {
|
List<TelemetryData> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>());
|
if (dataList.isEmpty()) {
|
return;
|
}
|
|
try {
|
byte[] blobData = SerializeUtil.serialize(dataList);
|
|
ArdFlightTaskPart part = ArdFlightTaskPart.builder()
|
.boxSn(boxSn)
|
.count(dataList.size())
|
.dataPart(blobData)
|
.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<TelemetryData> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>());
|
if (!dataList.isEmpty()) {
|
saveChunk(boxSn);
|
}
|
}
|
|
/**
|
* 获取完整航线轨迹（合并所有分片）
|
*/
|
@Override
|
public List<TelemetryData> getCompleteFlightData(ArdFlightTaskPart ardFlightTaskPart) {
|
List<TelemetryData> allData = new ArrayList<>();
|
String boxSn = ardFlightTaskPart.getBoxSn();
|
List<ArdFlightTaskPart> parts = ardflightTaskPartMapper.selectByTaskStartTime(ardFlightTaskPart);
|
for (ArdFlightTaskPart part : parts) {
|
try {
|
List<TelemetryData> partData = SerializeUtil.deserialize(part.getDataPart());
|
allData.addAll(partData);
|
} catch (Exception e) {
|
log.error("反序列化 boxSn={} 分片数据失败", boxSn, e);
|
}
|
}
|
return allData;
|
}
|
}
|
