Websocket 使用指南
创建 TextWebSocketHandler 子类
package socket;
@Component
public class MyWebSocketHandler implements WebSocketHandler {
// 1. 连接建立成功 (生命周期开始)
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
System.out.println("连接已建立: " + session.getId());
// 常用操作：
// 1. 将 session 存入全局 Map (userId -> session)
// 2. 设置 session 属性 (session.getAttributes().put("uid", userId))
// 3. 发送欢迎消息
session.sendMessage(new TextMessage("欢迎加入！"));
}
// 2. 接收到消息 (活跃阶段)
@Override
public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
// 处理业务逻辑
}
// 3. 传输错误 (活跃阶段的异常)
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
System.err.println("传输错误: " + exception.getMessage());
// 通常这里需要手动关闭 session
if (session.isOpen()) {
session.close();
}
}
// 4. 连接关闭 (生命周期结束)
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
System.out.println("连接已关闭: " + session.getId() + ", 状态码: " + closeStatus);
// 关键善后操作：
// 1. 从全局 Map 中移除 session (防止内存泄漏)
// 2. 广播通知其他人该用户已下线
// 3. 记录日志
}
// 5. 是否支持部分消息 (通常返回 false)
@Override
public boolean supportsPartialMessages() {
return false;
}
}
注册 WebSocket
以WebSocket AutoConfig 示例
其中重要的方式是 前三行
at socket.WebSocketConfig.registerWebSocketHandlers(WebSocketConfig.java:21)
at org.springframework.web.socket.config.annotation.DelegatingWebSocketConfiguration.registerWebSocketHandlers(DelegatingWebSocketConfiguration.java:51)
at org.springframework.web.socket.config.annotation.WebSocketConfigurationSupport.webSocketHandlerMapping(WebSocketConfigurationSupport.java:34)
WebSocketConfigurationSupport#webSocketHandlerMapping
@Bean
public HandlerMapping webSocketHandlerMapping() {
// 创建一个线程池, 把线程池传递给 HandlerRegistry
ServletWebSocketHandlerRegistry registry = new ServletWebSocketHandlerRegistry(defaultSockJsTaskScheduler());
// 由子类实现具体的注册逻辑 默认是 DelegatingWebSocketConfiguration
// 子类会
registerWebSocketHandlers(registry);
return registry.getHandlerMapping();
}
WebSocket-Session 的生命周期
一、生命周期的四个核心阶段
1. 握手阶段 (Handshake / Connection Opening)
这是生命周期的起点。
- 动作：客户端发送一个带有
Upgrade: websocket头的 HTTP 请求。 - 服务端处理：
- 拦截请求（如 Spring 中的
HandshakeInterceptor）。 - 验证权限（Token 校验、IP 白名单等）。
- 如果验证通过，协议升级为 WebSocket，创建 Session 对象。
- 如果验证失败，返回 HTTP 401/403，连接终止，Session 不会创建。
- 拦截请求（如 Spring 中的
- 关键事件：
afterConnectionEstablished(Spring) /onOpen(JSR-356)。 - 此时状态：Session 已建立，处于
OPEN状态，可以开始收发消息。
2. 活跃阶段 (Active / Connected)
这是生命周期的主要阶段，持续时间可能从几秒到几天不等。
- 特征：
- TCP 连接保持打开。
- 服务端和客户端可以随时互相推送消息（全双工）。
- Session 对象中通常存储了用户上下文（如 UserID、房间ID、属性 Attributes）。
- 心跳检测 (Heartbeat/Ping-Pong)：
- 为了防止防火墙切断空闲连接或检测客户端是否“假死”，双方会定期发送 Ping/Pong 帧。
- 如果长时间未收到心跳，服务端会主动判定 Session 失效并关闭它。
- 异常处理：如果在此期间发生网络波动或代码异常，会触发错误回调。
3. 关闭阶段 (Closing / Connection Closing)
这是生命周期的终点触发点。关闭可以由任意一方发起。
- 正常关闭：
- 客户端调用
socket.close()或服务端调用session.close()。 - 发送一个 Close Frame（包含状态码，如 1000 表示正常关闭）。
- 双方收到 Close Frame 后，断开 TCP 连接。
- 客户端调用
- 异常关闭：
- 网络中断、客户端浏览器崩溃、服务器宕机、心跳超时。
- 此时可能没有正常的 Close Frame，直接触发 TCP
FIN或RST。
- 关键事件：
afterConnectionClosed(Spring) /onClose(JSR-356)。
4. 销毁阶段 (Destroyed / Cleanup)
这是生命周期的善后阶段。
- 动作：
- 释放 Session 占用的内存资源。
- 清理服务端缓存（如从
ConcurrentHashMap<UserId, Session>中移除该 Session）。 - 更新业务状态（如将用户状态改为“离线”、通知房间内其他人“某人已离开”）。
- 注意：这一步必须在代码中显式处理，否则会导致内存泄漏或在线人数统计不准。
HandshakeInterceptor
作用 在握手阶段进行拦截（在 afterConnectionEstablished 之前执行）：
beforeHandshake: 验证 Token，失败则返回false阻止连接。afterHandshake: 连接刚建立，可以做些初始化日志记录。
执行流程
at org.springframework.web.socket.server.support.OriginHandshakeInterceptor.beforeHandshake(OriginHandshakeInterceptor.java:93)
at org.springframework.web.socket.server.support.HandshakeInterceptorChain.applyBeforeHandshake(HandshakeInterceptorChain.java:59)
at org.springframework.web.socket.server.support.WebSocketHttpRequestHandler.handleRequest(WebSocketHttpRequestHandler.java:163)
at org.springframework.web.servlet.mvc.HttpRequestHandlerAdapter.handle(HttpRequestHandlerAdapter.java:51)
at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:967)
at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:901)
at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:970)
at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:861)
WS 接口就像 普通http 请求一样同样使用 DIspatcherServerlet.doDispatch 分发请求，这里通过 WebSocketHandlerMapping 得到 WebSocketHttpRequestHandler .
这里体现到了 Spring 的 装饰器模式，org.springframework.web.socket.server.support.WebSocketHttpRequestHandler#handleRequest 执行 拦截器的before 以及after 方法是在 HandshakeInterceptorChain 中，这里体现了 责任链模式。
常见状态码 (CloseStatus)
当连接关闭时，会携带一个状态码，理解它们有助于排查问题：
| 状态码 | 含义 | 场景 |
|---|---|---|
| 1000 | Normal Closure | 正常完成任务，主动关闭。 |
| 1001 | Going Away | 服务端关闭或浏览器导航到新页面。 |
| 1002 | Protocol Error | 协议错误， endpoint 终止连接。 |
| 1006 | Abnormal Closure | 最常见。没有收到 Close Frame 就断开了（如网络断了、进程崩溃）。 |
| 1008 | Policy Violation | 消息太大或违反策略。 |
| 1011 | Internal Server Error | 服务端遇到意外情况。 |
| 1015 | TLS Handshake Failure | SSL/TLS 握手失败。 |
开发中的关键注意事项
-
线程安全：
WebSocketSession的sendMessage方法通常是线程安全的，但如果你自己维护了一个Map<String, WebSocketSession>来存储会话，对这个 Map 的读写（put/remove）必须是线程安全的（使用ConcurrentHashMap）。
-
内存泄漏风险：
- 最严重的错误：在
afterConnectionClosed中忘记从全局集合中移除 Session。这会导致随着时间推移，服务端内存中堆积大量无效 Session 对象，最终 OOM (Out Of Memory)。
- 最严重的错误：在
-
心跳机制 (Heartbeat)：
- 不要依赖 TCP 自带的 Keep-Alive，它在应用层反应太慢。
- 必须在应用层实现心跳：客户端每 30 秒发一个 "ping"，服务端收到回 "pong"。如果服务端 90 秒没收到任何消息，主动关闭 Session。
-
集群环境下的 Session：
- 单机版的
Map<UserId, Session>在集群（多实例）环境下失效。 - 如果用户 A 连接在节点 1，用户 B 连接在节点 2，A 想发消息给 B，节点 1 找不到 B 的 Session。
- 解决方案：引入消息中间件（Redis Pub/Sub, RabbitMQ, Kafka）。节点 1 把消息发给 Redis，节点 2 订阅 Redis 收到消息后，再转发给本地的 Session。
- 单机版的
-
握手拦截器的异步陷阱：
- 在
beforeHandshake中尽量不要做耗时的同步 IO 操作（如查数据库），这会阻塞 Netty/Tomcat 的 IO 线程。如果必须查库，建议使用异步方式或预先缓存 Token。
- 在
sequenceDiagram
participant Client as 客户端
participant Server as 服务端 (Spring)
participant Store as 全局缓存/DB
Client->>Server: HTTP Request (Upgrade: websocket)
Note over Server: HandshakeInterceptor<br/>(验证 Token/权限)
alt 验证失败
Server-->>Client: HTTP 401/403
Note over Server: 生命周期未开始
else 验证成功
Server->>Server: 创建 WebSocketSession
Server->>Store: 存入 Session (userId -> session)
Server-->>Client: HTTP 101 Switching Protocols
Note over Server: 阶段 1: 连接建立 (afterConnectionEstablished)
loop 活跃阶段 (心跳/业务消息)
Client->>Server: 发送消息
Server->>Server: handleMessage
Server->>Client: 推送消息
Note over Client,Server: Ping/Pong 心跳检测
end
opt 异常或主动关闭
Client->>Server: Close Frame (或网络断开)
Server->>Server: afterConnectionClosed
Server->>Store: 移除 Session (防止泄漏)
Server->>Store: 更新用户状态为离线
Note over Server: 阶段 4: 销毁/清理
end
end
WebSocket 包类说明文档
这个 WebSocket 模块实现了完整的连接管理、心跳检测、集群消息推送功能，支持高并发和分布式部署。
1. WebSocketConfig - WebSocket 配置类
职责：WebSocket 的入口配置，注册 WebSocket 处理器
启用 WebSocket 支持
注册 WebSocket 处理器到 /ws/connection 路径
配置跨域访问规则
2. WebSocketConnectionProperties - 连接配置属性类
职责：管理 WebSocket 连接的配置参数
heartbeatInterval: 心跳间隔时间（默认 30 秒）
expiredMultiple: 超时倍数（默认 2 倍）
preventDuplicateConnection: 是否启用防重复连接
maxConnectionLimit: 最大连接数限制（默认 100）
connectionTimeout: 连接超时时间
3. ConnectionRegistry - 连接注册表
职责：管理所有活跃的 WebSocket 连接
维护活跃连接的 Map（connectionKey → WebSocketSessionWrapper）
提供用户标识到连接键的映射
提供系统代码到连接键的映射
支持按用户、按系统查询在线连接
线程安全（使用 ConcurrentHashMap）
4. WebSocketSessionWrapper - 会话包装器
职责：封装 WebSocket 会话及相关元数据
webSocketSession: 原始 WebSocket 会话
timeout: Netty 心跳超时任务
connectTime: 连接建立时间戳
userIdentifier: 用户标识
systemCode: 系统代码
5. ConnectionValidator - 连接验证器
职责：验证和管理 WebSocket 连接的有效性
生成唯一的 connectionKey（userId:systemCode:clientIdentifier）
使用 Redis 防止重复连接（分布式锁）
处理已存在的连接（关闭旧连接）
检查连接数限制
清理 Redis 连接记录
6. WebSocketConnectionManager - 连接管理器（心跳监控）
职责：管理 WebSocket 连接的生命周期和心跳检测
使用 Netty 的 HashedWheelTimer 实现高效定时任务
启动连接时创建心跳超时任务
收到消息时更新心跳（取消旧任务，创建新任务）
处理连接超时（发送超时通知并关闭连接）
停止连接时清理资源
提供在线用户统计功能
7. WebSocketHandler - WebSocket 处理器
职责：处理 WebSocket 连接的生命周期事件和消息
afterConnectionEstablished: 连接建立后的初始化
handleTextMessage: 处理接收到的文本消息
心跳消息（HEARTBEAT → HEARTBEAT_ACK）
业务消息
afterConnectionClosed: 连接关闭后的清理
handleTransportError: 处理传输错误
8. WebSocketMessage - 消息模型
职责：定义集群间推送的消息结构
type: 消息类型（UNICAST/BROADCAST/GROUP）
targetId: 目标 ID（用户 ID 或系统代码）
content: 消息内容
timestamp: 时间戳
9. WebSocketMessageListener - Redis 消息监听器
职责：监听 Redis 频道，处理集群间的消息同步
订阅 Redis 频道 websocket:message:all
异步处理接收到的消息（使用线程池）
根据消息类型分发处理：
UNICAST: 单播给指定用户
BROADCAST: 广播给所有在线用户
GROUP: 组播给指定系统的所有用户
统计消息发送结果（成功/失败数量）
10. WebSocketMessageService - 消息推送服务
职责：提供消息推送的业务逻辑（支持集群）
sendToUser: 向指定用户推送消息
优先检查本地节点
本地没有则发布到 Redis
broadcast: 广播消息到所有节点
sendToGroup: 向指定系统的所有用户推送
isUserOnline: 检查用户是否在线
通过 Redis Pub/Sub 实现集群间消息同步
11. WebSocketPushController - 消息推送 REST 接口
职责：提供 HTTP 接口供外部调用消息推送功能
POST /api/websocket/push/user: 推送消息给指定用户
POST /api/websocket/push/broadcast: 广播消息
POST /api/websocket/push/group: 推送消息给指定系统
GET /api/websocket/push/online-users: 查询在线用户列表
架构图
graph TB
subgraph "客户端层"
Client1[客户端 1]
Client2[客户端 2]
Client3[客户端 3 - 不同系统]
end
subgraph "WebSocket 接入层"
WSConfig[WebSocketConfig<br/>配置入口]
WSHandler[WebSocketHandler<br/>消息处理器]
end
subgraph "连接管理层"
ConnMgr[WebSocketConnectionManager<br/>连接管理器]
ConnVal[ConnectionValidator<br/>连接验证器]
ConnReg[ConnectionRegistry<br/>连接注册表]
Wrapper[WebSocketSessionWrapper<br/>会话包装器]
end
subgraph "消息服务层"
MsgService[WebSocketMessageService<br/>消息推送服务]
MsgListener[WebSocketMessageListener<br/>Redis 消息监听器]
PushCtrl[WebSocketPushController<br/>REST API]
end
subgraph "中间件层"
Redis[(Redis<br/>连接存储 + 消息队列)]
end
subgraph "配置类"
Properties[WebSocketConnectionProperties<br/>连接配置]
end
%% 客户端连接
Client1 -->|WebSocket 连接 | WSConfig
Client2 -->|WebSocket 连接 | WSConfig
Client3 -->|WebSocket 连接 | WSConfig
%% WebSocket 配置路由到处理器
WSConfig --> WSHandler
%% 处理器与连接管理层交互
WSHandler --> ConnMgr
WSHandler --> ConnVal
%% 连接管理层内部依赖
ConnMgr --> ConnVal
ConnMgr --> ConnReg
ConnVal --> ConnReg
ConnReg --> Wrapper
%% 连接管理层与 Redis 交互
ConnVal -->|存储连接信息 | Redis
ConnMgr -->|心跳检测 | ConnMgr
%% 消息服务层与连接管理层交互
MsgService --> ConnReg
MsgListener --> ConnReg
%% 消息服务层与 Redis 交互
MsgService -->|发布消息 | Redis
MsgListener -->|订阅消息 | Redis
%% REST API 调用消息服务
PushCtrl --> MsgService
%% 配置类被各组件使用
Properties -.-> ConnMgr
Properties -.-> ConnVal
%% 样式
style WSConfig fill:#e1f5fe
style WSHandler fill:#e1f5fe
style ConnMgr fill:#fff3e0
style ConnVal fill:#fff3e0
style ConnReg fill:#fff3e0
style Wrapper fill:#fff3e0
style MsgService fill:#f3e5f5
style MsgListener fill:#f3e5f5
style PushCtrl fill:#f3e5f5
style Redis fill:#c8e6c9
style Properties fill:#ffebee
核心流程图
- 连接建立流程
graph LR
A[客户端发起连接] --> B[WebSocketConfig]
B --> C[WebSocketHandler<br/>afterConnectionEstablished]
C --> D[WebSocketConnectionManager.start]
D --> E[ConnectionValidator<br/>generateConnectionKey]
E --> F[检查连接数限制]
F --> G{验证通过？}
G -->|否 | H[拒绝连接]
G -->|是 | I[ConnectionValidator<br/>validateNewConnection]
I --> J[Redis SETNX<br/>分布式锁]
J --> K{连接已存在？}
K -->|是 | L[关闭旧连接]
K -->|否 | M[创建心跳超时任务]
L --> M
M --> N[创建 WebSocketSessionWrapper]
N --> O[ConnectionRegistry<br/>addSession]
O --> P[连接建立成功]
- 心跳检测流程
graph LR
A[收到客户端消息] --> B{消息类型？}
B -->|HEARTBEAT| C[WebSocketHandler<br/>handleTextMessage]
C --> D[WebSocketConnectionManager<br/>onHeartbeat]
D --> E[取消旧的超时任务]
E --> F[创建新的超时任务]
F --> G[更新 ConnectionRegistry]
G --> H[发送 HEARTBEAT_ACK]
B -->|其他消息 | I[updateHeartbeat]
I --> E
F --> J[延迟时间到达？]
J -->|是 | K[触发超时回调]
K --> L[handleConnectionTimeout]
L --> M[发送超时通知]
M --> N[关闭连接]
N --> O[从 Registry 移除]
- 消息推送流程（集群支持）
graph LR
A[外部调用 Push API] --> B[WebSocketPushController]
B --> C[WebSocketMessageService]
C --> D{消息类型？}
D -->|单播 | E[检查本地是否有用户]
E --> F{用户在本地？}
F -->|是 | G[直接发送]
F -->|否 | H[发布到 Redis]
D -->|广播 | H
D -->|组播 | H
H --> I[Redis Pub/Sub]
I --> J[所有节点监听器]
J --> K[WebSocketMessageListener<br/>onMessage]
K --> L[异步处理消息]
L --> M{消息类型？}
M -->|UNICAST| N[sendToUser]
M -->|BROADCAST| O[broadcast]
M -->|GROUP| P[sendToGroup]
N --> Q[从 Registry 获取会话]
O --> Q
P --> Q
Q --> R{会话存在且打开？}
R -->|是 | S[sendMessage]
R -->|否 | T[跳过]
- 重复连接处理流程
graph LR
A[新连接请求] --> B[生成 connectionKey]
B --> C[Redis SETNX]
C --> D{Key 已存在？}
D -->|否 | E[连接成功]
D -->|是 | F[获取旧 sessionId]
F --> G[从 Registry 获取旧会话]
G --> H{旧会话存在且打开？}
H -->|否 | I[清理 Redis 记录]
I --> E
H -->|是 | J[发送重复连接通知]
J --> K[关闭旧会话]
K --> L[从 Registry 移除]
L --> M[允许新连接]
关键技术点
1. 防重复连接机制
使用 Redis 的 SETNX 实现分布式锁
connectionKey = userId:systemCode:clientIdentifier
检测到重复连接时，先关闭旧连接再建立新连接
2. 心跳检测机制
使用 Netty 的 HashedWheelTimer（高性能时间轮算法）
每次收到消息都重置超时任务
超时后发送通知并关闭连接
3. 集群消息同步
使用 Redis Pub/Sub 实现集群间消息广播
每个节点订阅相同的频道
收到消息后先检查本地，本地没有再通过 Redis 推送
4. 线程安全设计
ConnectionRegistry 使用 ConcurrentHashMap
消息发送时使用 synchronized 锁住 session
消息监听器使用独立线程池异步处理
5. 资源管理
连接关闭时自动取消超时任务
清理 Redis 中的连接记录
应用关闭时优雅停止线程池和时间轮
配置示例
websocket:
connection:
heartbeat-interval: 30000 # 心跳间隔 30 秒
expired-multiple: 2 # 超时倍数 2 倍
prevent-duplicate-connection: true # 启用防重复连接
max-connection-limit: 100 # 最大连接数 100
connection-timeout: 60000 # 连接超时 60 秒
代码
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
9
0- 0
已为社区贡献2条内容
所有评论(0)