1. 项目概述：从零构建一个健壮的WebSocket推送服务
最近在重构一个后台管理系统的实时通知模块，需求很简单：当管理员在后台操作后，前端页面能立刻收到一条消息提示。听起来像是WebSocket的典型应用场景，对吧？但真做起来，你会发现远不止在SpringBoot里加个@ServerEndpoint注解那么简单。消息怎么确保送到？连接断了怎么办？用户成千上万时，如何高效地给特定人群发消息？这些问题，都是把一个“玩具级”的WebSocket服务升级为“生产级”推送服务必须跨过的坎。
这个项目，就是一次完整的实战记录。我们不只关注如何用SpringBoot和WebSocket建立连接，更要深入解决校验、心跳、分组这些核心的工程问题。我会带你从零开始，搭建一个具备用户身份校验、自动心跳保活、支持用户分组广播的WebSocket服务。过程中，我会分享那些官方文档里不会写的坑，比如为什么你的心跳（PING-PONG）机制总是不生效，如何优雅地处理用户上下线，以及在大规模连接下如何做简单的性能优化。无论你是想为你的应用增加实时能力，还是正在被WebSocket的各种不稳定所困扰，这篇内容应该都能给你提供一套可直接复用的解决方案。
2. WebSocket核心机制与SpringBoot集成选型
在动手写代码之前，我们得先搞清楚WebSocket到底是什么，以及为什么在众多实时通信方案中我们选择了它。这决定了我们后续所有技术决策的底层逻辑。
2.1 WebSocket与HTTP、SSE的对比
很多人会把WebSocket和HTTP长轮询、Server-Sent Events (SSE)搞混。简单来说，HTTP请求就像你每次都要打电话问客服“有新消息吗？”，问完就挂断。长轮询是电话不挂，等客服有消息了再告诉你，然后挂断，你再打下一个。这种方式开销大，延迟高。
SSE是服务器向浏览器单向推送数据的技术，它基于HTTP，浏览器通过一个持久的连接监听服务器发来的事件流。它的优点是协议简单，天然支持断线重连。但缺点是单向（只能服务器推给浏览器），并且在一些老式浏览器上支持不佳。
而WebSocket，则是在HTTP握手成功后，建立了一个全双工的TCP长连接。就像你和客服之间拉了一条专线，双方随时可以主动说话，没有请求-响应的概念。这对于需要频繁双向通信的场景（如聊天、实时协作、游戏）是最高效的。我们的消息推送虽然主要是服务端推，但客户端的心跳确认（PONG）和可能的业务ACK，也需要这个双向通道。
2.2 为什么是SpringBoot + 原生WebSocket API？
SpringBoot集成WebSocket主要有两种方式：一是使用Spring提供的WebSocketHandler抽象和STOMP子协议；二是直接使用JSR-356定义的javax.websocket标准API（即@ServerEndpoint注解）。
STOMP在Spring生态中很强大，它相当于在WebSocket之上定义了一套消息格式和路由规则，非常适合复杂的消息代理场景，比如结合RabbitMQ或Kafka。但它也带来了额外的复杂性和学习成本。
对于我们这个相对纯粹的消息推送服务——核心是连接管理、心跳和分组广播——STOMP显得有些“重”了。直接使用@ServerEndpoint让我们能更精细地控制每一个连接的生命周期，实现自定义的心跳、校验逻辑，代码也更直观。因此，我选择了原生API方案，并通过Spring的ServerEndpointExporter来暴露它，这样既能享受Spring的依赖注入，又能保持底层控制的灵活性。
2.3 项目基础环境搭建
首先，创建一个标准的SpringBoot项目（这里使用SpringBoot 2.7.x， 3.x版本在配置上略有不同但核心逻辑一致）。在pom.xml中，我们只需要引入WebSocket的starter依赖。
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency>接下来，我们需要一个核心配置类来启用WebSocket支持。这里的关键是ServerEndpointExporterBean，它负责将所有带有@ServerEndpoint注解的类注册为WebSocket端点。
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.server.standard.ServerEndpointExporter; @Configuration public class WebSocketConfig { /** * 这个Bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint。 * 如果部署在外部容器（如Tomcat）中，容器会自己提供，这个Bean可以省略。 */ @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }注意：如果你将项目打包成WAR包部署到独立的Tomcat等Servlet容器，容器自身会扫描和注册
@ServerEndpoint，此时再定义ServerEndpointExporter会导致端点被注册两次，从而引发错误。这种情况下，你应该移除这个Bean。
3. 实现核心WebSocket端点与连接管理
有了基础框架，我们来构建最核心的WebSocket服务器端点。这个类将处理所有连接的生命周期事件：建立、关闭、错误以及消息收发。
3.1 定义WebSocket端点类
我们创建一个PushWebSocketEndpoint类。使用@Component和@ServerEndpoint注解将其声明为一个端点。@ServerEndpoint的value属性定义了客户端连接的URI路径。
这里有一个非常重要的点：WebSocket端点的每个连接都会创建一个新的端点实例。这意味着你不能在类成员变量中直接保存连接状态（如Session），因为它们是实例级别的。我们必须使用静态的ConcurrentHashMap来在全局管理所有连接。
import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; @Component @ServerEndpoint("/ws/push/{userId}") // 在路径中携带用户ID，用于初步标识 public class PushWebSocketEndpoint { // 静态变量，用来记录当前在线连接数。注意它是线程安全的。 private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0); // 用来存放每个客户端对应的WebSocketSession对象，Key为用户ID private static final ConcurrentHashMap<String, Session> SESSION_POOL = new ConcurrentHashMap<>(); // 与某个客户端的连接会话，需要通过它来给客户端发送数据 private Session session; // 当前连接的用户ID private String userId; /** * 连接建立成功调用的方法 * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据 * @param userId 路径参数，从连接URI中获取 */ @OnOpen public void onOpen(Session session, @PathParam("userId") String userId) { this.session = session; this.userId = userId; // 将当前session存入全局map if (SESSION_POOL.putIfAbsent(userId, session) == null) { // 如果之前不存在该用户的连接，则在线数加1 int cnt = ONLINE_COUNT.incrementAndGet(); log.info("有新连接加入：用户ID={}，当前在线人数为：{}", userId, cnt); } else { // 如果用户已存在连接（例如多标签页登录），可以选择踢掉旧连接或拒绝新连接 // 这里我们选择踢掉旧的，建立新的 Session oldSession = SESSION_POOL.get(userId); if (oldSession != null && oldSession.isOpen()) { try { oldSession.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "新连接建立，旧连接被关闭")); } catch (IOException e) { log.error("关闭旧连接异常", e); } } SESSION_POOL.put(userId, session); log.info("用户ID={} 已存在连接，已替换为新连接", userId); } // 连接建立后，可以主动发送一条欢迎消息 sendMessage(session, "连接WebSocket服务器成功"); } // 省略其他方法... }3.2 处理连接关闭与异常
连接关闭是常态，我们必须妥善处理，及时清理资源，避免内存泄漏。
/** * 连接关闭调用的方法 */ @OnClose public void onClose() { if (this.userId != null && SESSION_POOL.remove(this.userId, this.session)) { // 从map中成功移除当前session后，在线数减1 int cnt = ONLINE_COUNT.decrementAndGet(); log.info("有一连接关闭：用户ID={}，当前在线人数为：{}", this.userId, cnt); } // 可以在这里触发一些业务逻辑，比如通知该用户的好友“该用户已下线” } /** * 发生错误时调用 * @param session * @param error */ @OnError public void onError(Session session, Throwable error) { log.error("WebSocket发生错误，用户ID={}", this.userId, error); // 通常错误也会导致连接关闭，onClose方法会被调用，所以这里主要做日志记录。 }3.3 实现消息发送工具方法
我们需要一个公共的、线程安全的发送消息方法。因为WebSocket的Session.getBasicRemote().sendText()方法是同步的，在并发下可能有问题，我们应该使用AsyncRemote进行异步发送，并处理可能的异常。
/** * 发送消息给指定用户 * @param userId 用户ID * @param message 消息内容 */ public static void sendMessageToUser(String userId, String message) { Session targetSession = SESSION_POOL.get(userId); if (targetSession != null && targetSession.isOpen()) { sendMessage(targetSession, message); } else { log.warn("用户ID={} 不在线或连接已关闭，消息发送失败: {}", userId, message); // 这里可以结合业务，将消息存入数据库或消息队列，待用户上线后推送 } } /** * 群发消息给所有在线用户 * @param message 消息内容 */ public static void broadcastMessage(String message) { SESSION_POOL.forEach((uid, session) -> { if (session.isOpen()) { sendMessage(session, message); } }); } /** * 内部使用的异步发送方法，封装了异常处理 * @param session 目标会话 * @param message 消息内容 */ private static void sendMessage(Session session, String message) { try { // 使用异步发送，避免阻塞业务线程 session.getAsyncRemote().sendText(message); } catch (Exception e) { log.error("发送WebSocket消息失败，Session ID: {}", session.getId(), e); } }4. 用户身份校验：从路径参数到Token鉴权
在@OnOpen方法中，我们通过路径参数{userId}拿到了用户标识。但这存在严重的安全风险：任何知道URL格式的人都可以伪装成其他用户建立连接。因此，路径参数仅用于初步路由和标识，绝不能作为身份验证的依据。
真正的身份校验，应该在连接建立时的握手阶段完成。WebSocket握手是基于HTTP的，我们可以在连接URI中携带Token（如JWT），并在服务端进行验证。
4.1 客户端连接时携带Token
前端连接时，不能简单地用new WebSocket(“ws://localhost:8080/ws/push/123”)。更安全的做法是将Token放在查询参数中。
// 前端示例 const userId = '123'; const token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'; // 你的JWT Token const ws = new WebSocket(`ws://localhost:8080/ws/push?token=${token}&userId=${userId}`);4.2 服务端握手拦截与校验
在JSR-356中，我们可以实现一个ServerEndpointConfig.Configurator来拦截握手过程。通过重写modifyHandshake方法，我们可以拿到HTTP请求的HandshakeRequest对象，从中提取Token并进行验证。
首先，创建一个配置器：
import javax.servlet.http.HttpServletRequest; import javax.websocket.HandshakeResponse; import javax.websocket.server.HandshakeRequest; import javax.websocket.server.ServerEndpointConfig; import java.util.List; import java.util.Map; public class TokenHandshakeConfigurator extends ServerEndpointConfig.Configurator { @Override public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) { // 获取HTTP Servlet请求对象 Map<String, Object> userProperties = sec.getUserProperties(); HttpServletRequest httpServletRequest = (HttpServletRequest) request.getHttpSession(); userProperties.put(HttpServletRequest.class.getName(), httpServletRequest); // 从请求参数中获取token和userId Map<String, List<String>> parameters = request.getParameterMap(); List<String> tokenList = parameters.get("token"); List<String> userIdList = parameters.get("userId"); if (tokenList != null && !tokenList.isEmpty() && userIdList != null && !userIdList.isEmpty()) { String token = tokenList.get(0); String userId = userIdList.get(0); // 进行Token验证（这里需要你实现自己的JWT解析和验证逻辑） boolean isValid = validateToken(token, userId); if (isValid) { // 验证通过，将userId放入用户属性，供@OnOpen方法使用 userProperties.put("userId", userId); userProperties.put("token", token); } else { // 验证失败，可以在这里抛出异常，阻止连接建立 throw new IllegalArgumentException("Token验证失败"); } } else { throw new IllegalArgumentException("连接参数缺失"); } } private boolean validateToken(String token, String userId) { // 实现你的JWT验证逻辑，例如使用jjwt库 // 校验签名、过期时间，并确认token中的userId与传入的一致 // 返回true/false // 这里只是一个示例，实际需要完整实现 try { // 伪代码: Jwts.parser().setSigningKey(key).parseClaimsJws(token); // 从claims中取出userId进行比对 return true; // 假设验证成功 } catch (Exception e) { return false; } } }然后，修改我们的端点注解，指定使用这个配置器：
@ServerEndpoint(value = "/ws/push", configurator = TokenHandshakeConfigurator.class) public class PushWebSocketEndpoint { // ... @OnOpen public void onOpen(Session session, EndpointConfig config) { // 从配置中获取验证通过的userId this.userId = (String) config.getUserProperties().get("userId"); this.session = session; // ... 后续连接管理逻辑 } // ... }实操心得：Token校验一定要在握手阶段完成。如果在
@OnMessage方法里才校验，攻击者已经建立了连接，会消耗你的服务器资源。握手阶段失败，连接根本不会建立，这是最经济的安全防线。
5. 心跳机制（PING-PONG）实现与连接健康度管理
WebSocket连接可能因为网络波动、代理超时、客户端崩溃等原因无声无息地断开。心跳机制（Heartbeat）就是用来检测连接是否依然存活的生命线。其原理是服务端定期向客户端发送一个PING帧（一种特殊类型的WebSocket控制帧），客户端收到后必须回复一个PONG帧。
5.1 为什么需要自己实现心跳？
你可能听说过WebSocket协议本身有PING/PONG帧。但遗憾的是，JSR-356（Java WebSocket API）并没有向应用层暴露主动发送PING帧的接口。Session对象的getBasicRemote().sendPing()方法并不存在。底层容器（如Tomcat）可能会自动处理PING/PONG，但这对于应用层是透明的，我们无法依赖它来主动探测并处理死连接。
因此，我们需要在应用层模拟心跳。通常有两种方式：
- 业务消息充当心跳：客户端定期发送一条特定的业务消息（如
{"type":"heartbeat"}），服务端收到后回复。这种方式简单，但混淆了业务和保活逻辑。 - 独立的PING-PONG协议：服务端定期发送PING消息（普通文本消息），客户端约定收到后回复PONG消息。我们采用这种方式，因为它更清晰。
5.2 服务端心跳调度器
我们利用Spring的ScheduledExecutorService或@Scheduled注解，创建一个定时任务，遍历所有连接，发送PING并检查超时。
首先，定义一个心跳管理类：
import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import javax.websocket.Session; import java.io.IOException; import java.util.Date; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @Component @EnableScheduling public class WebSocketHeartbeatScheduler { // 记录每个session最后一次收到PONG的时间 private static final Map<String, Long> LAST_PONG_TIME = new ConcurrentHashMap<>(); // 心跳间隔（毫秒） private static final long HEARTBEAT_INTERVAL = 30000; // 30秒 // 超时时间（毫秒），超过此时间未收到PONG则认为连接死亡 private static final long HEARTBEAT_TIMEOUT = 60000; // 60秒 /** * 定时发送PING并检查超时连接 * fixedRate单位是毫秒 */ @Scheduled(fixedRate = HEARTBEAT_INTERVAL) @Async // 使用异步执行，避免阻塞调度线程 public void heartbeatCheck() { long now = System.currentTimeMillis(); PushWebSocketEndpoint.SESSION_POOL.forEach((userId, session) -> { if (session.isOpen()) { try { // 发送PING消息（应用层） session.getAsyncRemote().sendText("PING"); // 检查是否超时 Long lastPongTime = LAST_PONG_TIME.get(session.getId()); if (lastPongTime != null && (now - lastPongTime) > HEARTBEAT_TIMEOUT) { log.warn("用户ID={} 心跳超时，即将关闭连接", userId); session.close(); } } catch (IOException e) { log.error("发送PING或关闭连接失败，用户ID={}", userId, e); } } }); } /** * 更新收到PONG的时间，由端点类在收到PONG消息时调用 * @param sessionId WebSocket Session ID */ public static void updatePongTime(String sessionId) { LAST_PONG_TIME.put(sessionId, System.currentTimeMillis()); } /** * 连接关闭时清理记录 * @param sessionId */ public static void removePongRecord(String sessionId) { LAST_PONG_TIME.remove(sessionId); } }5.3 端点类处理PONG消息
修改PushWebSocketEndpoint类，增加对PONG消息的处理，并在连接关闭时清理记录。
@Component @ServerEndpoint(value = "/ws/push", configurator = TokenHandshakeConfigurator.class) public class PushWebSocketEndpoint { // ... 其他成员变量和方法 /** * 收到客户端消息后调用的方法 * @param message 客户端发送过来的消息 * @param session 可选的参数 */ @OnMessage public void onMessage(String message, Session session) { log.debug("收到来自用户ID={} 的消息: {}", this.userId, message); // 处理心跳回复 if ("PONG".equalsIgnoreCase(message.trim())) { WebSocketHeartbeatScheduler.updatePongTime(session.getId()); log.debug("收到用户ID={} 的心跳回复", this.userId); return; // 心跳消息不进入业务处理 } // 这里是处理其他业务消息的逻辑... // processBusinessMessage(message); } @OnClose public void onClose() { // ... 原有的清理逻辑 WebSocketHeartbeatScheduler.removePongRecord(this.session.getId()); // ... } }5.4 客户端心跳响应
前端也需要相应配合，在收到服务端的“PING”消息后，立刻回复“PONG”。
// 前端WebSocket事件监听 ws.onmessage = function(event) { const msg = event.data; if (msg === 'PING') { // 立即回复PONG ws.send('PONG'); return; } // 处理其他业务消息... console.log('收到业务消息:', msg); };踩坑记录：心跳超时时间
HEARTBEAT_TIMEOUT不能设置得太短。因为网络延迟、客户端GC暂停都可能导致PONG回复慢。通常设置为心跳间隔的2-3倍是比较合理的。另外，一定要在连接关闭时清理LAST_PONG_TIME记录，否则这个Map会一直增长，造成内存泄漏。
6. 用户分组与定向消息广播
简单的全局广播（broadcastMessage）在很多场景下并不适用。比如，我们只想给“北京地区的用户”或者“购买了A产品的用户”发送通知。这就需要分组功能。
6.1 设计分组数据结构
我们需要一个高效的数据结构来维护“组”和“组内用户”的关系。考虑到并发性，我们继续使用ConcurrentHashMap。
import org.springframework.stereotype.Component; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArraySet; @Component public class WebSocketGroupManager { // key: 组名 (例如: “group_admin”, “city_beijing”), value: 该组下的用户ID集合 private static final ConcurrentHashMap<String, Set<String>> GROUP_MEMBERS = new ConcurrentHashMap<>(); /** * 将用户加入指定组 * @param groupName 组名 * @param userId 用户ID */ public void joinGroup(String groupName, String userId) { // computeIfAbsent 是线程安全的，如果组不存在则创建一个新的CopyOnWriteArraySet Set<String> userIds = GROUP_MEMBERS.computeIfAbsent(groupName, k -> new CopyOnWriteArraySet<>()); userIds.add(userId); log.info("用户ID={} 加入组: {}", userId, groupName); } /** * 将用户移出指定组 * @param groupName 组名 * @param userId 用户ID */ public void leaveGroup(String groupName, String userId) { Set<String> userIds = GROUP_MEMBERS.get(groupName); if (userIds != null) { userIds.remove(userId); log.info("用户ID={} 离开组: {}", userId, groupName); // 如果组空了，可以选择移除这个组，避免内存浪费 if (userIds.isEmpty()) { GROUP_MEMBERS.remove(groupName); } } } /** * 用户断开连接时，将其从所有组中移除 * @param userId 用户ID */ public void removeUserFromAllGroups(String userId) { GROUP_MEMBERS.forEach((groupName, userIds) -> { if (userIds.remove(userId)) { log.info("用户ID={} 从组 {} 中移除（连接断开）", userId, groupName); } if (userIds.isEmpty()) { GROUP_MEMBERS.remove(groupName); } }); } /** * 向指定组的所有在线用户发送消息 * @param groupName 组名 * @param message 消息内容 */ public void sendMessageToGroup(String groupName, String message) { Set<String> userIds = GROUP_MEMBERS.get(groupName); if (userIds != null && !userIds.isEmpty()) { userIds.forEach(userId -> { // 复用之前写的单发方法 PushWebSocketEndpoint.sendMessageToUser(userId, message); }); log.info("向组 {} 发送消息，组内成员数: {}", groupName, userIds.size()); } else { log.warn("组 {} 不存在或为空，消息未发送: {}", groupName, message); } } /** * 获取某个组的所有在线成员ID（快照） * @param groupName * @return */ public Set<String> getGroupMembers(String groupName) { Set<String> members = GROUP_MEMBERS.get(groupName); return members == null ? new CopyOnWriteArraySet<>() : new CopyOnWriteArraySet<>(members); } }6.2 在连接生命周期中管理分组
用户通常在连接建立后，通过发送一条“加入组”的指令来订阅某个分组。我们在PushWebSocketEndpoint中处理这个消息。
@Component @ServerEndpoint(value = "/ws/push", configurator = TokenHandshakeConfigurator.class) public class PushWebSocketEndpoint { // ... 注入GroupManager @Autowired private static WebSocketGroupManager groupManager; // 注意：这里需要特殊处理静态注入 // 解决@ServerEndpoint类中@Autowired静态成员注入为null的问题 // 通过一个非静态的setter方法，将Spring容器中的Bean赋值给静态变量 private static WebSocketGroupManager staticGroupManager; @Autowired public void setGroupManager(WebSocketGroupManager groupManager) { PushWebSocketEndpoint.staticGroupManager = groupManager; } @OnMessage public void onMessage(String message, Session session) { // ... 心跳处理逻辑 // 处理业务消息，这里假设消息是JSON格式 try { // 使用简单的JSON解析，实际项目建议用Jackson/Gson // 假设消息格式: {"type": "join_group", "group": "admin"} if (message.contains("\"type\": \"join_group\"")) { // 解析groupName String groupName = ...; // 从message中解析出组名 staticGroupManager.joinGroup(groupName, this.userId); sendMessage(this.session, "已成功加入组: " + groupName); } else if (message.contains("\"type\": \"leave_group\"")) { // 离开组 String groupName = ...; staticGroupManager.leaveGroup(groupName, this.userId); sendMessage(this.session, "已离开组: " + groupName); } else { // 其他业务消息... } } catch (Exception e) { log.error("处理消息失败，用户ID={}, 消息内容={}", this.userId, message, e); sendMessage(this.session, "消息处理错误: " + e.getMessage()); } } @OnClose public void onClose() { // ... 原有的清理逻辑 // 用户断开时，从所有组中移除 if (staticGroupManager != null && this.userId != null) { staticGroupManager.removeUserFromAllGroups(this.userId); } // ... } }技术细节：
@ServerEndpoint是由WebSocket容器管理的，不是Spring Bean，所以其内部无法直接使用@Autowired注入Spring管理的Bean。我们通过一个静态变量和setter方法巧妙地解决了这个问题。这是一种常见的模式。
6.3 业务层调用分组广播
现在，在任何Spring管理的Bean（如Service、Controller）中，你都可以轻松地向特定组发送消息了。
@Service public class NotificationService { @Autowired private WebSocketGroupManager groupManager; public void notifyAdmins(String message) { groupManager.sendMessageToGroup("group_admin", message); } public void notifyUsersInCity(String city, String message) { String groupName = "city_" + city; groupManager.sendMessageToGroup(groupName, message); } }7. 性能优化与生产环境考量
当连接数上升到几千甚至上万时，最初的简单实现可能会遇到性能瓶颈。这里分享几个关键的优化点。
7.1 连接Session存储的优化
我们之前用ConcurrentHashMap<String, Session>存储连接。当需要广播时，会遍历整个Map。如果连接数巨大（例如10万），这个遍历操作会非常耗时，并且会阻塞心跳调度器的线程。
优化方案：按组存储Session引用我们可以在WebSocketGroupManager中，不仅存储用户ID，还直接存储其对应的Session弱引用（WeakReference<Session>）。这样，在发送组播消息时，可以直接获取到组内的Session集合进行发送，无需遍历全局Map。但要注意处理Session关闭后弱引用被GC的情况，需要定期清理无效的引用。
更优方案：引入本地缓存或分布式方案对于单机，可以使用Caffeine或Guava Cache来缓存Session信息，并设置合适的过期策略。对于集群环境，Session不能存在单机内存中，必须引入外部存储如Redis，并配合广播机制（如Redis Pub/Sub）来通知集群内所有节点进行消息推送。这会复杂很多，通常需要引入Spring的WebSocketMessageBroker和STOMP over RabbitMQ/Kafka。
7.2 心跳检查的优化
我们之前的heartbeatCheck方法是遍历所有Session。当连接数很大时，这个循环本身会成为性能热点，并且发送PING是IO操作，在单线程中顺序执行会非常慢。
优化方案：分桶与异步化
- 分桶：将所有的Session分散到多个“桶”（Bucket）中，每个桶由一个独立的线程或定时任务负责心跳检查。这样可以并行处理，提高效率。
- 批量异步发送：使用
CompletableFuture或反应式编程模型，将“发送PING”这个IO操作批量异步执行，避免阻塞心跳检查线程。
// 伪代码：分桶心跳检查思路 @Component public class OptimizedHeartbeatScheduler { private static final int BUCKET_COUNT = 10; private List<ConcurrentHashMap<String, Session>> sessionBuckets; @PostConstruct public void init() { // 初始化10个桶 sessionBuckets = new ArrayList<>(BUCKET_COUNT); for (int i = 0; i < BUCKET_COUNT; i++) { sessionBuckets.add(new ConcurrentHashMap<>()); } } // 根据sessionId的hash值决定放入哪个桶 public void addSession(Session session, String userId) { int bucketIndex = Math.abs(userId.hashCode()) % BUCKET_COUNT; sessionBuckets.get(bucketIndex).put(userId, session); } // 启动10个定时任务，每个任务负责一个桶 @Scheduled(fixedRate = 30000) public void heartbeatBucket0() { checkBucket(0); } // ... 为其他桶也定义类似的任务，或者用一个任务循环处理所有桶但使用线程池 }7.3 消息推送的可靠性保证
我们的sendMessageToUser方法在用户不在线时只是打印了警告。在生产环境中，这通常不够。我们需要一个“离线消息”机制。
简单方案：持久化到数据库当发送消息时，如果目标用户不在线，将消息存入数据库的一张offline_message表中，包含userId，content，createTime等字段。当用户重新建立WebSocket连接后，在@OnOpen方法中，查询该用户的离线消息并推送，然后删除或标记已发送。
进阶方案：消息队列对于高并发、高可靠的场景，应该引入消息队列（如RocketMQ, Kafka）。业务系统将推送事件发送到MQ，由一个独立的推送服务消费MQ，该服务负责维护WebSocket连接和发送。这样实现了业务与推送的解耦，并且可以利用MQ的持久化、重试等特性保证消息不丢失。
7.4 连接数限制与拒绝服务防护
不加限制地允许连接可能导致资源耗尽。我们需要在TokenHandshakeConfigurator的modifyHandshake方法中增加一些防护逻辑。
public class TokenHandshakeConfigurator extends ServerEndpointConfig.Configurator { private static final int MAX_CONNECTIONS_PER_IP = 50; // 每个IP最大连接数 private static final ConcurrentHashMap<String, AtomicInteger> IP_CONNECTION_COUNT = new ConcurrentHashMap<>(); @Override public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) { // ... Token验证逻辑 // 获取客户端IP String clientIp = getClientIp(request); // 检查IP连接数 AtomicInteger count = IP_CONNECTION_COUNT.computeIfAbsent(clientIp, k -> new AtomicInteger(0)); if (count.incrementAndGet() > MAX_CONNECTIONS_PER_IP) { count.decrementAndGet(); // 恢复计数 throw new IllegalArgumentException("连接数超限"); } // 将IP和计数器引用存入用户属性，方便连接关闭时递减 sec.getUserProperties().put("clientIp", clientIp); sec.getUserProperties().put("ipCounter", count); } // 在连接关闭的监听器中递减计数（需要在Endpoint中获取并调用） public static void decrementIpCount(EndpointConfig config) { String clientIp = (String) config.getUserProperties().get("clientIp"); AtomicInteger counter = (AtomicInteger) config.getUserProperties().get("ipCounter"); if (counter != null) { counter.decrementAndGet(); if (counter.get() <= 0) { IP_CONNECTION_COUNT.remove(clientIp, counter); } } } private String getClientIp(HandshakeRequest request) { // 从请求头中获取真实IP，注意处理代理（如X-Forwarded-For） // 这里是简化版 Map<String, List<String>> headers = request.getHeaders(); List<String> ipHeaders = headers.get("X-Forwarded-For"); if (ipHeaders != null && !ipHeaders.isEmpty()) { return ipHeaders.get(0).split(",")[0].trim(); } // 否则从HttpServletRequest中获取 HttpServletRequest req = (HttpServletRequest) request.getHttpSession(); return req.getRemoteAddr(); } }然后在PushWebSocketEndpoint的@OnClose方法中调用TokenHandshakeConfigurator.decrementIpCount(this.session.getUserProperties())。
8. 前端集成示例与常见问题排查
服务端准备好了，前端如何对接？这里给出一个精简但完整的Vue 3组件示例，并附上几个我踩过的坑。
8.1 Vue 3组件示例
<template> <div> <p>连接状态: {{ status }}</p> <button @click="connect" :disabled="isConnected">连接</button> <button @click="disconnect" :disabled="!isConnected">断开</button> <button @click="joinAdminGroup" :disabled="!isConnected">加入管理员组</button> <ul> <li v-for="(msg, index) in messages" :key="index">{{ msg }}</li> </ul> </div> </template> <script setup> import { ref, onUnmounted } from 'vue'; const ws = ref(null); const status = ref('未连接'); const isConnected = ref(false); const messages = ref([]); const connect = () => { const userId = 'user_123'; const token = 'your_jwt_token_here'; // 应从登录状态获取 const wsUrl = `ws://${location.host}/ws/push?token=${token}&userId=${userId}`; ws.value = new WebSocket(wsUrl); ws.value.onopen = () => { status.value = '已连接'; isConnected.value = true; messages.value.push('WebSocket连接已建立'); }; ws.value.onmessage = (event) => { const msg = event.data; if (msg === 'PING') { ws.value.send('PONG'); console.log('已回复PONG'); return; } messages.value.push(`收到: ${msg}`); }; ws.value.onerror = (error) => { console.error('WebSocket错误:', error); status.value = '连接错误'; }; ws.value.onclose = () => { status.value = '已断开'; isConnected.value = false; messages.value.push('WebSocket连接已关闭'); }; }; const disconnect = () => { if (ws.value) { ws.value.close(); ws.value = null; } }; const joinAdminGroup = () => { if (ws.value && ws.value.readyState === WebSocket.OPEN) { const joinCmd = JSON.stringify({ type: 'join_group', group: 'group_admin' }); ws.value.send(joinCmd); messages.value.push('已发送加入管理员组请求'); } }; // 组件卸载时自动断开连接 onUnmounted(() => { disconnect(); }); </script>8.2 常见问题与排查清单
连接失败，返回404
- 检查端点路径：确认前端连接的URL（
/ws/push）与后端@ServerEndpoint注解中的value完全一致。 - 检查配置Bean：确认
ServerEndpointExporterBean已正确配置。如果项目是SpringBoot内嵌容器，必须有这个Bean。 - 检查跨域：如果前端与后端域名/端口不同，需要配置CORS。对于WebSocket，CORS在握手阶段生效。你可以在
TokenHandshakeConfigurator的modifyHandshake方法中添加响应头：response.getHeaders().put(“Access-Control-Allow-Origin”, request.getHeaders().get(“Origin”)); （注意生产环境要严格限制Origin）。
- 检查端点路径：确认前端连接的URL（
连接建立后立刻断开
- 检查Token校验：在
TokenHandshakeConfigurator的modifyHandshake中抛出的任何异常都会导致握手失败，连接关闭。查看服务器日志，确认Token验证逻辑无误。 - 检查Nginx等代理配置：如果你使用了Nginx反向代理，必须配置其支持WebSocket。关键配置如下：
location /ws/ { proxy_pass http://backend_server; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_set_header Host $host; proxy_read_timeout 3600s; # 长连接超时时间 }
- 检查Token校验：在
心跳不工作，连接一段时间后自动断开
- 确认心跳逻辑被执行：在服务端
heartbeatCheck方法内打日志，看定时任务是否正常触发。 - 确认PING/PONG消息格式：服务端发的是纯文本
”PING”，前端判断的是event.data === ‘PING’。务必注意大小写和可能的空格。建议前后端统一使用JSON格式，如{“type”: “heartbeat”, “data”: “ping”}。 - 检查防火墙/代理超时：很多网络设备（如阿里云SLB）会对空闲连接设置超时（通常60秒）。你的心跳间隔必须小于这个超时时间。建议设置为30秒。
- 确认心跳逻辑被执行：在服务端
发送消息时出现
IllegalStateException: The remote endpoint was in state [TEXT_FULL_WRITING]- 原因：同一个
Session上，前一个异步发送操作还没完成，又触发了新的发送，导致状态冲突。 - 解决：确保你的发送方法是线程安全的。我们之前使用的
session.getAsyncRemote().sendText()是线程安全的，但如果你在多个线程中同时调用同一个Session的发送方法，仍有小概率出错。更稳妥的做法是使用同步发送+队列。可以为每个Session维护一个消息队列，用一个单线程池依次发送。对于吞吐要求不极高的推送场景，同步发送session.getBasicRemote().sendText()在简单加锁后反而更稳定。
- 原因：同一个
内存泄漏，连接数只增不减
- 检查
@OnClose和@OnError方法：确保在所有连接关闭的路径上（正常关闭、异常关闭、心跳超时强制关闭），都从SESSION_POOL和GROUP_MEMBERS等全局容器中移除了对应的Session和用户信息。 - 使用弱引用或定期清理：如前所述，考虑使用
WeakReference<Session>，或者定期扫描SESSION_POOL，移除已经!session.isOpen()的死连接。
- 检查
这套从连接管理、安全校验、心跳保活到分组广播的WebSocket实现方案，经过多个中等流量项目的验证，稳定性和扩展性都不错。它最大的价值在于清晰地将各个关注点（连接、安全、健康、路由）分离，代码结构一目了然，后续无论是加监控、改持久化方案还是接入消息队列，都有清晰的切入点可以操作。在实际部署时，记得根据压测结果调整线程池、心跳参数和JVM内存设置，特别是SESSION_POOL的规模，要做好预估。
