WebSocket 基础架构：从零搭建企业 IM 的实时通信底座

一、为什么用 WebSocket？——从轮询到推送的进化

“新消息来了得刷新页面才能看到？这算什么 IM。”

这是我刚做完第一版消息功能时，室友试用了五秒钟之后甩给我的评价。说实话，我当时实现的消息流程是这样的：

员工 A 发送消息 → 存入数据库
员工 B 要看到新消息 → 刷新页面 → 调接口拉列表

这玩意儿顶多算个留言板，跟 IM 一毛钱关系都没有。

真正的 IM 必须是服务端主动推消息给客户端。而 HTTP 请求-响应模型天生做不到这一点——客户端不问，服务端不能说。常用的解法有三种：

方案	原理	优点	缺点
短轮询	前端定时 3 秒调一次接口	实现简单	大部分请求是空的，浪费带宽
长轮询	请求挂起直到有新消息或超时	比短轮询省请求数	服务端线程一直被占用
WebSocket	TCP 全双工连接，服务端随时推送	实时、高效	需要处理断线重连、心跳、认证

对于 IM 来说，WebSocket 是唯一正确的选择。 不是因为其他方案做不出来，而是因为短轮询和长轮询的架构缺陷在消息量大时会放大到不可接受——每一条消息都要走一遍 HTTP 握手、认证、路由，这些开销在 WebSocket 里只需要一次连接时就搞定。

我一开始确实想过偷懒用短轮询，但算了一笔账：假设 50 个员工在线，每人 3 秒轮询一次，每分钟 1000 次请求。其中 980 次返回"没有新消息"。WebSocket 一条连接挂着就行，有新消息才推。差距不是几倍，是两个数量级。

二、Spring 原生 WebSocket——为什么不选 STOMP？

Spring 对 WebSocket 有两层支持：

1. 原生 WebSocket（JSR 356 / TextWebSocketHandler）

直接基于 TCP 层的 WebSocket 协议，手动收发文本帧。代码结构就是一个 Handler 类 + 四个生命周期方法。

2. STOMP over WebSocket

在 WebSocket 之上加了一层 STOMP 协议，支持订阅（/topic）和点对点（/user/queue）语义。Spring 提供了 @Controller + @MessageMapping 的注解，看起来跟写 HTTP 接口差不多。

我选了原生 WebSocket，原因很简单：

Workforce Hub 的推送场景很明确——服务端给特定用户推消息，不需要客户端"订阅"某个频道
STOMP 多了一层协议开销，每个 STOMP 帧都有 command + headers + body 结构，比纯 JSON 更臃肿
事件类型已经够多了（12 种，后面会讲），不需要 STOMP 的路由语义再来一层复杂度
原生 WebSocket 调试简单——抓包看就是 JSON 字符串，不需要解析 STOMP 帧格式

说实话，如果是一个需要"聊天室"或者"多 Topic 订阅"的场景，STOMP 可能更合适。但我们只有一个通道——推事件。没必要为一个通道引入一整套消息协议。

三、连接管理——握手、鉴权与生命周期

3.1 整体架构

IM WebSocket 的核心组件就四个：

HandshakeInterceptor：握手阶段负责从 HTTP Session 中提取当前登录用户
WebSocketHandler：连接建立、消息接收、连接关闭、异常处理的生命周期管理
SessionRegistry：维护"用户 ID → WebSocket 连接"的映射表，实现按用户路由推送
PushService：推送门面，收口所有业务模块的实时事件

3.2 握手鉴权——复用 HTTP Session

WebSocket 连接建立的握手阶段还是一个 HTTP 请求，浏览器会自动带上 Cookie。这意味着我们可以直接从现有的登录会话中提取用户信息，不需要重新认证。

// HandshakeInterceptor 的核心逻辑
@Override
public boolean beforeHandshake(ServerHttpRequest request,
                               ServerHttpResponse response,
                               WebSocketHandler wsHandler,
                               Map<String, Object> attributes) {
    // Step 1: 从 Spring Security 上下文或 HTTP Session 中提取当前用户
    WfhUserDetails user = resolveCurrentUser(request);
    if (user == null) {
        response.setStatusCode(HttpStatus.UNAUTHORIZED);  // 没登录 → 401
        return false;
    }
    if (user.getUserType() != UserType.USER) {
        response.setStatusCode(HttpStatus.FORBIDDEN);     // 不是员工用户 → 403
        return false;
    }
    // Step 2: 检查员工是否被禁用（已离职/停用的员工不能连接）
    if (!imRepository.existsLoginAllowedOrgUser(user.getUserId())) {
        response.setStatusCode(HttpStatus.FORBIDDEN);
        return false;
    }
    // Step 3: 把用户 ID 注入 WebSocket Session 属性，后续 Handler 直接用
    attributes.put(ImWebSocketHandler.USER_ID_ATTRIBUTE, user.getUserId());
    attributes.put(ImWebSocketHandler.DISPLAY_NAME_ATTRIBUTE, user.getDisplayName());
    return true;
}

