forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
@@ -1,11 +1,9 @@ package com.ard.sample.component.websocket.handler; import com.alibaba.fastjson2.JSON; import com.ard.sample.common.model.ResponseResult; 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; @@ -14,13 +12,19 @@ import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; import javax.annotation.PreDestroy; import java.io.IOException; import java.util.*; 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; /** * WebSocket处理器（设备对接） * 核心能力：心跳检测、订阅/取消订阅、定向消息推送、广播消息 */ @Slf4j @Component public class DockWebSocketHandler extends TextWebSocketHandler { @@ -28,71 +32,165 @@ @Autowired private LiveStreamHandler _liveStreamService; @Autowired private IControlService _controlService; // ====================== 心跳相关常量（可配置化建议放到配置文件） ====================== /** 心跳间隔（秒）：客户端需每隔此时间发送一次PING消息 */ private static final long HEARTBEAT_INTERVAL = 30; /** 心跳超时阈值（秒）：超过此时间未收到PING则判定连接失效 */ 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 // ====================== 核心存储（线程安全） ====================== /** 用户会话映射：userId -> WebSocketSession */ private final Map<String, WebSocketSession> userSessions = new ConcurrentHashMap<>(); // 订阅关系：key=订阅标识(sn+type)，value=订阅用户ID集合 /** 订阅关系映射：订阅标识(sn:type) -> 订阅用户ID集合 */ private final Map<String, Set<String>> subscriptions = new ConcurrentHashMap<>(); // 关键：为每个WebSocketSession分配独立的锁对象，确保写操作同步 /** 会话锁映射：WebSocketSession -> 独立锁对象（防止并发写冲突） */ 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 delayExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { Thread thread = new Thread(runnable, "websocket-关闭-延迟-线程"); thread.setDaemon(true); // 设为守护线程，不阻塞JVM退出 // ====================== 线程池 ====================== /** 心跳检测线程池（单线程守护线程） */ 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; }); // ====================== 初始化 ====================== public DockWebSocketHandler() { // 启动心跳检测定时任务：首次延迟CHECK_INTERVAL执行，之后每隔CHECK_INTERVAL执行 heartbeatCheckExecutor.scheduleAtFixedRate( this::checkHeartbeatTimeout, HEARTBEAT_CHECK_INTERVAL, HEARTBEAT_CHECK_INTERVAL, TimeUnit.SECONDS ); log.info("WebSocket心跳检测任务已启动，检测间隔：{}秒，超时阈值：{}秒", HEARTBEAT_CHECK_INTERVAL, HEARTBEAT_TIMEOUT); } // ====================== 连接生命周期 ====================== /** * 连接建立后回调 * 连接建立成功回调 */ @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { // 从会话属性中获取认证后的用户ID // 从会话属性获取认证后的用户ID（需前置拦截器完成认证） String userId = (String) session.getAttributes().get("userId"); if (userId == null) { session.close(CloseStatus.POLICY_VIOLATION.withReason("未认证")); session.close(CloseStatus.POLICY_VIOLATION.withReason("未认证，拒绝连接")); log.warn("WebSocket连接建立失败：会话无用户认证信息"); return; } // 初始化会话相关存储 userSessions.put(userId, session); // 为新会话初始化锁对象 sessionLocks.put(session, new Object()); //System.out.println("用户[" + userId + "]已连接，当前在线人数：" + userSessions.size()); log.info("用户[{}]已连接，当前在线人数：{}", userId, userSessions.size()); userLastHeartbeatTime.put(userId, System.currentTimeMillis()); log.info("用户[{}]WebSocket连接建立成功，当前在线人数：{}", userId, userSessions.size()); } /** * 处理客户端发送的消息 * 连接关闭回调（同步清理，无延时） */ @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { String userId = (String) session.getAttributes().get("userId"); if (userId != null) { // 同步清理所有关联数据 userSessions.remove(userId); sessionLocks.remove(session); userLastHeartbeatTime.remove(userId); removeUserAllSubscriptions(userId); log.info("用户[{}]WebSocket连接关闭（状态码：{}，原因：{}），当前在线人数：{}", userId, status.getCode(), status.getReason(), userSessions.size()); } else { // 无认证用户直接清理锁 sessionLocks.remove(session); log.warn("无认证用户WebSocket连接关闭，状态码：{}", status.getCode()); } } // ====================== 消息处理 ====================== /** * 处理客户端文本消息（包含心跳、订阅/取消订阅） */ @Override protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { String userId = (String) session.getAttributes().get("userId"); String jsonMessage = message.getPayload(); if (userId == null) { sendMessage(session, "未认证，无法处理消息"); return; } String jsonPayload = message.getPayload(); try { // 解析客户端消息 WebSocketMessage webSocketMessage = JSON.parseObject(jsonMessage, WebSocketMessage.class); String type = webSocketMessage.getType(); // 解析WebSocket消息体 WebSocketMessage webSocketMessage = JSON.parseObject(jsonPayload, WebSocketMessage.class); String messageType = webSocketMessage.getType(); Payload payload = webSocketMessage.getPayload(); // 验证消息类型是否合法 //if (!Arrays.asList("status", "control", "progress", "state").contains(type)) { if(!MessageType.fromString(type)){ session.sendMessage(new TextMessage("不支持的消息类型：" + type)); // 1. 处理心跳PING消息 if (MessageType.PING.getType().equals(messageType)) { handleHeartbeatPing(session, userId); return; } // 处理订阅/取消订阅 handleSubscription(userId, type, payload); // 2. 验证消息类型合法性 if (!MessageType.fromString(messageType)) { sendMessage(session, "不支持的消息类型：" + messageType); log.warn("用户[{}]发送非法消息类型：{}", userId, messageType); return; } // 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) { System.err.println("消息处理错误：" + e.getMessage()); session.sendMessage(new TextMessage("消息格式错误")); log.error("用户[{}]消息处理异常", userId, e); sendMessage(session, "消息格式错误或处理失败：" + e.getMessage()); } } /** * 处理心跳PING消息 */ private void handleHeartbeatPing(WebSocketSession session, String userId) throws IOException { // 更新最后心跳时间 userLastHeartbeatTime.put(userId, System.currentTimeMillis()); // 回复PONG消息 WebSocketMessage pongMessage = new WebSocketMessage(); pongMessage.setType(MessageType.PONG.getType()); pongMessage.setPayload(new Payload()); sendMessage(session, JSON.toJSONString(pongMessage)); log.debug("用户[{}]心跳PING接收成功，已回复PONG", userId); } /** @@ -100,145 +198,272 @@ */ private void handleSubscription(String userId, String type, Payload payload) { String sn = payload.getSn(); boolean subscribe = payload.isSubscribe(); String subscriptionKey = getSubscriptionKey(sn, type); // 生成唯一订阅标识 boolean isSubscribe = payload.isSubscribe(); String subscriptionKey = getSubscriptionKey(sn, type); if (subscribe) { // 订阅操作 if (isSubscribe) { // 订阅操作：不存在则创建，存在则添加用户 subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId); //System.out.println("用户[" + userId + "]订阅了 设备[" + sn + "]的[" + type + "]类型消息"); log.info("用户[{}]订阅了 设备[{}]的[{}]类型消息", userId, sn, type); if(MessageType.LIVESTREAM.getType().equals(type)){ // 开启直播 // 开启直播逻辑 System.out.println("开启直播"); _liveStreamService.StartLiveStreaming(sn); // 页面刷新会先断连再重连，新订阅到达时取消上一次断连触发的延迟停流 if (MessageType.LIVESTREAM.getType().equals(type)) { cancelPendingLiveStop(sn); } } else { // 停止直播--前端不主动关注直播，则在断开连接时停止 // if(MessageType.LIVESTREAM.getType().equals(type)){ // System.out.println("停止直播"); // // _liveStreamService.StopLiveStreaming(sn); // log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type); // // 直播类型订阅：触发开启直播 // if (MessageType.LIVESTREAM.getType().equals(type)) { // _liveStreamService.StartLiveStreaming(sn); // log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId); // } } else { // 取消订阅操作 if (subscriptions.containsKey(subscriptionKey)) { subscriptions.get(subscriptionKey).remove(userId); if (subscriptions.get(subscriptionKey).isEmpty()) { Set<String> subscriberIds = subscriptions.get(subscriptionKey); subscriberIds.remove(userId); log.info("用户[{}]取消订阅设备[{}]的[{}]类型消息", userId, sn, type); // 订阅组为空时移除，并延迟停止直播（直播类型）。 // 不立即停流：页面刷新会先断连再重连，立即停流会导致视频长时间黑屏。 if (subscriberIds.isEmpty()) { subscriptions.remove(subscriptionKey); if (MessageType.LIVESTREAM.getType().equals(type)) { scheduleStopLiveStream(sn); } } //System.out.println("用户[" + userId + "]取消订阅了 设备[" + sn + "]的[" + type + "]类型消息"); log.info("用户[{}]取消订阅了 设备[{}]的[{}]类型消息", userId, sn, type); } } } // ====================== 心跳检测 ====================== /** * 连接关闭后回调 * 检测心跳超时连接并主动关闭 */ @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { String userId = (String) session.getAttributes().get("userId"); if (userId != null) { log.info("用户[{}]连接关闭，将延时5秒执行清理操作（给前端刷新缓冲）", userId); private void checkHeartbeatTimeout() { long currentTime = System.currentTimeMillis(); // 遍历所有用户心跳记录，检查超时 for (Map.Entry<String, Long> entry : userLastHeartbeatTime.entrySet()) { String userId = entry.getKey(); long lastHeartbeatTime = entry.getValue(); long timeoutMs = currentTime - lastHeartbeatTime; // 延时5秒执行清理逻辑 delayExecutor.schedule(() -> { // 判定超时：超过阈值则主动关闭连接 if (timeoutMs > HEARTBEAT_TIMEOUT * 1000) { WebSocketSession session = userSessions.get(userId); log.warn("用户[{}]心跳超时（{}秒未发送PING），将主动关闭连接", userId, timeoutMs / 1000); try { // 主动关闭连接 if (session != null && session.isOpen()) { session.close(CloseStatus.SESSION_NOT_RELIABLE.withReason( String.format("心跳超时：%d秒未收到PING", timeoutMs / 1000) )); } } catch (IOException e) { log.error("主动关闭用户[{}]超时连接失败", userId, e); } finally { // 强制清理用户所有关联数据 userSessions.remove(userId); sessionLocks.remove(session); // 移除会话锁 // 关闭远程调试模式,暂时写死sn，后续改为动态获取sn // ResponseResult responseResult = _controlService.controlDock("7CTDLCE00A1UPE","debug_mode_close",null); // log.info("关闭远程调试模式"); // if(responseResult.getCode() == ResponseResult.CODE_SUCCESS){ // log.info("关闭远程调试模式成功"); // } // 清除该用户的所有订阅 sessionLocks.remove(session); userLastHeartbeatTime.remove(userId); removeUserAllSubscriptions(userId); //System.out.println("用户[" + userId + "]已断开连接，当前在线人数：" + userSessions.size()); // log.info("用户[{}]已断开连接，当前在线人数：{}", userId, userSessions.size()); log.info("用户[{}]延时5秒清理完成，当前在线人数：{}", userId, userSessions.size()); } catch (Exception e) { log.error("用户[{}]延时清理操作异常", userId, e); } }, 5, TimeUnit.SECONDS); // 延时5秒执行 } else { // 无认证用户，直接清理会话锁（无需延时） sessionLocks.remove(session); log.info("无认证用户连接关闭，直接清理会话锁"); } } /** * 清除用户的所有订阅 */ private void removeUserAllSubscriptions(String userId) { for (Map.Entry<String, Set<String>> entry : subscriptions.entrySet()) { // 停止直播逻辑 entry.getValue().remove(userId); if (entry.getValue().isEmpty()) { if (entry.getKey().contains(MessageType.LIVESTREAM.getType())) { String sn = entry.getKey().split(":")[0]; _liveStreamService.StopLiveStreaming(sn); } subscriptions.remove(entry.getKey()); } } } // ====================== 消息推送接口 ====================== /** * 向订阅了特定设备特定类型的用户发送消息 * 向订阅指定设备指定类型的用户推送消息 * * @param sn 设备序列号 * @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, new HashSet<>()); Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet()); if (subscriberIds.isEmpty()) { //System.out.println("没有用户订阅 设备[" + sn + "]的[" + type + "]类型消息"); log.debug("设备[{}]的[{}]类型消息无订阅用户，无需推送", sn, type); return; } try { // 向所有订阅者发送消息 for (String userId : subscriberIds) { WebSocketSession session = userSessions.get(userId); // 1. 检查会话是否有效 if (session == null || !session.isOpen()) { // 移除无效会话 userSessions.remove(userId); sessionLocks.remove(session); continue; } // 2. 获取当前会话的锁对象，确保同步写操作 Object lock = sessionLocks.get(session); if (lock == null) { continue; // 理论上不会出现，防止NPE } // 3. 同步发送消息，避免并发冲突 synchronized (lock) { session.sendMessage(new TextMessage(content)); System.out.println("向用户[" + userId + "]发送 设备[" + sn + "]的[" + type + "]类型消息:" + content); } // if (session != null && session.isOpen()) { // session.sendMessage(new TextMessage(content)); // System.out.println("向用户[" + userId + "]发送 设备[" + sn + "]的[" + type + "]类型消息"); // System.out.println(content); // } // 遍历订阅用户推送消息 for (String userId : subscriberIds) { WebSocketSession session = userSessions.get(userId); if (session == null || !session.isOpen()) { // 清理无效会话 userSessions.remove(userId); sessionLocks.remove(session); userLastHeartbeatTime.remove(userId); log.warn("用户[{}]会话无效，已清理", userId); continue; } } catch (IOException e) { // System.err.println("发送消息错误：" + e.getMessage()); log.error("发送消息错误：{}", e.getMessage()); // 线程安全发送消息 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) { log.error("向用户[{}]推送消息失败", userId, e); } } } /** * 生成订阅标识（sn+type的组合） * 新增：向所有在线客户端广播消息 * * @param content 消息内容（JSON字符串） */ public void broadcastMessageToAllClients(String content) { if (userSessions.isEmpty()) { log.debug("无在线客户端，无需广播消息"); return; } // log.info("开始向所有在线客户端（{}人）广播消息", userSessions.size()); // 遍历所有在线会话发送消息 for (Map.Entry<String, WebSocketSession> entry : userSessions.entrySet()) { String userId = entry.getKey(); WebSocketSession session = entry.getValue(); if (session == null || !session.isOpen()) { // 清理无效会话 userSessions.remove(userId); sessionLocks.remove(session); userLastHeartbeatTime.remove(userId); log.warn("用户[{}]会话无效，已清理", userId); continue; } // 线程安全发送广播消息 try { sendMessage(session, content); log.debug("向用户[{}]广播消息成功", userId); } catch (IOException e) { log.error("向用户[{}]广播消息失败", userId, e); } } } // ====================== 通用工具方法 ====================== /** * 线程安全发送消息（封装同步逻辑） */ private void sendMessage(WebSocketSession session, String content) throws IOException { Object lock = sessionLocks.get(session); if (lock != null && session.isOpen()) { synchronized (lock) { session.sendMessage(new TextMessage(content)); } } } /** * 生成订阅标识（设备SN + 消息类型） */ private String getSubscriptionKey(String sn, String type) { return sn + ":" + type; return String.format("%s:%s", sn, type); } /** * 清除用户所有订阅关系 */ private void removeUserAllSubscriptions(String userId) { // 迭代器遍历，避免并发修改异常 Iterator<Map.Entry<String, Set<String>>> iterator = subscriptions.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<String, Set<String>> entry = iterator.next(); Set<String> subscriberIds = entry.getValue(); // 移除当前用户 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]; 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); } } // ====================== 资源销毁 ====================== /** * 容器销毁时关闭线程池，防止内存泄漏 */ @PreDestroy public void destroy() { // 关闭心跳检测线程池 if (heartbeatCheckExecutor != null && !heartbeatCheckExecutor.isShutdown()) { heartbeatCheckExecutor.shutdown(); try { // 等待1秒优雅关闭，超时则强制关闭 if (!heartbeatCheckExecutor.awaitTermination(1, TimeUnit.SECONDS)) { heartbeatCheckExecutor.shutdownNow(); log.warn("WebSocket心跳检测线程池强制关闭"); } } catch (InterruptedException e) { heartbeatCheckExecutor.shutdownNow(); Thread.currentThread().interrupt(); // 恢复中断状态 log.error("WebSocket心跳检测线程池关闭被中断", e); } } // 关闭延迟停流线程池 if (liveStopExecutor != null && !liveStopExecutor.isShutdown()) { liveStopExecutor.shutdownNow(); } log.info("WebSocket处理器资源已销毁"); } }
