Spring Boot使用WebSocket

跟其他http的控制层类似，我们需要实现一个基本的 WebSocket 服务器端点。

PlatformAsyncWebSocket.java

package com.rmeservice.platform.websocket;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;@Slf4j
@Component
@ServerEndpoint(value = "/ws/platformAsync/{userId}")
public class PlatformAsyncWebSocket {// 用来存储每一个客户端对象对应的 WsController 对象private static final Map<String, PlatformAsyncWebSocket> onlineUsers = new ConcurrentHashMap<>();// 声明 Session 对象，通过该对象可以给指定的用户发送请求private Session session;/*** 连接建立时被调用*/@OnOpenpublic void onOpen(Session session, EndpointConfig config) {this.session = session;Map<String, List<String>> requestParameterMap = this.session.getRequestParameterMap();List<String> userIds = requestParameterMap.get("userId");if (userIds!= null &&!userIds.isEmpty()) {String userId = userIds.get(0);onlineUsers.put(userId, this);log.info("用户 {} 建立 WebSocket 连接成功", userId);} else {log.warn("连接建立时未获取到有效的 userId");}}/*** 接收到客户端消息时被调用*/@OnMessagepublic void onMessage(String message, Session session) {try {Map<String, List<String>> requestParameterMap = session.getRequestParameterMap();List<String> userIds = requestParameterMap.get("userId");if (userIds!= null &&!userIds.isEmpty()) {String userId = userIds.get(0);log.info("从用户 {} 接收到消息: {}", userId, message);// 处理接收到的消息，可根据具体业务需求添加逻辑} else {log.warn("接收消息时未获取到有效的 userId");}} catch (Exception e) {log.error("处理客户端消息时发生异常", e);}}/*** 连接被关闭时调用*/@OnClosepublic void onClose(Session session) {try {Map<String, List<String>> requestParameterMap = session.getRequestParameterMap();List<String> userIds = requestParameterMap.get("userId");if (userIds!= null &&!userIds.isEmpty()) {String userId = userIds.get(0);onlineUsers.remove(userId);session.close();log.info("用户 {} 的 WebSocket 连接关闭", userId);} else {log.warn("连接关闭时未获取到有效的 userId");}} catch (Exception e) {log.error("关闭连接时发生异常", e);}}/*** 推送消息,将消息推送给某个指定的用户*/public void sendMsg(String userId, String message) {try {PlatformAsyncWebSocket wsController = onlineUsers.get(userId);if (wsController!= null && wsController.session.isOpen()) {wsController.session.getBasicRemote().sendText(message);log.info("向用户 {} 发送消息: {}", userId, message);} else {log.warn("无法向用户 {} 发送消息，可能是用户未连接或连接已关闭", userId);}} catch (IOException e) {log.error("向用户 {} 发送消息时发生异常: {}", userId, e.getMessage());// 可考虑添加重试机制或通知管理员等操作}}
}

代码解释

类定义和成员变量：
- @Slf4j：使用 Lombok 提供的注解，自动生成 log 日志对象。
- @Component：将类标记为 Spring 组件，由 Spring 容器管理。
- @ServerEndpoint(value = "/ws/platformAsync/{userId}")：将类声明为 WebSocket 服务器端点，客户端可通过 /ws/platformAsync/{userId} 连接。
- private static final Map<String, PlatformAsyncWebSocket> onlineUsers = new ConcurrentHashMap<>();：存储已连接用户及其对应的 PlatformAsyncWebSocket 实例。
- private Session session;：存储与客户端的 WebSocket 会话。
onOpen 方法：
- 当 WebSocket 连接建立时，将 session 存储在成员变量中。
- 从 session 的请求参数中获取 userId，若存在则存储到 onlineUsers 中，并记录连接成功日志；若不存在则记录警告日志。
onMessage 方法：
- 接收到消息时，从 session 的请求参数中获取 userId，若存在则记录接收到的消息及用户信息；若不存在则记录警告日志。
- 对处理消息的逻辑进行异常处理，将异常记录到日志中。
onClose 方法：
- 连接关闭时，从 session 的请求参数中获取 userId，若存在则从 onlineUsers 中移除并关闭 session，记录关闭信息；若不存在则记录警告日志。
- 对关闭连接的操作进行异常处理，将异常记录到日志中。
sendMsg 方法：
- 根据 userId 查找 PlatformAsyncWebSocket 实例，若存在且会话打开，发送消息并记录日志；若不存在或会话关闭，记录警告日志。
- 对发送消息的操作进行异常处理，记录异常信息，可考虑添加重试机制或通知管理员。