1. 修改用户为空时，获取不到无人机遥测数据。

Xiajijian

2026-02-11 1fe0965d93e376fd5ae7392365f2f7e7ce7f9b4f

 ard-work/src/main/java/com/ard/work/component/websocket/handler/BaseSubscriptionManager.java

@@ -16,17 +16,18 @@
    // 存储订阅的用户
    protected final Map<String, Set<Long>> subscriptions = new ConcurrentHashMap<>();
    // 定时任务调度器
    protected final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    protected final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10);
    // 存储推送任务
    protected final Map<String, ScheduledFuture<?>> pushTasks = new ConcurrentHashMap<>();

    /**
     * 订阅指定设备的数据
     */
    public void subscribe(String boxSn, Long userId,Integer period) {
        subscriptions.computeIfAbsent(boxSn, k -> new HashSet<>()).add(userId);
    public void subscribe(String boxSn, Long userId, Integer period) {
        subscriptions.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
        if (subscriptions.get(boxSn).size() == 1) {
            startPushTask(boxSn,period);
            log.debug("首次订阅，启动推送任务");
            startPushTask(boxSn, period);
        }
    }

@@ -34,25 +35,33 @@
     * 取消订阅
     */
    public void unsubscribe(String boxSn, Long userId) {
        Set<Long> userIds = subscriptions.get(boxSn);
        if (userIds != null) {
        subscriptions.compute(boxSn, (key, userIds) -> {
            if (userIds == null) {
                return null;
            }
            userIds.remove(userId);
            if (userIds.isEmpty()) {
                stopPushTask(boxSn);
                log.debug("无人订阅设备 {}，移除订阅记录", boxSn);
                return null; // 直接返回 null 等价于 remove(key)
            }
        }
            return userIds;
        });
    }

    /**
     * 启动推送任务
     */
    private void startPushTask(String boxSn,Integer period) {
    private void startPushTask(String boxSn, Integer period) {
        // 如果已经存在任务，则不再创建
        if (pushTasks.containsKey(boxSn)) {
            return;
        }
        ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
            sendDataToSubscribers(boxSn);
            try {
                sendDataToSubscribers(boxSn);
            } catch (Exception e) {
                log.error("Error in push task for boxSn={}", boxSn, e);
            }
        }, 0, period, TimeUnit.SECONDS);

        pushTasks.put(boxSn, future);
@@ -79,7 +88,6 @@
    }



    /**
     * 向订阅者发送数据
     */
@@ -90,7 +98,7 @@
        if (userIds != null) {
            for (Long userId : userIds) {
                try {
                    log.warn("给用户{}定时推送",userId);
                    log.debug("给用户{}定时推送", userId);
                    WebSocketUtil.sendMessage(userId, JSON.toJSONString(baseMessage));
                } catch (Exception e) {
                    log.error("Error sending message to session {}", userId, e);