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 APISpringBoot集成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 groupIdorg.springframework.boot/groupId artifactIdspring-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 ConcurrentHashMapString, 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请求对象 MapString, Object userProperties sec.getUserProperties(); HttpServletRequest httpServletRequest (HttpServletRequest) request.getHttpSession(); userProperties.put(HttpServletRequest.class.getName(), httpServletRequest); // 从请求参数中获取token和userId MapString, ListString parameters request.getParameterMap(); ListString tokenList parameters.get(token); ListString 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-356Java 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 MapString, 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 ConcurrentHashMapString, SetString GROUP_MEMBERS new ConcurrentHashMap(); /** * 将用户加入指定组 * param groupName 组名 * param userId 用户ID */ public void joinGroup(String groupName, String userId) { // computeIfAbsent 是线程安全的如果组不存在则创建一个新的CopyOnWriteArraySet SetString 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) { SetString 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) { SetString 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 SetString getGroupMembers(String groupName) { SetString 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存储的优化我们之前用ConcurrentHashMapString, Session存储连接。当需要广播时会遍历整个Map。如果连接数巨大例如10万这个遍历操作会非常耗时并且会阻塞心跳调度器的线程。优化方案按组存储Session引用我们可以在WebSocketGroupManager中不仅存储用户ID还直接存储其对应的Session弱引用WeakReferenceSession。这样在发送组播消息时可以直接获取到组内的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 ListConcurrentHashMapString, 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表中包含userIdcontentcreateTime等字段。当用户重新建立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 ConcurrentHashMapString, 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 // 这里是简化版 MapString, ListString headers request.getHeaders(); ListString 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 clickconnect :disabledisConnected连接/button button clickdisconnect :disabled!isConnected断开/button button clickjoinAdminGroup :disabled!isConnected加入管理员组/button ul li v-for(msg, index) in messages :keyindex{{ 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(); }); /script8.2 常见问题与排查清单连接失败返回404检查端点路径确认前端连接的URL/ws/push与后端ServerEndpoint注解中的value完全一致。检查配置Bean确认ServerEndpointExporterBean已正确配置。如果项目是SpringBoot内嵌容器必须有这个Bean。检查跨域如果前端与后端域名/端口不同需要配置CORS。对于WebSocketCORS在握手阶段生效。你可以在TokenHandshakeConfigurator的modifyHandshake方法中添加响应头response.getHeaders().put(“Access-Control-Allow-Origin”, request.getHeaders().get(“Origin”)); 注意生产环境要严格限制Origin。连接建立后立刻断开检查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; # 长连接超时时间 }心跳不工作连接一段时间后自动断开确认心跳逻辑被执行在服务端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和用户信息。使用弱引用或定期清理如前所述考虑使用WeakReferenceSession或者定期扫描SESSION_POOL移除已经!session.isOpen()的死连接。这套从连接管理、安全校验、心跳保活到分组广播的WebSocket实现方案经过多个中等流量项目的验证稳定性和扩展性都不错。它最大的价值在于清晰地将各个关注点连接、安全、健康、路由分离代码结构一目了然后续无论是加监控、改持久化方案还是接入消息队列都有清晰的切入点可以操作。在实际部署时记得根据压测结果调整线程池、心跳参数和JVM内存设置特别是SESSION_POOL的规模要做好预估。