forked from uav/ard-uav-server
blame | history | patch | commit | commitdiff | ignore whitespace
liusuyi
2025-12-24 49c807f8c2e80cc012ed755acbe1308c5de0c65c 
 ard-work/src/main/java/com/ard/work/service/impl/ArdFlightTaskPartServiceImpl.java


@@ -1,7 +1,10 @@


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;


@@ -11,9 +14,8 @@


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.Comparator;


import java.util.Date;


import java.util.List;


import java.util.concurrent.ConcurrentHashMap;


@@ -30,7 +32,7 @@


    /**


     * 缓存：boxSn -> 当前分片数据


     */


    private final ConcurrentHashMap<String, List<TelemetryData>> cacheMap = new ConcurrentHashMap<>();


    private final ConcurrentHashMap<String, List<UavFlightStateDTO>> cacheMap = new ConcurrentHashMap<>();




    /**


     * 分片保存阈值条数


@@ -43,12 +45,33 @@


     */


    @Scheduled(fixedRate = 1000)


    public void readTelemetryTask() {


        long now = System.currentTimeMillis();


        UavTelemetryDataCache.forEach((boxSn, telemetryData) -> {


            // 只记录相对高度大于1米的数据


            float relativeHeight = telemetryData.getUltrasonic(); // 或者 telemetryData.getRelativeHeight()，根据你的字段


            if (relativeHeight > 1) {


                addTelemetryData(telemetryData);




            // 1️⃣ 从状态缓存里取当前状态


            UavState uavState = UavStateCache.get(boxSn);


            if (uavState == null) {


                return; // 状态不存在，不记录


            }




            // 2️⃣ 使用遥测自身时间戳判断是否还活着


            long dataTs = telemetryData.getTimestamp(); // 设备时间 / 服务器接收时间


            if (now - dataTs > 3000) {


                return; // 超过 3 秒没有新遥测 → 认为已离线或冻结，不记录


            }




            // 3️⃣ 飞行状态必须是飞行中


            if(uavState.getFlightControllerState().getFlightStatus()!=2)


            {


                return; // 飞行状态不是飞行中，不记录


            }




            UavFlightStateDTO dto = new UavFlightStateDTO(


                    uavState,


                    telemetryData,


                    dataTs


            );


            addTelemetryData(dto);


        });


    }




@@ -56,10 +79,10 @@


     * 添加一条遥测数据到缓存


     */


    @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) {


@@ -72,7 +95,7 @@


     */


    @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;


        }


@@ -85,7 +108,10 @@


                    .count(dataList.size())


                    .dataPart(blobData)


                    .build();


            part.setCreateTime(new Date());


            // 使用 list 中第一条数据的时间作为 createTime


            long firstTimeMillis = dataList.get(0).getTimestamp();


            Date createTime = new Date(firstTimeMillis);


            part.setCreateTime(createTime);


            ardflightTaskPartMapper.insert(part);




            log.info("设备 {} 分片保存 {} 条飞行数据到数据库成功", boxSn, dataList.size());


@@ -104,7 +130,7 @@


     */


    @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);


        }


@@ -114,18 +140,21 @@


     * 获取完整航线轨迹（合并所有分片）


     */


    @Override


    public List<TelemetryData> getCompleteFlightData(ArdFlightTaskPart ardFlightTaskPart) {


        List<TelemetryData> allData = new ArrayList<>();


    public List<UavFlightStateDTO> getCompleteFlightData(ArdFlightTaskPart ardFlightTaskPart) {


        List<UavFlightStateDTO> 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());


                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;


    }


}