WebSocket实现聊天

一、先补个基础：WebSocket 是什么？

在传统的 HTTP 通信中，客户端（比如浏览器）要获取新消息，只能 “主动问” 服务器（比如每隔 30 秒发一次请求查新消息），这种方式实时性差、浪费资源。

而 WebSocket 是一种 “全双工” 通信协议：客户端和服务器只需建立一次连接，之后双方可以随时互相发消息（就像打电话，接通后能随时说话），完美适配 “私信实时收发” 的场景。

这段代码就是 服务器端的 WebSocket 服务实现，负责管理用户连接、接收用户发送的私信、实时推送给接收者。

二、代码结构拆解：每个部分是干嘛的？

先看整体结构：代码用 @ServerEndpoint 注解标记这是一个 WebSocket 服务，包含「连接管理」「消息处理」「异常处理」「主动推送」四大核心功能，还有一些静态变量用于存储状态。

1. 静态变量：存储全局状态

▼
java
复制代码
// 1. 线程池：处理“异步存库”（避免存数据库的耗时操作阻塞实时推送）
private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(10);

// 2. 存储用户连接：key=用户ID，value=WebSocket会话（Session）
// 用 ConcurrentHashMap 是因为多用户并发操作时安全（避免线程问题）
private static final Map<Long, Session> USER_SESSIONS = new ConcurrentHashMap<>();

// 3. 服务层注入：WebSocket 是多实例的，所以用静态变量存储 Service（否则注入会失败）
private static PrivateMessageService privateMessageService;
private static UserService userService;

// 4. 静态注入的 setter 方法：Spring 会调用这个方法给静态 Service 赋值
@Resource
public void setPrivateMessageService(PrivateMessageService service) {
    PrivateMessageWebSocket.privateMessageService = service;
}
@Resource
public void setUserService(UserService service) {
    PrivateMessageWebSocket.userService = service;
}

关键理解：

• WebSocket 实例是 “多例” 的（每个用户连接都会创建一个实例），而 Spring 的 Service 是 “单例” 的，所以不能直接用 @Autowired 注入，必须通过「静态变量 + 静态 setter」的方式注入 Service。
• USER_SESSIONS 是核心：记录当前哪些用户在线（有 WebSocket 连接），后续推送消息时要靠它找到接收者的连接。

2. 核心注解方法：WebSocket 生命周期回调

WebSocket 有固定的生命周期（建立连接→收发消息→关闭连接 / 异常），代码用以下注解方法对应这些生命周期，当触发对应事件时，Spring 会自动调用这些方法。

（1）@OnOpen：用户建立 WebSocket 连接时触发

▼
java
复制代码
/**
 * 用户打开页面、建立WebSocket连接时调用（比如进入私信聊天页）
 * @param session 当前用户的WebSocket会话（相当于“通话通道”，用于后续发消息）
 * @param userId 从URL路径中获取的用户ID（比如连接地址是 ws://xxx/ws/private-message/100，这里userId就是100）
 */
@OnOpen
public void onOpen(Session session, @PathParam("userId") Long userId) {
    // 把“用户ID”和“他的会话”存入全局map，标记该用户已在线
    USER_SESSIONS.put(userId, session);
    log.info("用户[{}]建立WebSocket连接，当前在线数：{}", userId, USER_SESSIONS.size());
}

场景举例：用户 A（ID=100）打开私信页面，前端会发起连接请求 new WebSocket("ws://localhost:8080/ws/private-message/100")，服务器收到后调用 onOpen，把 100 和对应的 session 存到 USER_SESSIONS 里，此时在线数 + 1。

（2）@OnMessage：接收客户端（用户）发送的私信时触发

这是最核心的方法，负责 “接收消息→存数据库→推送给接收者” 三步，代码里已经分了步骤，我们逐行拆：

▼
java
复制代码
/**
 * 用户发送私信时调用（比如用户A给用户B发“你好”，前端会通过WebSocket把消息发给服务器）
 * @param message 前端传过来的消息内容（JSON格式，对应 PrivateMessageSendDTO）
 * @param senderId 发送者ID（从URL路径获取，比如发送者是100）
 */
