package com.ard.sample.component.websocket.handler;
|
|
import com.alibaba.fastjson2.JSON;
|
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 lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Component;
|
import org.springframework.web.socket.CloseStatus;
|
import org.springframework.web.socket.TextMessage;
|
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.TimeUnit;
|
|
/**
|
* WebSocket处理器（设备对接）
|
* 核心能力：心跳检测、订阅/取消订阅、定向消息推送、广播消息
|
*/
|
@Slf4j
|
@Component
|
public class DockWebSocketHandler extends TextWebSocketHandler {
|
|
@Autowired
|
private LiveStreamHandler _liveStreamService;
|
|
// ====================== 心跳相关常量（可配置化建议放到配置文件） ======================
|
/** 心跳间隔（秒）：客户端需每隔此时间发送一次PING消息 */
|
private static final long HEARTBEAT_INTERVAL = 30;
|
/** 心跳超时阈值（秒）：超过此时间未收到PING则判定连接失效 */
|
private static final long HEARTBEAT_TIMEOUT = 90;
|
/** 心跳检测任务间隔（秒）：定时扫描超时连接 */
|
private static final long HEARTBEAT_CHECK_INTERVAL = 10;
|
|
// ====================== 核心存储（线程安全） ======================
|
/** 用户会话映射：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<>();
|
|
// ====================== 线程池 ======================
|
/** 心跳检测线程池（单线程守护线程） */
|
private final ScheduledExecutorService heartbeatCheckExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
|
Thread thread = new Thread(runnable, "websocket-heartbeat-check-thread");
|
thread.setDaemon(true); // 守护线程，不阻塞JVM退出
|
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（需前置拦截器完成认证）
|
String userId = (String) session.getAttributes().get("userId");
|
if (userId == null) {
|
session.close(CloseStatus.POLICY_VIOLATION.withReason("未认证，拒绝连接"));
|
log.warn("WebSocket连接建立失败：会话无用户认证信息");
|
return;
|
}
|
|
// 初始化会话相关存储
|
userSessions.put(userId, session);
|
sessionLocks.put(session, new Object());
|
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");
|
if (userId == null) {
|
sendMessage(session, "未认证，无法处理消息");
|
return;
|
}
|
|
String jsonPayload = message.getPayload();
|
try {
|
// 解析WebSocket消息体
|
WebSocketMessage webSocketMessage = JSON.parseObject(jsonPayload, WebSocketMessage.class);
|
String messageType = webSocketMessage.getType();
|
Payload payload = webSocketMessage.getPayload();
|
|
// 1. 处理心跳PING消息
|
if (MessageType.PING.getType().equals(messageType)) {
|
handleHeartbeatPing(session, userId);
|
return;
|
}
|
|
// 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) {
|
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);
|
}
|
|
/**
|
* 处理订阅/取消订阅逻辑
|
*/
|
private void handleSubscription(String userId, String type, Payload payload) {
|
String sn = payload.getSn();
|
boolean isSubscribe = payload.isSubscribe();
|
String subscriptionKey = getSubscriptionKey(sn, type);
|
|
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);
|
// }
|
} else {
|
// 取消订阅操作
|
if (subscriptions.containsKey(subscriptionKey)) {
|
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)) {
|
_liveStreamService.StopLiveStreaming(sn);
|
log.info("设备[{}]直播流已停止（无订阅用户）", sn);
|
}
|
}
|
}
|
}
|
}
|
|
// ====================== 心跳检测 ======================
|
/**
|
* 检测心跳超时连接并主动关闭
|
*/
|
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;
|
|
// 判定超时：超过阈值则主动关闭连接
|
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, String uId) {
|
String subscriptionKey = getSubscriptionKey(sn, type);
|
Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet());
|
|
if (subscriberIds.isEmpty()) {
|
log.debug("设备[{}]的[{}]类型消息无订阅用户，无需推送", sn, type);
|
return;
|
}
|
|
// 遍历订阅用户推送消息
|
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;
|
}
|
|
// 线程安全发送消息
|
try {
|
if(MessageType.LIVESTREAM.getType().equals(type)) {
|
WebSocketSession liveSession = userSessions.get(uId);
|
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);
|
}
|
}
|
}
|
|
/**
|
* 新增：向所有在线客户端广播消息
|
*
|
* @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 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];
|
_liveStreamService.StopLiveStreaming(sn);
|
log.info("设备[{}]无订阅用户，已停止直播流", sn);
|
}
|
iterator.remove(); // 安全移除空订阅组
|
}
|
}
|
}
|
}
|
|
// ====================== 资源销毁 ======================
|
/**
|
* 容器销毁时关闭线程池，防止内存泄漏
|
*/
|
@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);
|
}
|
}
|
log.info("WebSocket处理器资源已销毁");
|
}
|
}
|
