package com.xiaopeng.pushMessage.webSocket.service; import io.netty.channel.Channel; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import java.util.*; import java.util.concurrent.ConcurrentHashMap; /** * 基于 AppUser 的 WebSocket 连接管理器 * 使用 Spring WebSocketSession(共用 Spring Boot HTTP 端口) * 支持按用户 ID 管理连接,一个用户可以有多个设备连接 */ @Component @Slf4j public class WebSocketUserChannelConnectionManager { /** * Key: 用户ID (appUserId) * Value: 该用户的所有 WebSocketSession 集合 */ private static final ConcurrentHashMap> userConnections = new ConcurrentHashMap<>(); /** * Key: session.getId() * Value: 用户ID (appUserId) */ private static final ConcurrentHashMap sessionToUser = new ConcurrentHashMap<>(); /** * Key: session.getId() * Value: WebSocketSession */ private static final ConcurrentHashMap allConnections = new ConcurrentHashMap<>(); // ==================== 连接管理 ==================== /** * 添加用户连接 */ public void addUserConnection(String appUserId, WebSocketSession session) { String sessionId = session.getId(); userConnections.computeIfAbsent(appUserId, k -> ConcurrentHashMap.newKeySet()).add(session); sessionToUser.put(sessionId, appUserId); allConnections.put(sessionId, session); log.info("用户连接已添加,用户ID: {}, sessionId: {}, 该用户连接数: {}", appUserId, sessionId, userConnections.get(appUserId).size()); } /** * 移除用户连接 */ public void removeUserConnection(WebSocketSession session) { String sessionId = session.getId(); String appUserId = sessionToUser.get(sessionId); if (appUserId != null) { Set userSessions = userConnections.get(appUserId); if (userSessions != null) { userSessions.remove(session); if (userSessions.isEmpty()) { userConnections.remove(appUserId); } } sessionToUser.remove(sessionId); log.info("用户连接已移除,用户ID: {}, sessionId: {}, 该用户剩余连接数: {}", appUserId, sessionId, userSessions != null ? userSessions.size() : 0); } allConnections.remove(sessionId); } // ==================== 消息发送 ==================== /** * 向指定用户的所有设备发送消息 */ public int sendMessageToUser(String appUserId, String message) { Set sessions = userConnections.get(appUserId); if (sessions == null || sessions.isEmpty()) { log.warn("用户 {} 没有活跃连接", appUserId); return 0; } int successCount = 0; for (WebSocketSession session : sessions) { if (session != null && session.isOpen()) { try { synchronized (session) { session.sendMessage(new TextMessage(message)); } successCount++; } catch (Exception e) { log.error("向用户 {} 发送消息失败,sessionId: {}, 错误: {}", appUserId, session.getId(), e.getMessage()); } } } log.info("向用户 {} 发送消息完成,成功发送到 {} 个设备", appUserId, successCount); return successCount; } /** * 向指定 session 发送消息 */ public boolean sendMessageToChannel(String sessionId, String message) { WebSocketSession session = allConnections.get(sessionId); if (session != null && session.isOpen()) { try { synchronized (session) { session.sendMessage(new TextMessage(message)); } log.info("消息已发送到连接: {}", sessionId); return true; } catch (Exception e) { log.error("发送消息失败,sessionId: {}, 错误: {}", sessionId, e.getMessage()); return false; } } else { log.warn("连接不存在或已关闭,sessionId: {}", sessionId); return false; } } /** * 向所有连接广播消息 */ public int broadcastMessage(String message) { int successCount = 0; for (WebSocketSession session : allConnections.values()) { if (session != null && session.isOpen()) { try { synchronized (session) { session.sendMessage(new TextMessage(message)); } successCount++; } catch (Exception e) { log.error("广播消息失败,sessionId: {}, 错误: {}", session.getId(), e.getMessage()); } } } log.info("广播消息完成,成功发送到 {} 个连接", successCount); return successCount; } /** * 向多个用户发送消息 */ public int sendMessageToUsers(List appUserIds, String message) { int successUserCount = 0; for (String appUserId : appUserIds) { int deviceCount = sendMessageToUser(appUserId, message); if (deviceCount > 0) { successUserCount++; } } log.info("向 {} 个用户发送消息完成,成功发送到 {} 个用户", appUserIds.size(), successUserCount); return successUserCount; } // ==================== 查询 ==================== public Set getUserConnections(String appUserId) { return userConnections.getOrDefault(appUserId, Collections.emptySet()); } public int getUserConnectionCount(String appUserId) { Set sessions = userConnections.get(appUserId); return sessions != null ? sessions.size() : 0; } public int getTotalConnectionCount() { return allConnections.size(); } public int getOnlineUserCount() { return userConnections.size(); } public Set getOnlineUserIds() { return userConnections.keySet(); } public boolean isUserOnline(String appUserId) { Set sessions = userConnections.get(appUserId); if (sessions == null || sessions.isEmpty()) { return false; } return sessions.stream().anyMatch(WebSocketSession::isOpen); } public boolean isConnectionExists(String sessionId) { WebSocketSession session = allConnections.get(sessionId); return session != null && session.isOpen(); } /** * 根据 sessionId 获取用户ID */ public String getUserIdBySessionId(String sessionId) { return sessionToUser.get(sessionId); } public Map getConnectionStats() { Map stats = new HashMap<>(); stats.put("totalConnections", getTotalConnectionCount()); stats.put("onlineUsers", getOnlineUserCount()); stats.put("onlineUserIds", getOnlineUserIds()); Map userConnectionCounts = new HashMap<>(); for (Map.Entry> entry : userConnections.entrySet()) { userConnectionCounts.put(entry.getKey(), entry.getValue().size()); } stats.put("userConnectionCounts", userConnectionCounts); return stats; } }