@OnMessage
public void onMessage(String message, @PathParam("userId") Long senderId) {
    try {
        // 步骤1：解析前端传的JSON消息，转成Java对象（PrivateMessageSendDTO）
        // 比如前端发的JSON是 {"receiverId":200,"content":"你好","msgType":1}，这里会转成DTO
        PrivateMessageSendDTO dto = JSON.parseObject(message, PrivateMessageSendDTO.class);
        Long receiverId = dto.getReceiverId(); // 接收者ID（比如200）

        // 步骤2：校验参数（避免空消息、没填接收者的情况）
        if (ObjectUtils.isEmpty(receiverId) || ObjectUtils.isEmpty(dto.getContent())) {
            log.warn("私信参数无效：senderId={}, message={}", senderId, message);
            return;
        }

        // 步骤3：异步写入数据库（重点！先推送再存库，保证实时性）
        // 为什么用线程池异步？因为存数据库是耗时操作（比如100ms），如果同步执行，会阻塞后续的推送，导致接收者延迟收到消息
        EXECUTOR_SERVICE.submit(() -> {
            // 调用 PrivateMessageService 的方法，把私信存到数据库（包含标记“最新消息”、设置过期时间等逻辑）
            privateMessageService.sendPrivateMessage(senderId, dto);
        });

        // 步骤4：实时推送消息给接收者（如果接收者在线）
        // 从 USER_SESSIONS 里查接收者（比如200）是否有在线连接
        Session receiverSession = USER_SESSIONS.get(receiverId);
        if (receiverSession != null && receiverSession.isOpen()) {
            // 构建推送用的VO（给前端返回的格式，包含发送者昵称、头像等，方便前端显示）
            User sender = userService.getById(senderId); // 查发送者的用户信息
            String senderNickname = sender != null ? sender.getNickname() : "匿名用户";
            String senderAvatar = sender != null ? sender.getAvatar() : "";

            PrivateMessageVO pushVO = new PrivateMessageVO();
            pushVO.setSenderId(senderId);          // 发送者ID
            pushVO.setSenderNickname(senderNickname); // 发送者昵称
            pushVO.setSenderAvatar(senderAvatar);   // 发送者头像
            pushVO.setReceiverId(receiverId);      // 接收者ID
            pushVO.setContent(dto.getContent());    // 消息内容
            pushVO.setMsgType(dto.getMsgType());    // 消息类型（1=文本，2=图片）
            pushVO.setIsRead(0);                   // 初始未读
            pushVO.setSendTime(new Date());         // 发送时间

            // 把VO转成JSON，通过接收者的Session推送给前端
            receiverSession.getBasicRemote().sendText(JSON.toJSONString(pushVO));
            log.info("私信推送成功：senderId={}, receiverId={}", senderId, receiverId);
        } else {
            // 如果接收者不在线（USER_SESSIONS里没有他的Session），只存数据库，等他上线后再补推
            log.info("接收者[{}]不在线，私信将存入数据库", receiverId);
        }

    } catch (Exception e) {
        // 捕获所有异常，避免单个消息处理失败导致整个WebSocket服务崩溃
        log.error("处理私信消息失败：senderId={}, message={}", senderId, message, e);
    }
}

场景举例： 用户 A（100）给用户 B（200）发 “你好”，前端把消息转成 JSON 发给服务器，服务器调用 onMessage：

1. 解析出接收者是 200，内容是 “你好”；
2. 用线程池异步把这条消息存到数据库；
3. 查 USER_SESSIONS，如果 B 在线（有 Session），就把包含 A 昵称、头像的消息推给 B 的前端，B 页面实时显示 “你好”；如果 B 不在线，只存库，等 B 下次上线再补推。

（3）@OnClose：用户断开 WebSocket 连接时触发

▼
java
复制代码
/**
 * 用户关闭页面、断开连接时调用（比如关闭私信页、退出登录）
 * @param userId 断开连接的用户ID
 * @param session 要关闭的会话
 */
