From a737ae8e1253ce4883a3c5c91f10f83ad3ac8a52 Mon Sep 17 00:00:00 2001
From: zhaowenyu <35589726@qq.com>
Date: Tue, 02 Dec 2025 16:29:30 +0800
Subject: [PATCH] 2025-12-02 zwy websocket增加心跳
---
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java | 226 ++++++++++++++++++++++++++++++++++++++++++++-----------
1 files changed, 179 insertions(+), 47 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..d8f4a74 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
@@ -1,11 +1,12 @@
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.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
@@ -16,6 +17,9 @@
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.TimeUnit;
@Slf4j
@Component
@@ -24,14 +28,48 @@
@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 CHECK_INTERVAL = 10;
+
+ // ====================== 核心存储 ======================
// 存储用户会话：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<>();
+ // 存储用户最后心跳时间：userId -> 最后心跳时间戳（毫秒）
+ private final Map<String, Long> userLastHeartbeatTime = new ConcurrentHashMap<>();
+
+ // ====================== 线程池 ======================
+ // 延时任务线程池（单线程，守护线程），用于处理连接关闭后的延时清理
+ private final ScheduledExecutorService delayExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable, "websocket-关闭-延迟-线程");
+ thread.setDaemon(true); // 设为守护线程，不阻塞JVM退出
+ return thread;
+ });
+
+ // 心跳检测线程池（单线程，守护线程），定时检查超时连接
+ private final ScheduledExecutorService heartbeatCheckExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable, "websocket-心跳检测-线程");
+ thread.setDaemon(true);
+ return thread;
+ });
+
+ // ====================== 初始化心跳检测任务 ======================
+ public DockWebSocketHandler() {
+ // 项目启动后，每隔 CHECK_INTERVAL 秒执行一次心跳检测
+ heartbeatCheckExecutor.scheduleAtFixedRate(this::checkHeartbeatTimeout,
+ CHECK_INTERVAL, CHECK_INTERVAL, TimeUnit.SECONDS);
+ }
/**
* 连接建立后回调
@@ -48,12 +86,14 @@
userSessions.put(userId, session);
// 为新会话初始化锁对象
sessionLocks.put(session, new Object());
- //System.out.println("用户[" + userId + "]已连接，当前在线人数：" + userSessions.size());
+ // 初始化用户最后心跳时间（连接建立即视为一次心跳）
+ userLastHeartbeatTime.put(userId, System.currentTimeMillis());
+
log.info("用户[{}]已连接，当前在线人数：{}", userId, userSessions.size());
}
/**
- * 处理客户端发送的消息
+ * 处理客户端发送的消息（包含心跳消息）
*/
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
@@ -66,18 +106,32 @@
String type = webSocketMessage.getType();
Payload payload = webSocketMessage.getPayload();
+ // ====================== 处理心跳消息 ======================
+ if (MessageType.PING.getType().equals(type)) {
+ // 1. 更新用户最后心跳时间
+ userLastHeartbeatTime.put(userId, System.currentTimeMillis());
+ // 2. 回复pong消息
+ WebSocketMessage pongMessage = new WebSocketMessage();
+ pongMessage.setType(MessageType.PONG.getType());
+ pongMessage.setPayload(new Payload()); // 空payload即可
+ sendMessage(session, JSON.toJSONString(pongMessage));
+ log.debug("用户[{}]心跳检测成功，已回复pong", userId);
+ System.out.println("用户["+userId+"]心跳检测成功，已回复pong");
+ return; // 心跳消息无需走后续订阅逻辑
+ }
+
// 验证消息类型是否合法
- //if (!Arrays.asList("status", "control", "progress", "state").contains(type)) {
- if(!MessageType.fromString(type)){
- session.sendMessage(new TextMessage("不支持的消息类型：" + type));
+ if (!MessageType.fromString(type)) {
+ sendMessage(session, "不支持的消息类型：" + type);
return;
}
+
// 处理订阅/取消订阅
handleSubscription(userId, type, payload);
} catch (Exception e) {
- System.err.println("消息处理错误：" + e.getMessage());
- session.sendMessage(new TextMessage("消息格式错误"));
+ log.error("用户[{}]消息处理错误", userId, e);
+ sendMessage(session, "消息格式错误");
}
}
@@ -87,33 +141,29 @@
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); // 生成唯一订阅标识
if (subscribe) {
// 订阅操作
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)){
+ if (MessageType.LIVESTREAM.getType().equals(type)) {
// 开启直播
- // 开启直播逻辑
- System.out.println("开启直播");
- _liveStreamService.StartLiveStreaming(sn, videoType);
+ log.info("开启设备[{}]直播", sn);
+ _liveStreamService.StartLiveStreaming(sn);
}
} 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()) {
subscriptions.remove(subscriptionKey);
+ // 若取消的是直播订阅且无其他订阅者，停止直播
+ if (MessageType.LIVESTREAM.getType().equals(type)) {
+ log.info("设备[{}]无订阅者，停止直播", sn);
+ _liveStreamService.StopLiveStreaming(sn);
+ }
}
- //System.out.println("用户[" + userId + "]取消订阅了 设备[" + sn + "]的[" + type + "]类型消息");
log.info("用户[{}]取消订阅了 设备[{}]的[{}]类型消息", userId, sn, type);
}
}
@@ -126,12 +176,61 @@
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());
+ log.info("用户[{}]连接关闭（状态码：{}），将延时5秒执行清理操作", userId, status.getCode());
+
+ // 延时5秒执行清理逻辑
+ delayExecutor.schedule(() -> {
+ try {
+ // 清理核心存储
+ userSessions.remove(userId);
+ sessionLocks.remove(session);
+ userLastHeartbeatTime.remove(userId); // 清理心跳时间
+
+ // 清除该用户的所有订阅
+ removeUserAllSubscriptions(userId);
+
+ log.info("用户[{}]延时5秒清理完成，当前在线人数：{}", userId, userSessions.size());
+ } catch (Exception e) {
+ log.error("用户[{}]延时清理操作异常", userId, e);
+ }
+ }, 5, TimeUnit.SECONDS); // 延时5秒执行
+ } else {
+ // 无认证用户，直接清理
+ sessionLocks.remove(session);
+ log.info("无认证用户连接关闭，直接清理会话锁");
+ }
+ }
+
+ /**
+ * 心跳超时检测：断开超过HEARTBEAT_TIMEOUT未发心跳的用户
+ */
+ private void checkHeartbeatTimeout() {
+ long currentTime = System.currentTimeMillis();
+ // 遍历所有在线用户，检查心跳超时
+ for (Map.Entry<String, Long> entry : userLastHeartbeatTime.entrySet()) {
+ String userId = entry.getKey();
+ Long lastHeartbeat = entry.getValue();
+ WebSocketSession session = userSessions.get(userId);
+
+ // 1. 计算超时时间差
+ long timeoutDiff = currentTime - lastHeartbeat;
+ if (timeoutDiff > HEARTBEAT_TIMEOUT * 1000) {
+ log.warn("用户[{}]心跳超时（{}秒未发ping），将断开连接", userId, timeoutDiff / 1000);
+ try {
+ // 2. 主动关闭连接
+ if (session != null && session.isOpen()) {
+ session.close(CloseStatus.SESSION_NOT_RELIABLE.withReason("心跳超时"));
+ }
+ } catch (IOException e) {
+ log.error("关闭用户[{}]超时连接失败", userId, e);
+ } finally {
+ // 3. 清理该用户的所有数据
+ userSessions.remove(userId);
+ sessionLocks.remove(session);
+ userLastHeartbeatTime.remove(userId);
+ removeUserAllSubscriptions(userId);
+ }
+ }
}
}
@@ -139,10 +238,23 @@
* 清除用户的所有订阅
*/
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());
+ // 遍历所有订阅，移除该用户并检查是否需要停止直播
+ Iterator<Map.Entry<String, Set<String>>> iterator = subscriptions.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<String, Set<String>> entry = iterator.next();
+ Set<String> userIds = entry.getValue();
+ if (userIds.contains(userId)) {
+ userIds.remove(userId);
+ // 若订阅组为空，移除并停止直播（如果是直播类型）
+ if (userIds.isEmpty()) {
+ String subscriptionKey = entry.getKey();
+ if (subscriptionKey.contains(MessageType.LIVESTREAM.getType())) {
+ String sn = subscriptionKey.split(":")[0];
+ _liveStreamService.StopLiveStreaming(sn);
+ log.info("设备[{}]无订阅者，停止直播", sn);
+ }
+ iterator.remove(); // 安全移除空订阅组
+ }
}
}
}
@@ -155,42 +267,43 @@
Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, new HashSet<>());
if (subscriberIds.isEmpty()) {
- //System.out.println("没有用户订阅 设备[" + 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);
+ userLastHeartbeatTime.remove(userId); // 清理心跳时间
continue;
}
- // 2. 获取当前会话的锁对象，确保同步写操作
+ // 同步发送消息，避免并发冲突
Object lock = sessionLocks.get(session);
if (lock == null) {
- continue; // 理论上不会出现，防止NPE
+ continue;
}
-
- // 3. 同步发送消息，避免并发冲突
synchronized (lock) {
session.sendMessage(new TextMessage(content));
- System.out.println("向用户[" + userId + "]发送 设备[" + sn + "]的[" + type + "]类型消息:" + content);
+ log.debug("向用户[{}]发送 设备[{}]的[{}]类型消息:{}", userId, sn, type, content);
}
-// if (session != null && session.isOpen()) {
-// session.sendMessage(new TextMessage(content));
-// System.out.println("向用户[" + userId + "]发送 设备[" + sn + "]的[" + type + "]类型消息");
-// System.out.println(content);
-// }
}
} catch (IOException e) {
-// System.err.println("发送消息错误：" + e.getMessage());
log.error("发送消息错误：{}", e.getMessage());
+ }
+ }
+
+ /**
+ * 通用发送消息方法（封装同步逻辑）
+ */
+ 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));
+ }
}
}
@@ -200,4 +313,23 @@
private String getSubscriptionKey(String sn, String type) {
return sn + ":" + type;
}
+
+ // ====================== 销毁线程池（可选，防止内存泄漏） ======================
+
+ public void destroy() {
+ delayExecutor.shutdown();
+ heartbeatCheckExecutor.shutdown();
+ try {
+ if (!delayExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+ delayExecutor.shutdownNow();
+ }
+ if (!heartbeatCheckExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+ heartbeatCheckExecutor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ delayExecutor.shutdownNow();
+ heartbeatCheckExecutor.shutdownNow();
+ }
+ log.info("WebSocket线程池已销毁");
+ }
}
\ No newline at end of file
--
Gitblit v1.9.3