关键细节：resolveCurrentUser 用了双重策略——先尝试 Spring Security 的 SecurityContextHolder（通常在线程上下文里），拿不到再从 HTTP Session 的 SPRING_SECURITY_CONTEXT 属性里取：

private WfhUserDetails resolveCurrentUser(ServerHttpRequest request) {
    // 策略 1：从当前线程的 SecurityContext 拿
    WfhUserDetails current = resolveFromAuthentication(
        SecurityContextHolder.getContext().getAuthentication());
    if (current != null) return current;

    // 策略 2：从 HTTP Session 拿（WebSocket 握手可能不在原请求线程）
    if (request instanceof ServletServerHttpRequest servletRequest) {
        HttpSession session = servletRequest.getServletRequest().getSession(false);
        if (session == null) return null;
        Object context = session.getAttribute(
            HttpSessionSecurityContextRepository.SPRING_SECURITY_CONTEXT_KEY);
        if (context instanceof SecurityContext securityContext) {
            return resolveFromAuthentication(securityContext.getAuthentication());
        }
    }
    return null;
}

整个鉴权决策流程如下：

flowchart TD A["WebSocket 握手请求"] --> B{"SecurityContextHolder 有认证信息？"} B -->|"有"| C{"是员工用户？"} B -->|"无"| D["从 HTTP Session 取 SPRING_SECURITY_CONTEXT"] D --> E{"Session 中有 SecurityContext？"} E -->|"有"| C E -->|"无"| F["返回 401 未登录"] C -->|"是"| G{"员工未被禁用？"} C -->|"否"| H["返回 403 非员工用户"] G -->|"是"| I["注入 userId + displayName 握手成功 ✅"] G -->|"否"| H style F fill:#FFCDD2 style H fill:#FFCDD2 style I fill:#C8E6C9

双重取用户的必要性：第 1 步从 SecurityContextHolder 拿是常规路径。但 WebSocket 握手请求和普通 HTTP 请求可能不在同一个线程上下文里，SecurityContextHolder 里是空的。所以我一开始只有策略 1 的时候，握手一直 401，排查了半个小时才发现问题。策略 2 从 Session 里兜底取，才彻底解决。

3.3 连接生命周期——Handler 的三个时刻

TextWebSocketHandler 提供了四个钩子方法，每个都对应连接生命周期的一个节点：

@Component
public class ImWebSocketHandler extends TextWebSocketHandler {

    // 1. 连接建立——握手成功后调用
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        Long userId = currentUserId(session);
        if (userId == null) {
            session.close(CloseStatus.POLICY_VIOLATION);
            return;
        }
        // 注册到 SessionRegistry
        sessionRegistry.register(userId, session);
        // 发送 CONNECTED 确认事件
        session.sendMessage(new TextMessage("""
            {"eventType":"CONNECTED","occurredAt":"%s"}
            """.formatted(LocalDateTime.now())));
    }

    // 2. 收到消息——客户端发来的 PING
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
        String payload = message.getPayload();
        if ("PING".equalsIgnoreCase(payload) || payload.contains("\"PING\"")) {
            session.sendMessage(new TextMessage("""
                {"eventType":"PONG","occurredAt":"%s"}
                """.formatted(LocalDateTime.now())));
        }
    }

    // 3. 连接关闭——浏览器关标签页、断网等
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        unregister(session);  // 从 Registry 中移除
    }

    // 4. 传输异常——比如代理超时、TCP 断开
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) {
        unregister(session);  // 重要：异常也一定要清理！
        if (session.isOpen()) {
            session.close(CloseStatus.SERVER_ERROR);
        }
    }
}

四个钩子的状态流转关系：

flowchart TD A["握手成功"] --> B["afterConnectionEstablished"] B --> B1["注册到 SessionRegistry"] B1 --> B2["发送 CONNECTED 事件"] B2 --> C["在线中 接收 PING → 回复 PONG"] C -->|"正常关闭"| D["afterConnectionClosed"] C -->|"传输异常"| E["handleTransportError"] D --> D1["从 Registry 注销"] D1 --> F["连接结束"] E --> E1["从 Registry 注销"] E1 --> E2{"session 仍 open？"} E2 -->|"是"| E3["主动 close"] E2 -->|"否"| E4["仅注销即可"] E3 --> F E4 --> F style B fill:#C8E6C9 style D fill:#FFF9C4 style E fill:#FFCDD2 style F fill:#E0E0E0