@OnClose
public void onClose(@PathParam("userId") Long userId, Session session) {
    // 从 USER_SESSIONS 中移除该用户，标记为离线
    USER_SESSIONS.remove(userId);
    try {
        // 关闭会话（释放资源）
        session.close();
    } catch (IOException e) {
        log.error("关闭会话失败：userId={}", userId, e);
    }
    log.info("用户[{}]断开WebSocket连接，当前在线数：{}", userId, USER_SESSIONS.size());
}

场景举例：用户 A 关闭私信页面，前端会主动断开 WebSocket 连接，服务器调用 onClose，把 A 的 ID 从 USER_SESSIONS 中移除，在线数 - 1。

（4）@OnError：WebSocket 连接异常时触发

▼
java
复制代码
/**
 * 连接出现异常时调用（比如网络断了、前端崩溃）
 * @param userId 异常用户的ID
 * @param session 异常的会话
 * @param throwable 异常信息
 */
@OnError
public void onError(@PathParam("userId") Long userId, Session session, Throwable throwable) {
    log.error("用户[{}]WebSocket连接异常", userId, throwable);
    // 异常时也要移除连接（避免存无效的Session）
    USER_SESSIONS.remove(userId);
    try {
        session.close(); // 关闭异常会话
    } catch (IOException e) {
        log.error("异常关闭会话失败：userId={}", userId, e);
    }
}

作用：处理意外情况（比如用户网络突然断开），避免无效的连接占用服务器资源，同时记录异常日志方便排查问题。

3. 主动推送方法：pushMessage（非回调，手动调用）

▼
java
复制代码
/**
 * 主动推送消息给用户（比如用户重连后，补推他离线时收到的未读消息）
 * @param userId 要推送的用户ID
 * @param vo 要推送的私信VO
 */
public static void pushMessage(Long userId, PrivateMessageVO vo) {
    // 查用户是否在线
    Session session = USER_SESSIONS.get(userId);
    if (session != null && session.isOpen()) {
        try {
            // 推送消息
            session.getBasicRemote().sendText(JSON.toJSONString(vo));
            log.info("主动推送消息给用户[{}]成功", userId);
        } catch (IOException e) {
            log.error("主动推送消息失败：userId={}", userId, e);
        }
    }
}

使用场景：比如用户 B 之前离线，收到了 3 条未读消息，当他再次上线建立 WebSocket 连接后，服务器可以调用这个方法，把这 3 条未读消息主动推送给 B，让 B 一上线就能看到。

三、完整流程串讲：从 “用户发消息” 到 “对方收消息”

用一个具体场景把所有环节串起来，你会更清楚：

假设 用户 A（ID=100）给用户 B（ID=200）发私信 “在吗？”，整个流程如下：

1. 建立连接：
   • A 打开私信页面，前端发起 WebSocket 连接：new WebSocket("ws://localhost:8080/ws/private-message/100")；
   • 服务器调用 @OnOpen，把 100 → A的Session 存到 USER_SESSIONS，在线数 = 1；
   • （如果 B 也打开了页面）B 同样建立连接，服务器把 200 → B的Session 存到 USER_SESSIONS，在线数 = 2。
2. A 发送消息：
   • A 在前端输入 “在吗？”，点击发送；
   • 前端把消息转成 JSON（比如 {"receiverId":200,"content":"在吗？","msgType":1}），通过 A 的 WebSocket 发给服务器；
   • 服务器调用 @OnMessage，解析 JSON 得到 DTO，校验参数（接收者 200 非空、内容非空）。
3. 异步存库：
   • 服务器用 EXECUTOR_SERVICE 线程池，异步调用 privateMessageService.sendPrivateMessage(100, dto)；
   • Service 层会做两件事：① 把之前 A 和 B 会话的 “最新消息” 标记为旧（isLatest=0）；② 插入这条新消息（isLatest=1，过期时间 = 当前 + 30 天，isRead=0）。
