forked from uav/ard-uav-server
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; }); @@ -153,6 +165,13 @@ // 3. 处理订阅/取消订阅逻辑 handleSubscription(userId, messageType, payload); // 直播类型订阅：触发开启直播 if (MessageType.LIVESTREAM.getType().equals(messageType) && payload.isSubscribe()) { String sn = payload.getSn(); _liveStreamService.StartLiveStreaming(sn,userId); log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId); } } catch (Exception e) { log.error("用户[{}]消息处理异常", userId, e); sendMessage(session, "消息格式错误或处理失败：" + e.getMessage()); @@ -185,13 +204,17 @@ if (isSubscribe) { // 订阅操作：不存在则创建，存在则添加用户 subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId); log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type); // 直播类型订阅：触发开启直播 // 页面刷新会先断连再重连，新订阅到达时取消上一次断连触发的延迟停流 if (MessageType.LIVESTREAM.getType().equals(type)) { _liveStreamService.StartLiveStreaming(sn); log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId); cancelPendingLiveStop(sn); } // log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type); // // 直播类型订阅：触发开启直播 // if (MessageType.LIVESTREAM.getType().equals(type)) { // _liveStreamService.StartLiveStreaming(sn); // log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId); // } } else { // 取消订阅操作 if (subscriptions.containsKey(subscriptionKey)) { @@ -199,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); } } } @@ -257,7 +280,7 @@ * @param type 消息类型（对应MessageType） * @param content 消息内容（JSON字符串） */ public void sendMessageToSubscribers(String sn, String type, String content) { public void sendMessageToSubscribers(String sn, String type, String content, String uId) { String subscriptionKey = getSubscriptionKey(sn, type); Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet()); @@ -280,6 +303,15 @@ // 线程安全发送消息 try { // 指定了接收用户时只推给该用户；uId 为空（如后台定时任务起流）时推给全部订阅者 if (MessageType.LIVESTREAM.getType().equals(type) && uId != null && !uId.isEmpty()) { WebSocketSession liveSession = userSessions.get(uId); if (liveSession != null && liveSession.isOpen()) { sendMessage(liveSession, content); log.info("单独向用户[{}]推送设备[{}]的[{}]类型消息：{}", uId, sn, type, content); } return; } sendMessage(session, content); log.debug("向用户[{}]推送设备[{}]的[{}]类型消息：{}", userId, sn, type, content); } catch (IOException e) { @@ -299,7 +331,7 @@ return; } log.info("开始向所有在线客户端（{}人）广播消息", userSessions.size()); // log.info("开始向所有在线客户端（{}人）广播消息", userSessions.size()); // 遍历所有在线会话发送消息 for (Map.Entry<String, WebSocketSession> entry : userSessions.entrySet()) { String userId = entry.getKey(); @@ -358,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); } } @@ -393,6 +460,10 @@ log.error("WebSocket心跳检测线程池关闭被中断", e); } } // 关闭延迟停流线程池 if (liveStopExecutor != null && !liveStopExecutor.isShutdown()) { liveStopExecutor.shutdownNow(); } log.info("WebSocket处理器资源已销毁"); } }