一个关键原则：永远在关闭和异常时都清理 SessionRegistry。

最开始我在 handleTransportError 里只调了 session.close()，没调 unregister(session)。结果 Registry 里存了一堆已断开的脏连接，推送消息时不停地报 IOException: Broken pipe，而前端早就重连了一个新连接。同一个用户 ID 在 Registry 里有三个连接——两个死的，一个活的。这叫"幽灵连接"，不清理的话 Registry 会越来越脏。

3.4 Session Registry——多设备路由

一个员工可能同时在手机浏览器和电脑上打开系统。先看一张多设备路由的整体示意：

Registry 的设计要同时支持：

按用户 ID 推送（广播到该用户的所有连接）
连接去重（同一个连接不能重复注册）
线程安全（WebSocket 连接可以在不同线程建立）

@Component
public class ImWebSocketSessionRegistry {

    // key: 用户 ID  →  value: 该用户的 WebSocket 连接集合
    private final ConcurrentHashMap<Long, CopyOnWriteArraySet<WebSocketSession>> 
        sessionsByUserId = new ConcurrentHashMap<>();

    public void register(Long userId, WebSocketSession session) {
        sessionsByUserId
            .computeIfAbsent(userId, ignored -> new CopyOnWriteArraySet<>())
            .add(session);
    }

    public void unregister(Long userId, WebSocketSession session) {
        CopyOnWriteArraySet<WebSocketSession> sessions = sessionsByUserId.get(userId);
        if (sessions == null) return;
        sessions.remove(session);
        if (sessions.isEmpty()) {
            sessionsByUserId.remove(userId, sessions);
        }
    }

    public int sendToUsers(Collection<Long> userIds, String payload) {
        int deliveredCount = 0;
        for (Long userId : new LinkedHashSet<>(userIds)) {
            CopyOnWriteArraySet<WebSocketSession> sessions = sessionsByUserId.get(userId);
            if (sessions == null || sessions.isEmpty()) continue;
            for (WebSocketSession session : sessions) {
                if (!session.isOpen()) {
                    unregister(userId, session);  // 顺带清理脏连接
                    continue;
                }
                synchronized (session) {
                    session.sendMessage(new TextMessage(payload));
                }
                deliveredCount++;
            }
        }
        return deliveredCount;
    }
}

数据结构的选择：

ConcurrentHashMap：应对多线程并发注册/注销
CopyOnWriteArraySet：连接建立后大部分是读操作（推送），CopyOnWriteArraySet 的读不需要加锁
synchronized(session)：WebSocket 规范不允许并发 sendMessage，同一个 session 必须串行发送

一个小细节：你在 sendToUsers 里能看到 if (!session.isOpen()) { unregister(...); continue; } 这行。这是兜底清理——如果连接异常关闭但 onClose 没触发（比如浏览器直接杀进程），推送时遇到死连接顺手把它清理掉。

四、心跳机制——为什么不依赖 TCP Keep-Alive？

TCP 协议有自己的 Keep-Alive 机制，但默认探测间隔是 2 小时——这对 IM 来说完全没用。你不可能让前端等 2 小时才发现"哦原来我早就断线了"。

我们的心跳方案很简单：

前端：每 30 秒发一条 "PING"
服务端：收到后立即回 {"eventType":"PONG","occurredAt":"..."}

// 前端（useImSocket.ts）
nextSocket.onopen = () => {
    heartbeatTimer = window.setInterval(() => {
        if (nextSocket.readyState === WebSocket.OPEN) {
            nextSocket.send('PING');
        }
    }, 30000);
};

// 后端（ImWebSocketHandler.handleTextMessage）
if ("PING".equalsIgnoreCase(payload) || payload.contains("\"PING\"")) {
    session.sendMessage(new TextMessage("""
        {"eventType":"PONG","occurredAt":"%s"}
        """.formatted(LocalDateTime.now())));
}

为什么条件是 payload.contains("\"PING\"")？这是一个调试兼容的小技巧——前端某些情况下可能发 "PING" 带引号的 JSON 而不是纯文本。包一层 contains 比严格 equals 更鲁棒。

前端探测到断线的逻辑在浏览器原生 WebSocket API 里——如果 30 秒内发 PING 后没收到任何消息，TCP 层就会触发 onclose 回调，前端开始重连。

五、事件类型——统一收口

所有业务模块（消息、群聊、已读、Agent）产生的实时事件都通过 ImWebSocketEvent 这个 record 统一收口：

public record ImWebSocketEvent(
    String eventType,      // 事件类型，比如 "IM_MESSAGE_CREATED"
    Long conversationId,   // 会话 ID
    Long messageId,        // 消息 ID
    Long seqNo,            // 序列号（前端用于去重和排序）
    Long actorUserId,      // 触发事件的用户 ID
    Object payload,        // 事件负载（JSON 对象）
    LocalDateTime occurredAt
) {}

