From 3f43e40ed1a1e9f66b3ad2862da9dbc4b710c712 Mon Sep 17 00:00:00 2001
From: nishiwei <13126973358@163.com>
Date: Fri, 23 Oct 2026 14:09:47 +0800
Subject: [PATCH] 智云3.0机场1适配内容提交
---
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java | 84 +++++++++++++++++++++++++++++++++++------
1 files changed, 71 insertions(+), 13 deletions(-)
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
index 36ab8cc..6ad0a20 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
@@ -4,7 +4,6 @@
import com.ard.sample.component.websocket.enums.MessageType;
import com.ard.sample.component.websocket.model.Payload;
import com.ard.sample.component.websocket.model.WebSocketMessage;
-import com.ard.sample.manage.service.IControlService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -19,6 +18,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
/**
@@ -32,9 +32,6 @@
@Autowired
private LiveStreamHandler _liveStreamService;
- @Autowired
- private IControlService _controlService;
-
// ====================== 心跳相关常量（可配置化建议放到配置文件） ======================
/** 心跳间隔（秒）：客户端需每隔此时间发送一次PING消息 */
private static final long HEARTBEAT_INTERVAL = 30;
@@ -42,6 +39,12 @@
private static final long HEARTBEAT_TIMEOUT = 90;
/** 心跳检测任务间隔（秒）：定时扫描超时连接 */
private static final long HEARTBEAT_CHECK_INTERVAL = 10;
+ /**
+ * 停流延迟（秒）：页面刷新或网络抖动会先断开旧连接再建立新连接，
+ * 断开时立即 live_stop 会让新页面需要重新起流，视频出现空档甚至丢失。
+ * 延迟期间重新订阅则取消停止。
+ */
+ private static final long LIVE_STOP_DELAY_SECONDS = 30;
// ====================== 核心存储（线程安全） ======================
/** 用户会话映射：userId -> WebSocketSession */
@@ -52,12 +55,21 @@
private final Map<WebSocketSession, Object> sessionLocks = new ConcurrentHashMap<>();
/** 用户最后心跳时间：userId -> 最后心跳时间戳（毫秒） */
private final Map<String, Long> userLastHeartbeatTime = new ConcurrentHashMap<>();
+ /** 待执行的延迟停流任务：设备SN -> 定时任务 */
+ private final Map<String, ScheduledFuture<?>> pendingLiveStops = new ConcurrentHashMap<>();
// ====================== 线程池 ======================
/** 心跳检测线程池（单线程守护线程） */
private final ScheduledExecutorService heartbeatCheckExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
Thread thread = new Thread(runnable, "websocket-heartbeat-check-thread");
thread.setDaemon(true); // 守护线程，不阻塞JVM退出
+ return thread;
+ });
+
+ /** 延迟停流线程池（单线程守护线程），停流可能走 HTTP，独立线程避免阻塞心跳检测 */
+ private final ScheduledExecutorService liveStopExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable, "websocket-live-stop-thread");
+ thread.setDaemon(true);
return thread;
});
@@ -192,6 +204,10 @@
if (isSubscribe) {
// 订阅操作：不存在则创建，存在则添加用户
subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId);
+ // 页面刷新会先断连再重连，新订阅到达时取消上一次断连触发的延迟停流
+ if (MessageType.LIVESTREAM.getType().equals(type)) {
+ cancelPendingLiveStop(sn);
+ }
// log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type);
// // 直播类型订阅：触发开启直播
@@ -206,12 +222,12 @@
subscriberIds.remove(userId);
log.info("用户[{}]取消订阅设备[{}]的[{}]类型消息", userId, sn, type);
- // 订阅组为空时移除，并停止直播（直播类型）
+ // 订阅组为空时移除，并延迟停止直播（直播类型）。
+ // 不立即停流：页面刷新会先断连再重连，立即停流会导致视频长时间黑屏。
if (subscriberIds.isEmpty()) {
subscriptions.remove(subscriptionKey);
if (MessageType.LIVESTREAM.getType().equals(type)) {
- _liveStreamService.StopLiveStreaming(sn);
- log.info("设备[{}]直播流已停止（无订阅用户）", sn);
+ scheduleStopLiveStream(sn);
}
}
}
@@ -287,10 +303,13 @@
// 线程安全发送消息
try {
- if(MessageType.LIVESTREAM.getType().equals(type)) {
+ // 指定了接收用户时只推给该用户；uId 为空（如后台定时任务起流）时推给全部订阅者
+ if (MessageType.LIVESTREAM.getType().equals(type) && uId != null && !uId.isEmpty()) {
WebSocketSession liveSession = userSessions.get(uId);
- sendMessage(liveSession, content);
- log.info("单独向用户[{}]推送设备[{}]的[{}]类型消息：{}", uId, sn, type, content);
+ if (liveSession != null && liveSession.isOpen()) {
+ sendMessage(liveSession, content);
+ log.info("单独向用户[{}]推送设备[{}]的[{}]类型消息：{}", uId, sn, type, content);
+ }
return;
}
sendMessage(session, content);
@@ -371,17 +390,52 @@
if (subscriberIds.remove(userId)) {
log.debug("用户[{}]已移除订阅：{}", userId, entry.getKey());
- // 订阅组为空时移除，并停止直播（直播类型）
+ // 订阅组为空时移除，并延迟停止直播（直播类型）。
+ // 页面刷新/短暂断网会重建连接，延迟停流可避免视频空档；期间重新订阅会取消停止。
if (subscriberIds.isEmpty()) {
String subscriptionKey = entry.getKey();
if (subscriptionKey.contains(MessageType.LIVESTREAM.getType())) {
String sn = subscriptionKey.split(":")[0];
- _liveStreamService.StopLiveStreaming(sn);
- log.info("设备[{}]无订阅用户，已停止直播流", sn);
+ scheduleStopLiveStream(sn);
}
iterator.remove(); // 安全移除空订阅组
}
}
+ }
+ }
+
+ /**
+ * 延迟停止直播流。
+ * 页面刷新/网络抖动会先断开旧连接再建立新连接，断开时立即停流会让新页面重新起流，
+ * 视频出现空档甚至丢失；这里延迟 LIVE_STOP_DELAY_SECONDS 秒再停，期间有新订阅则取消。
+ */
+ private void scheduleStopLiveStream(String sn) {
+ String subscriptionKey = getSubscriptionKey(sn, MessageType.LIVESTREAM.getType());
+ pendingLiveStops.compute(sn, (key, oldFuture) -> {
+ if (oldFuture != null && !oldFuture.isDone()) {
+ oldFuture.cancel(false);
+ }
+ return liveStopExecutor.schedule(() -> {
+ pendingLiveStops.remove(sn);
+ Set<String> subscriberIds = subscriptions.get(subscriptionKey);
+ // 延迟期间又有用户订阅（如页面刷新重连），不停止直播
+ if (subscriberIds != null && !subscriberIds.isEmpty()) {
+ log.info("设备[{}]延迟停流期间重新订阅，继续直播", sn);
+ return;
+ }
+ _liveStreamService.StopLiveStreaming(sn);
+ log.info("设备[{}]无订阅用户（已延迟{}秒），停止直播流", sn, LIVE_STOP_DELAY_SECONDS);
+ }, LIVE_STOP_DELAY_SECONDS, TimeUnit.SECONDS);
+ });
+ }
+
+ /**
+ * 取消待执行的延迟停流任务（设备重新被订阅时调用）。
+ */
+ private void cancelPendingLiveStop(String sn) {
+ ScheduledFuture<?> future = pendingLiveStops.remove(sn);
+ if (future != null && !future.isDone()) {
+ future.cancel(false);
}
}
@@ -406,6 +460,10 @@
log.error("WebSocket心跳检测线程池关闭被中断", e);
}
}
+ // 关闭延迟停流线程池
+ if (liveStopExecutor != null && !liveStopExecutor.isShutdown()) {
+ liveStopExecutor.shutdownNow();
+ }
log.info("WebSocket处理器资源已销毁");
}
}
\ No newline at end of file
--
Gitblit v1.9.3