增加回放

uavps-admin/src/main/resources/application.yml

@@ -7,7 +7,7 @@ ruoyi:
   # 版权年份
   copyrightYear: 2024
   # 文件路径 示例（ Windows配置D:/ruoyi/uploadPath，Linux配置 /home/ruoyi/uploadPath）
-  profile: D:/ruoyi/uploadPath
+  profile: D:/UavPs/uploadPath
   # 获取ip地址开关
   addressEnabled: false
   # 验证码类型 math 数字计算 char 字符验证

uavps-framework/src/main/java/com/uavps/framework/udp/TaskInfo.java

@@ -1,5 +1,11 @@
 package com.uavps.framework.udp;
 
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
 public enum TaskInfo {
     // 枚举实例，包含初始化坐标和转化比率
     INSTANCE(0.0f, 0.0f, true, 1.0f);
@@ -14,6 +20,10 @@ public enum TaskInfo {
     private String filePath;
     private Long lastTime = 0L;
 
+    private FileWriterThread writerThread;
+
+    // 用于存储待写入的UDP包数据
+    private final BlockingQueue<LogEntry> logQueue = new LinkedBlockingQueue<>();
 
     // 枚举构造函数
     private TaskInfo(float x, float y, boolean initFlag, float conversionRate) {
@@ -32,6 +42,67 @@ public enum TaskInfo {
         this.conversionRate = 0.00001f;
         this.bizId = bizId;
         this.filePath = filePath;
+        logQueue.clear();
+        if(writerThread != null) {
+            writerThread.shutdown();
+            try {
+                writerThread.join(); // 等待写入线程完成
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        writerThread = new FileWriterThread();
+        writerThread.start();
+    }
+
+    // 日志条目类
+    public static class LogEntry {
+        final String data;
+        final long time;
+
+        LogEntry(String data, long time) {
+            this.data = data;
+            this.time = time;
+        }
+    }
+
+    // 文件写入线程
+    private class FileWriterThread extends Thread {
+        private volatile boolean running = true;
+
+        @Override
+        public void run() {
+            File file = new File(filePath);
+
+            // 创建父目录（如果不存在）
+            File parentDir = file.getParentFile();
+            if (!parentDir.exists()) {
+                parentDir.mkdirs(); // 创建所有必要的父目录
+            }
+            try (FileOutputStream fos = new FileOutputStream(TaskInfo.INSTANCE.getFilePath(), true)) {
+                while (running || !logQueue.isEmpty()) {
+                    LogEntry entry = logQueue.take(); // 阻塞直到有数据
+                    // 格式化时间戳(10位数字 + 冒号)
+                    String timestamp = String.format("%010d:", entry.time);
+                    // 先写入时间戳
+                    fos.write(timestamp.getBytes());
+                    // 再写入实际数据
+                    fos.write(entry.data.getBytes());
+                    // 可选: 每条记录后加换行
+                    fos.write('\n');
+                    fos.flush();
+                }
+            } catch (IOException | InterruptedException e) {
+                if (running) { // 仅在非正常关闭时打印错误
+                    System.err.println("File writer error: " + e.getMessage());
+                }
+            }
+        }
+
+        public void shutdown() {
+            running = false;
+            this.interrupt(); // 中断阻塞的take操作
+        }
     }
 
     // Getter和Setter方法
@@ -108,6 +179,18 @@ public enum TaskInfo {
         this.lastTime = lastTime;
     }
 
+    public BlockingQueue<LogEntry> getLogQueue() {
+        return logQueue;
+    }
+
+    public FileWriterThread getWriterThread() {
+        return writerThread;
+    }
+
+    public void setWriterThread(FileWriterThread writerThread) {
+        this.writerThread = writerThread;
+    }
+
     // 显示坐标信息
     public void displayCoordinates() {
         System.out.printf("%s 坐标系 - X: %.2f, Y: %.2f (初始化标识: %s, 转化比率: %.6f)%n",

uavps-framework/src/main/java/com/uavps/framework/udp/UdpServerService.java

@@ -1,29 +1,20 @@
 package com.uavps.framework.udp;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.uavps.common.utils.spring.SpringUtils;
-import com.uavps.framework.FileUtils;
 import com.uavps.framework.config.UdpConfig;
 import com.uavps.framework.udp.utils.UdpDataUtils;
 import com.uavps.framework.utils.UdpUtils;
 import com.uavps.system.service.IUavpsAircraftService;
-import com.uavps.system.udp.vo.Aircraft;
-import com.uavps.system.udp.vo.AircraftFormation;
-import com.uavps.system.udp.vo.UdpData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
 import javax.websocket.Session;
-import java.io.FileOutputStream;
 import java.io.IOException;
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.SocketException;
-import java.util.Date;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.time.Instant;
 
 
 /**
@@ -54,22 +45,6 @@ public class UdpServerService {
         }
     }
 
-    // 用于存储待写入的UDP包数据
-    private static final BlockingQueue<LogEntry> logQueue = new LinkedBlockingQueue<>();
-
-    // 日志条目类
-    private static class LogEntry {
-        final byte[] data;
-        final int length;
-        final long time;
-
-        LogEntry(byte[] data, int length, long time) {
-            this.data = data;
-            this.length = length;
-            this.time = time;
-        }
-    }
-
     public void receive(Session session) {
         try {
             //创建一个服务端对象，注册端口
@@ -78,14 +53,13 @@ public class UdpServerService {
             DatagramPacket packet = new DatagramPacket(data, data.length);
 
             String uavData = "";
-            Date initDate = new Date();
-            String json = "";
+            long initMilli = Instant.now().toEpochMilli();
+
             while (true) {
-                json = "";
                 // 设置初始时间，用于计算每个数据的时间差
                 if(TaskInfo.INSTANCE.getInitFlag()){
-                    initDate = new Date();
-                    log.info("任务Id：{}，任务开始，开始时间是：{}", TaskInfo.INSTANCE.getBizId(), initDate);
+                    initMilli = Instant.now().toEpochMilli();
+                    log.info("任务Id：{}，任务开始，开始时间是：{}", TaskInfo.INSTANCE.getBizId(), initMilli);
                 }
 
                 //使用数据包接收客户端发送的数据
@@ -94,13 +68,14 @@ public class UdpServerService {
                 byte[] remoteData = UdpDataUtils.hexStringToBytes(uavData);
                 // 将日志条目放入队列
                 // 复制数据以避免后续修改影响
-                int length = remoteData.length;
-                byte[] dataCopy = new byte[length];
-                System.arraycopy(remoteData, 0, dataCopy, 0, length);
-                long lastTime = new Date().getTime() - initDate.getTime();
-                logQueue.add(new LogEntry(dataCopy, length, lastTime));
+//                int length = remoteData.length;
+//                byte[] dataCopy = new byte[length];
+//                System.arraycopy(remoteData, 0, dataCopy, 0, length);
+                long lastTime = Instant.now().toEpochMilli() - initMilli;
+                String jsonResult = UdpUtils.getJson(remoteData);
+                TaskInfo.INSTANCE.getLogQueue().add(new TaskInfo.LogEntry(jsonResult, lastTime));
                 TaskInfo.INSTANCE.setLastTime(lastTime);
-                session.getBasicRemote().sendText(UdpUtils.getJson(remoteData));
+                session.getBasicRemote().sendText(jsonResult);
             }
         } catch (IOException e) {
             log.error("UdpServerService.receive error", e);
@@ -108,40 +83,5 @@ public class UdpServerService {
         }
     }
 
-    // 文件写入线程
-    private static class FileWriterThread extends Thread {
-        private volatile boolean running = true;
-
-        @Override
-        public void run() {
-            try (FileOutputStream fos = new FileOutputStream(TaskInfo.INSTANCE.getFilePath(), true)) {
-                while (running || !logQueue.isEmpty()) {
-                    LogEntry entry = logQueue.take(); // 阻塞直到有数据
-
-                    // 格式化时间戳(10位数字 + 冒号)
-                    String timestamp = String.format("%010d:", entry.time);
-
-                    // 先写入时间戳
-                    fos.write(timestamp.getBytes());
 
-                    // 再写入实际数据
-                    fos.write(entry.data, 0, entry.length);
-
-                    // 可选: 每条记录后加换行
-                    fos.write('\n');
-
-                    fos.flush();
-                }
-            } catch (IOException | InterruptedException e) {
-                if (running) { // 仅在非正常关闭时打印错误
-                    System.err.println("File writer error: " + e.getMessage());
-                }
-            }
-        }
-
-        public void shutdown() {
-            running = false;
-            this.interrupt(); // 中断阻塞的take操作
-        }
-    }
 }

uavps-framework/src/main/java/com/uavps/framework/websocket/WebSocketServer.java

@@ -4,14 +4,11 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.uavps.common.config.RuoYiConfig;
 import com.uavps.common.utils.StringUtils;
 import com.uavps.common.utils.spring.SpringUtils;
-import com.uavps.common.utils.uuid.IdUtils;
 import com.uavps.common.utils.uuid.Seq;
 import com.uavps.framework.udp.TaskInfo;
 import com.uavps.framework.udp.UdpClientService;
 import com.uavps.framework.udp.UdpServerService;
 import com.uavps.framework.udp.utils.UdpDataUtils;
-import com.uavps.framework.utils.UdpUtils;
-import com.uavps.system.domain.UavpsAircraft;
 import com.uavps.system.domain.UavpsTask;
 import com.uavps.system.service.IUavpsAircraftService;
 import com.uavps.system.service.IUavpsAlgorithmParameterService;
@@ -19,7 +16,6 @@ import com.uavps.system.service.IUavpsTaskService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
-import org.springframework.util.CollectionUtils;
 
 import javax.websocket.*;
 import javax.websocket.server.ServerEndpoint;
@@ -29,8 +25,8 @@ import java.io.IOException;
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.InetAddress;
+import java.time.Instant;
 import java.util.Date;
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -100,7 +96,7 @@ public class WebSocketServer {
         if (session != null) {
             try {
                 if (message.startsWith("REPLAY:")) {
-                    long baseTime = System.currentTimeMillis();
+                    long baseTime = Instant.now().toEpochMilli();
                     //重放，获取数据库轨迹数据
                     String bizId = message.substring(7);
                     UavpsTask uavpsTask = uavpsTaskService.selectUavpsTaskByBizId(Long.parseLong(bizId));
@@ -111,14 +107,11 @@ public class WebSocketServer {
                             // 解析时间戳和数据
                             if (line.length() > 11 && line.charAt(10) == ':') {
                                     long timestamp = Long.parseLong(line.substring(0, 10));
-                                    byte[] data = line.substring(11).getBytes();
-                                    String json = UdpUtils.getJson(data);
-                                    long currentTime = System.currentTimeMillis();
+                                    String json = line.substring(11);
+                                    long currentTime = Instant.now().toEpochMilli();
                                     long targetTime = baseTime + timestamp;
-
                                     // 计算需要等待的时间
                                     long waitTime = targetTime - currentTime;
-
                                     if (waitTime > 0) {
                                         TimeUnit.MILLISECONDS.sleep(waitTime);
                                     }