SpringBoot3整合WebSocket詳細(xì)指南

更新時(shí)間：2024年12月16日 10:09:14 作者：CoderJia_

SpringBoot 3 整合 WebSocket 提供了一種高效的實(shí)時(shí)通信解決方案,通過(guò)本文的配置和示例,可以快速實(shí)現(xiàn),感興趣的哦朋友跟隨小編一起看看吧

1. 什么是WebSocket?

WebSocket 是一種網(wǎng)絡(luò)通信協(xié)議，提供全雙工通信通道，使服務(wù)器可以主動(dòng)向客戶(hù)端推送數(shù)據(jù)。與傳統(tǒng)的 HTTP 請(qǐng)求-響應(yīng)模式不同，WebSocket 在建立連接后，允許服務(wù)器和客戶(hù)端之間進(jìn)行雙向?qū)崟r(shí)通信。

主要特點(diǎn)：

建立在 TCP 協(xié)議之上
與 HTTP 協(xié)議有良好的兼容性
數(shù)據(jù)格式輕量，性能開(kāi)銷(xiāo)小
可以發(fā)送文本和二進(jìn)制數(shù)據(jù)
沒(méi)有同源限制，客戶(hù)端可以與任意服務(wù)器通信

2. 環(huán)境準(zhǔn)備

2.1 項(xiàng)目依賴(lài)

首先在pom.xml中添加必要的依賴(lài)：

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.2.5</version>
</parent>
<dependencies>
    <!-- WebSocket依賴(lài) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>
    <!-- Web依賴(lài) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- Lombok依賴(lài)（可選） -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

3. WebSocket配置

3.1 WebSocket配置類(lèi)

創(chuàng)建 WebSocket 配置類(lèi)，啟用 WebSocket 功能并注冊(cè)端點(diǎn)：

package com.coderjia.boot3websocket.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
/**
 * @author CoderJia
 * @create 2024/12/15 下午 08:11
 * @Description
 **/
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(webSocketHandler(), "/websocket")
                .setAllowedOrigins("*"); // 允許跨域訪問(wèn)
    }
    @Bean
    public WebSocketHandler webSocketHandler() {
        // 使用自定義的WebSocket處理器
        return new CustomWebSocketHandler();
    }
}

3.2 自定義WebSocket處理器

創(chuàng)建自定義的 WebSocket 處理器，處理消息收發(fā)：

package com.coderjia.boot3websocket.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
 * @author CoderJia
 * @create 2024/12/15 下午 08:21
 * @Description
 **/
@Component
@Slf4j
public class CustomWebSocketHandler extends TextWebSocketHandler {
    // 用于存儲(chǔ)WebSocket會(huì)話(huà)
    private final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>();
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String sessionId = session.getId();
        sessions.put(sessionId, session);
        log.info("WebSocket連接建立成功：{}", sessionId);
    }
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        log.info("收到消息：{}", payload);
        // 發(fā)送回復(fù)消息
        String replyMessage = "服務(wù)器收到消息：" + payload;
        session.sendMessage(new TextMessage(replyMessage));
    }
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        String sessionId = session.getId();
        sessions.remove(sessionId);
        log.info("WebSocket連接關(guān)閉：{}", sessionId);
    }
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.error("WebSocket傳輸錯(cuò)誤", exception);
    }
    // 廣播消息給所有連接的客戶(hù)端
    public void broadcastMessage(String message) {
        sessions.values().forEach(session -> {
            try {
                session.sendMessage(new TextMessage(message));
            } catch (IOException e) {
                log.error("廣播消息失敗", e);
            }
        });
    }
}

4. 控制器

創(chuàng)建 REST 控制器，用于測(cè)試消息廣播：

