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 | 482 +++++++++++++++++++++++++++++++++++++++++------------
1 files changed, 374 insertions(+), 108 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 89ccdd4..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
@@ -5,7 +5,6 @@
import com.ard.sample.component.websocket.model.Payload;
import com.ard.sample.component.websocket.model.WebSocketMessage;
import lombok.extern.slf4j.Slf4j;
-import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
@@ -13,10 +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 {
@@ -24,61 +32,165 @@
@Autowired
private LiveStreamHandler _liveStreamService;
- // 存储用户会话：userId -> WebSocketSession
- private final Map<String, WebSocketSession> userSessions = new ConcurrentHashMap<>();
-
- // 订阅关系：key=订阅标识(sn+type)，value=订阅用户ID集合
- private final Map<String, Set<String>> subscriptions = new ConcurrentHashMap<>();
-
- // 关键：为每个WebSocketSession分配独立的锁对象，确保写操作同步
- private final Map<WebSocketSession, Object> sessionLocks = new ConcurrentHashMap<>();
-
+ // ====================== 心跳相关常量（可配置化建议放到配置文件） ======================
+ /** 心跳间隔（秒）：客户端需每隔此时间发送一次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 */
+ private final Map<String, WebSocketSession> userSessions = new ConcurrentHashMap<>();
+ /** 订阅关系映射：订阅标识(sn:type) -> 订阅用户ID集合 */
+ private final Map<String, Set<String>> subscriptions = new ConcurrentHashMap<>();
+ /** 会话锁映射：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 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);
}
/**
@@ -86,118 +198,272 @@
*/
private void handleSubscription(String userId, String type, Payload payload) {
String sn = payload.getSn();
- boolean subscribe = payload.isSubscribe();
- String videoType = payload.getVideotype();
- 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, videoType);
+ // 页面刷新会先断连再重连，新订阅到达时取消上一次断连触发的延迟停流
+ if (MessageType.LIVESTREAM.getType().equals(type)) {
+ cancelPendingLiveStop(sn);
}
+// log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type);
+
+// // 直播类型订阅：触发开启直播
+// if (MessageType.LIVESTREAM.getType().equals(type)) {
+// _liveStreamService.StartLiveStreaming(sn);
+// log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
+// }
} else {
- // 停止直播
- if(MessageType.LIVESTREAM.getType().equals(type)){
- System.out.println("停止直播");
- _liveStreamService.StopLiveStreaming(sn);
- }
// 取消订阅操作
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) {
- userSessions.remove(userId);
- sessionLocks.remove(session); // 移除会话锁
- // 清除该用户的所有订阅
- removeUserAllSubscriptions(userId);
- //System.out.println("用户[" + userId + "]已断开连接，当前在线人数：" + userSessions.size());
- log.info("用户[{}]已断开连接，当前在线人数：{}", userId, userSessions.size());
- }
- }
+ 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;
- /**
- * 清除用户的所有订阅
- */
- private void removeUserAllSubscriptions(String userId) {
- for (Map.Entry<String, Set<String>> entry : subscriptions.entrySet()) {
- entry.getValue().remove(userId);
- if (entry.getValue().isEmpty()) {
- subscriptions.remove(entry.getKey());
+ // 判定超时：超过阈值则主动关闭连接
+ 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);
+ userLastHeartbeatTime.remove(userId);
+ removeUserAllSubscriptions(userId);
+ }
}
}
}
+ // ====================== 消息推送接口 ======================
/**
- * 向订阅了特定设备特定类型的用户发送消息
+ * 向订阅指定设备指定类型的用户推送消息
+ *
+ * @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处理器资源已销毁");
}
}
\ No newline at end of file
--
Gitblit v1.9.3