SpringBoot3整合WebSocket

news2025/7/18 21:39:21

一、WebSocket简介

WebSocket协议是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通信，允许服务器主动向客户端推送数据。
与传统的 HTTP 请求-响应模式不同，WebSocket 在建立连接后，允许服务器和客户端之间进行双向实时通信。

WebSocket主要特点：

建立在 TCP 协议之上。
与 HTTP 协议有着良好的兼容性。
数据格式比较轻量，性能开销小，通信高效。
可以发送文本，也可以发送二进制数据。
没有同源限制，客户端可以与任意服务器通信。
协议标识符是ws（如果加密，则为wss），服务器网址就是 URL。

二、SpringBoot3集成WebSocket

引入 WebSocket依赖：

        <!-- WebSocket依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
            <version>3.4.5</version>
        </dependency>

1、WebSocket配置类

创建 WebSocket 配置类，启用 WebSocket 功能并注册端点。

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Autowired
    private CustomTextWebSocketHandler customTextWebSocketHandler;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        /**
         * addHandler方法参数说明
         * 参数1：添加 WebSocket 处理器
         * 参数2：WebSocket的连接路径、端口为项目端口、路径为自定义。比如：URL = ws://127.0.0.1:38081/ws
         */
        registry.addHandler(customTextWebSocketHandler, "/ws")
                .setAllowedOrigins("*"); // 允许跨域访问
    }

}

2、自定义WebSocket处理器

创建自定义的 WebSocket 处理器，处理消息收发。

@Slf4j
@Component
@RequiredArgsConstructor
public class CustomTextWebSocketHandler extends TextWebSocketHandler {

    private final WebSocketSessionManager sessionManager;

    /**
     * 用于存储WebSocket会话
     */
    private final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String sessionId = session.getId();
        sessionManager.add(session);
        log.info("WebSocket连接建立成功：{}", sessionId);
    }


    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        log.info("WebSocket收到消息 ==> sessionId = {}, payload = {} ", session.getId(), message.getPayload());

        boolean open = session.isOpen();
        if (!open) {
            log.info("WebSocket发送消息失败 ==> sessionId = {} 不在线 ", session.getId());
            return;
        }

        try {
            /**
             * TODO 发送回复消息
             */
            String replyMessage = "服务器 -> 客户端 收到了，消息" + payload;
            session.sendMessage(new TextMessage(replyMessage));
        } catch (IOException e) {
            log.info("WebSocket发送消息异常 ==> sessionId = {}, e = ", session.getId(), e);
        }

    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.info("WebSocket传输错误 ==> sessionId = {}, exception = {} ", session.getId(), exception.getMessage());
        // 传输错误,关闭连接
        this.afterConnectionClosed(session, CloseStatus.PROTOCOL_ERROR);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("WebSocket连接关闭 ==> sessionId = {}, CloseStatus = {} ", session.getId(), status);
        sessionManager.remove(session);
    }

    /**
     * 广播发送消息给所有连接的 WebSocket客户端
     *
     * @param message
     */
    public void sendMessageByBroadcast(String message) {
        sessionManager.getAllSession()
                .stream().filter(WebSocketSession::isOpen)
                .forEach(session -> {
                    try {
                        session.sendMessage(new TextMessage(message));
                    } catch (IOException e) {
                        log.info("WebSocket 广播发送消息异常 ==> sessionId = {}, e = ", session.getId(), e);
                    }
                });
    }

}

3、创建WebSocket会话管理类

自定义创建 WebSocket会话管理类，根据业务我们可以扩展存储不同的信息。

@Slf4j
@Component
public class WebSocketSessionManager {

    /**
     * 用于存储 WebSocket会话
     * key: sessionId
     * value: WebSocketSession
     */
    private final Map<String, WebSocketSession> wsSessionMap = new ConcurrentHashMap<>();

    /**
     * 添加 WebSocket会话
     *
     * @param session
     */
    public void add(WebSocketSession session) {
        wsSessionMap.put(session.getId(), session);
    }

    /**
     * 移除 WebSocket会话
     *
     * @param session
     */
    public void remove(WebSocketSession session) {
        wsSessionMap.remove(session.getId());
    }

    /**
     * 获取 WebSocket会话
     *
     * @param sessionId
     * @return
     */
    public WebSocketSession get(String sessionId) {
        return wsSessionMap.get(sessionId);
    }

    /**
     * 获取所有 WebSocket会话
     */
    public List<WebSocketSession> getAllSession() {
        Collection<WebSocketSession> sessions = wsSessionMap.values();
        return Optional.ofNullable(sessions).orElse(new ArrayList<>()).stream().toList();
    }
}

4、controller

创建测试controller，用来测试消息广播。

@RestController
@RequestMapping("/api/wsTest")
@RequiredArgsConstructor
public class WSTestController {

    private final CustomTextWebSocketHandler customTextWebSocketHandler;

    /**
     * 广播发送消息接口
     * 给所有连接的 WebSocket 客户端
     *
     * @param message
     * @return
     */
    @GetMapping("/sendMessageByBroadcast")
    public ResponseEntity<String> sendMessageByBroadcast(@RequestParam("message") String message) {
        customTextWebSocketHandler.sendMessageByBroadcast(message);
        return ResponseEntity.ok("广播发送消息接口 => 发送成功");
    }


}