@RestController
@RequestMapping("/api/websocket")
public class WebSocketController {
    private final CustomWebSocketHandler webSocketHandler;
    public WebSocketController(CustomWebSocketHandler webSocketHandler) {
        this.webSocketHandler = webSocketHandler;
    }
    @PostMapping("/broadcast")
    public ResponseEntity<String> broadcastMessage(@RequestBody String message) {
        webSocketHandler.broadcastMessage(message);
        return ResponseEntity.ok("消息廣播成功");
    }
}

5. 前端實(shí)現(xiàn)

5.1 HTML頁(yè)面

<!DOCTYPE html>
<html lang="en">
<head>
    <title>WebSocket測(cè)試</title>
</head>
<body>
    <div>
        <h2>WebSocket測(cè)試頁(yè)面</h2>
        <div>
            <input type="text" id="messageInput" placeholder="輸入消息">
            <button onclick="sendMessage()">發(fā)送</button>
        </div>
        <div id="messages" style="margin-top: 20px;"></div>
    </div>
    <script>
        let ws = null;
        function connect() {
            ws = new WebSocket('ws://localhost:8080/websocket');
            ws.onopen = function() {
                console.log('WebSocket連接已建立');
                appendMessage('系統(tǒng)消息：連接已建立');
            };
            ws.onmessage = function(event) {
                appendMessage('收到消息：' + event.data);
            };
            ws.onclose = function() {
                console.log('WebSocket連接已關(guān)閉');
                appendMessage('系統(tǒng)消息：連接已關(guān)閉');
            };
            ws.onerror = function(error) {
                console.error('WebSocket錯(cuò)誤：', error);
                appendMessage('系統(tǒng)消息：連接發(fā)生錯(cuò)誤');
            };
        }
        function sendMessage() {
            const messageInput = document.getElementById('messageInput');
            const message = messageInput.value;
            if (ws && message) {
                ws.send(message);
                appendMessage('發(fā)送消息：' + message);
                messageInput.value = '';
            }
        }
        function appendMessage(message) {
            const messagesDiv = document.getElementById('messages');
            const messageElement = document.createElement('div');
            messageElement.textContent = message;
            messagesDiv.appendChild(messageElement);
        }
        // 頁(yè)面加載完成后連接WebSocket
        window.onload = connect;
    </script>
</body>
</html>

6. 測(cè)試WebSocket功能

啟動(dòng) SpringBoot 應(yīng)用
打開(kāi)多個(gè)瀏覽器窗口訪問(wèn) HTML 頁(yè)面
在任意窗口發(fā)送消息，觀察其他窗口是否收到消息
使用 POST 請(qǐng)求測(cè)試廣播功能：

curl -X POST http://localhost:8080/api/websocket/broadcast \
     -H "Content-Type: text/plain" \
     -d "這是一條廣播消息"

7. 進(jìn)階功能

7.1 心跳檢測(cè)

為了保持WebSocket連接的穩(wěn)定性，可以實(shí)現(xiàn)心跳機(jī)制：

    @Scheduled(fixedRate = 10000) // 每10秒發(fā)送一次心跳，需要啟動(dòng)類(lèi)或配置類(lèi)上添加@EnableScheduling
    public void sendHeartbeat() {
        String heartbeat = "heartbeat";
        sessions.values().forEach(session -> {
            try {
                session.sendMessage(new TextMessage(heartbeat));
            } catch (IOException e) {
                log.error("發(fā)送心跳消息失敗", e);
            }
        });
    }

7.2 消息重試機(jī)制

當(dāng)消息發(fā)送失敗時(shí)，實(shí)現(xiàn)重試機(jī)制：

public void sendMessageWithRetry(WebSocketSession session, String message, int maxRetries) {
    int retryCount = 0;
    while (retryCount < maxRetries) {
        try {
            session.sendMessage(new TextMessage(message));
            return;
        } catch (IOException e) {
            retryCount++;
            log.error("消息發(fā)送失敗，嘗試重試 {}/{}", retryCount, maxRetries);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    log.error("消息發(fā)送失敗，達(dá)到最大重試次數(shù)");
}