| | |
| | | package com.ard.work.service.impl; |
| | | |
| | | import cn.hutool.core.util.SerializeUtil; |
| | | 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 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; |
| | |
| | | /** |
| | | * 缓存:boxSn -> 当前分片数据 |
| | | */ |
| | | private final ConcurrentHashMap<String, List<TelemetryData>> cacheMap = new ConcurrentHashMap<>(); |
| | | private final ConcurrentHashMap<String, List<UavFlightStateDTO>> cacheMap = new ConcurrentHashMap<>(); |
| | | |
| | | /** |
| | | * 分片保存阈值条数 |
| | |
| | | @Scheduled(fixedRate = 1000) |
| | | public void readTelemetryTask() { |
| | | UavTelemetryDataCache.forEach((boxSn, telemetryData) -> { |
| | | addTelemetryData(telemetryData); |
| | | |
| | | // 从状态缓存里取当前状态 |
| | | UavState uavState = UavStateCache.get(boxSn); |
| | | if (uavState == null) { |
| | | return; // 状态不存在,不记录 |
| | | } |
| | | |
| | | UavFlightStateDTO dto = new UavFlightStateDTO( |
| | | uavState, |
| | | telemetryData |
| | | ); |
| | | |
| | | addTelemetryData(dto); |
| | | }); |
| | | } |
| | | |
| | |
| | | * 添加一条遥测数据到缓存 |
| | | */ |
| | | @Override |
| | | public void addTelemetryData(TelemetryData telemetryData) { |
| | | String boxSn = telemetryData.getBoxSn(); |
| | | List<TelemetryData> list = cacheMap.computeIfAbsent(boxSn, k -> new ArrayList<>()); |
| | | list.add(telemetryData); |
| | | 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) { |
| | |
| | | */ |
| | | @Async |
| | | public void saveChunk(String boxSn) { |
| | | List<TelemetryData> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>()); |
| | | List<UavFlightStateDTO> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>()); |
| | | if (dataList.isEmpty()) { |
| | | return; |
| | | } |
| | |
| | | */ |
| | | @Override |
| | | public void flushRemaining(String boxSn) { |
| | | List<TelemetryData> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>()); |
| | | List<UavFlightStateDTO> dataList = cacheMap.getOrDefault(boxSn, new ArrayList<>()); |
| | | if (!dataList.isEmpty()) { |
| | | saveChunk(boxSn); |
| | | } |