// 当前支持 12 种事件类型
public static final Set<String> OUTBOX_EVENT_TYPES = Set.of(
    IM_MESSAGE_CREATED,              // 新消息
    IM_MESSAGE_EDITED,               // 消息编辑
    IM_MESSAGE_RECALLED,             // 消息撤回
    IM_READ_POSITION_UPDATED,        // 已读位点更新
    IM_CONVERSATION_PREFERENCE_UPDATED,  // 会话偏好变更
    IM_GROUP_PROFILE_UPDATED,        // 群资料变更
    IM_GROUP_MEMBERS_ADDED,          // 新成员入群
    IM_GROUP_MEMBERS_REMOVED,        // 成员被移出
    IM_GROUP_MEMBER_LEFT,            // 成员主动退群
    IM_GROUP_DISMISSED,              // 群解散
    IM_AGENT_MESSAGE_CREATED,        // Agent 回复消息
    IM_AGENT_REPLY_FAILED            // Agent 回复失败
);

用 record 而不是普通 class 有原因——事件是不可变的，发出去就不应该被修改。record 的所有字段都是 final，天然杜绝了"有人改了事件的eventType 但推出去的还是老的"这种 bug。

六、前端 WebSocket 管理——断线重连

前端连接管理我用了一个 composable（useImSocket），核心是三个状态：

连接流程：
  connect() → onopen → wsConnected = true → 启动心跳
                      → 拉取最新消息

断线流程：
  onclose → wsConnected = false → scheduleReconnect()
                                → 3 次以内：递增重试（0.8s / 2s / 3.2s）
                                → 3 次以后：固定 60 秒重试

消息处理：
  onmessage → 解析 JSON → 根据 eventType 分发到不同处理逻辑
                        → IM_MESSAGE_CREATED → 追加到对应会话的消息列表
                        → IM_READ_POSITION_UPDATED → 更新已读位点
                        → ...

重连策略的状态机：

flowchart TD A["connect 发起连接"] --> B{"onopen 成功？"} B -->|"是"| C["wsConnected = true"] C --> D["重置重连次数 = 0"] D --> E["启动 30s 心跳定时器"] E --> F["在线接收消息"] B -->|"否, onclose"| G{"重连次数 ≤ 3？"} F -->|"断线触发 onclose"| G G -->|"是 前3次"| H["延迟递增 0.8s → 2s → 3.2s"] G -->|"否 3次以上"| I["延迟固定 60s"] H --> J["重连次数 + 1"] I --> J J --> A style C fill:#C8E6C9 style H fill:#FFF9C4 style I fill:#FFCDD2

策略设计思路：前 3 次用递增延迟（0.8s → 2s → 3.2s），因为可能是临时网络抖动——比如切 Wi-Fi、短暂断网，几秒内就能恢复。超过 3 次就拉长到 60 秒，这时候更可能是服务端挂了或者网彻底断了，高频重连只会浪费资源。

一个容易被忽略的细节：重连成功后会调 refreshImAfterReconnect()，把断线期间的消息全部重新拉一遍。WebSocket 推送的是增量事件，断线期间的推送会丢失，必须用 HTTP 拉取补齐。

七、总结

核心收获：

握手鉴权要双通道取用户。 不能只信 SecurityContextHolder，WebSocket 握手线程可能不在原 HTTP 请求上线文里。从 Session 兜底取才稳。
生命周期四个钩子，每个都要清理 Registry。 尤其是 handleTransportError——最容易忘记清理的地方，偏偏是最需要清理的地方。
CopyOnWriteArraySet 适合读多写少的场景。 WebSocket 连接的注册/注销远少于推送，用这个数据结构可以做到无锁读。
心跳不能依赖 TCP 自己的 Keep-Alive。 30 秒应用层 PING/PONG 才是 IM 该有的做法。

下一步计划：这篇只搭起了 WebSocket 的骨架——连接建立、用户路由、心跳。下一篇文章会写 Outbox 可靠投递：怎么保证"事务提交了 + WebSocket 推送成功了"才让前端看到消息，失败的时候怎么重试。

个人感悟：做 IM 最怕的不是功能做不出来，而是连接管理没做好导致的"隐形 bug"——消息发了但对方没收到，而且没有任何报错。WebSocket 的可靠性 90% 在连接生命周期管理上，剩下的 10% 才在消息本身。

本机测试 WebSocket 连接：

$ websocat ws://localhost:8080/ws/employee/im --header="Cookie: SESSION=xxx"
{"eventType":"CONNECTED","occurredAt":"2026-05-23T10:30:00"}