4. 实时推送给 B：
   • 服务器从 USER_SESSIONS 中查 200 的 Session（如果 B 在线，能查到）；
   • 构建 PrivateMessageVO（包含 A 的昵称、头像、消息内容等），转成 JSON 推给 B 的 Session；
   • B 的前端收到 JSON，解析后在页面上显示 “A 发来消息：在吗？”。
5. 如果 B 不在线：
   • 服务器查不到 200 的 Session，只执行 “异步存库”，不推送；
   • 等 B 下次上线建立连接后，服务器可以调用 pushMessage(200, 未读消息VO)，把离线时的消息补推给 B。
6. A 关闭页面：
   • 前端断开连接，服务器调用 @OnClose，从 USER_SESSIONS 中移除 100，在线数 = 1（只剩 B）。

四、关键细节：为什么要这么设计？

1. 为什么用线程池异步存库？ 存数据库是耗时操作（比如 IO 读写需要 50-100ms），如果同步执行，会阻塞 @OnMessage 方法，导致推送消息延迟。用线程池异步处理，能让 “推送消息” 优先执行，保证实时性。
2. 为什么用 ConcurrentHashMap 存连接？ 多用户会并发建立 / 断开连接（比如同时有 100 个用户上线），普通 HashMap 在并发操作时会出现线程安全问题（比如死循环），ConcurrentHashMap 是线程安全的，适合这种场景。
3. 为什么 Service 要用静态注入？ WebSocket 是 “多实例” 的（每个用户连接都会 new 一个 PrivateMessageWebSocket 对象），而 Spring 的 Service 是 “单例” 的（整个应用只有一个 PrivateMessageService 实例）。如果用普通 @Autowired 注入，每个 WebSocket 实例都会拿到一个新的 Service 实例，这不符合 Spring 设计，还可能导致数据不一致。所以用 “静态变量 + 静态 setter” 的方式，让所有 WebSocket 实例共享同一个 Service 实例。
4. 为什么要标记 “最新消息”（isLatest=1）？ 会话列表（比如 “与 B 的对话”“与 C 的对话”）只需要显示每条对话的最新一条消息，不需要查所有历史消息。存库时标记 isLatest=1，后续查会话列表时只查 isLatest=1 的数据，能大幅优化查询性能。

五、前端怎么配合？

服务器端的 WebSocket 服务需要前端配合才能工作，前端核心代码大概是这样（以 JavaScript 为例）：

▼
javascript
复制代码
// 1. 建立连接（userId 是当前登录用户的ID，比如100）
const userId = 100;
const ws = new WebSocket(`ws://localhost:8080/ws/private-message/${userId}`);

// 2. 连接成功时触发
ws.onopen = () => {
  console.log("WebSocket连接成功");
};

// 3. 接收服务器推送的消息（比如收到A发来的“在吗？”）
ws.onmessage = (event) => {
  const messageVO = JSON.parse(event.data); // 解析服务器推送的VO
  console.log("收到消息：", messageVO);
  // 把消息显示到页面上（比如添加到聊天记录列表）
  renderMessage(messageVO);
};

// 4. 发送消息（给接收者200发“在吗？”）
function sendMessage() {
  const dto = {
    receiverId: 200,
    content: "在吗？",
    msgType: 1 // 文本类型
  };
  ws.send(JSON.stringify(dto)); // 转成JSON发给服务器
}

// 5. 连接关闭时触发
ws.onclose = () => {
  console.log("WebSocket连接关闭");
  // 可选：自动重连（比如网络断了后重试）
  reconnect();
};

// 6. 连接异常时触发
ws.onerror = (error) => {
  console.error("WebSocket异常：", error);
};

总结

这段代码的核心是 “用 WebSocket 实现私信的实时收发”，通过 @ServerEndpoint 定义服务端点，用 @OnOpen/@OnMessage/@OnClose/@OnError 处理连接生命周期，用 USER_SESSIONS 管理在线用户，用线程池异步存库保证实时性，最终实现 “用户发消息→服务器实时推送给接收者” 的完整流程。

如果还有某个细节没懂（比如 Service 层的 sendPrivateMessage 逻辑、前端重连机制），可以再针对性问～