Spring Boot 整合原生 WebSocket的實際操作

更新時間：2026年01月08日 09:34:06 作者：JavaGuru_LiuYu

文章介紹了WebSocket協(xié)議的基本概念、特點及在Spring Boot中的整合步驟,通過示例代碼和實際操作,展示了如何配置WebSocket處理器、握手?jǐn)r截器、會話管理和服務(wù)類,感興趣的朋友跟隨小編一起看看吧

1、簡述

1.1 什么是 WebSocket ?

WebSocket是一種在單個TCP連接上進(jìn)行全雙工、雙向通信的網(wǎng)絡(luò)協(xié)議，它實現(xiàn)了瀏覽器與服務(wù)器之間的持久性實時對話。與傳統(tǒng)的HTTP請求-響應(yīng)模式不同，WebSocket在初次握手后，雙方可以隨時主動發(fā)送數(shù)據(jù)。

1.2 WebSocket特點

特點維度	具體說明	帶來的優(yōu)勢與考量
全雙工通信	在單個TCP連接上，客戶端與服務(wù)器可以同時、獨立地發(fā)送和接收數(shù)據(jù)。	實現(xiàn)了真正的實時雙向?qū)υ?，是?gòu)建聊天室、在線游戲、協(xié)同編輯等強交互應(yīng)用的理想?yún)f(xié)議。
獨立協(xié)議	通過一次HTTP Upgrade握手，將連接升級為獨立的 ws或wss（加密）協(xié)議進(jìn)行通信。	握手后切換至專為實時交互設(shè)計的輕量級幀協(xié)議，擺脫了HTTP的無狀態(tài)和頭部冗余，通信效率更高。
持久化連接	連接一旦建立便會一直保持，直到顯式關(guān)閉，期間可進(jìn)行無數(shù)次數(shù)據(jù)交換。	徹底消除了HTTP的重復(fù)連接建立開銷（如TCP握手、SSL協(xié)商），極大降低了延遲，并減少了服務(wù)器資源消耗。
低開銷數(shù)據(jù)幀	數(shù)據(jù)傳輸采用自定義的幀格式，每個消息的協(xié)議頭部額外開銷極小。	特別適合高頻、小數(shù)據(jù)量的通信場景（如心跳、實時坐標(biāo)更新），網(wǎng)絡(luò)利用率極高，延遲極低。
支持二進(jìn)制與文本	可在同一連接中無縫傳輸文本數(shù)據(jù)和二進(jìn)制數(shù)據(jù)（如ArrayBuffer, Blob）	功能全面強大，既能高效傳輸JSON等文本，也能直接處理文件、圖片、音視頻流，無需像SSE那樣進(jìn)行Base64編解碼。
靈活但需自建的API	瀏覽器提供原生的 WebSocket API，包含 onopen, onmessage, send, close 等核心事件和方法。	為雙向?qū)崟r通信提供了標(biāo)準(zhǔn)化的底層構(gòu)建塊，控制力強。

2、Spring Boot 整合 WebSocket

2.1 引入依賴

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2.2 WebSocket 配置類

@Configuration
@EnableWebSocket
@Slf4j
public class WebSocketConfig implements WebSocketConfigurer {
    @Resource
    private MyWebSocketHandler myWebSocketHandler;
    @Resource
    private WebSocketAuthInterceptor authInterceptor;
    @Value("${server.port}")
    private String port;
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry
                // 注冊 WebSocket 處理器并設(shè)置連接路勁
                .addHandler(myWebSocketHandler, "/ceshi")
                // 添加鑒權(quán)攔截器
                .addInterceptors(authInterceptor)
                // 允許跨域（生產(chǎn)需限制）
                .setAllowedOrigins("*");
        log.info("======================================");
        log.info("?? WebSocket 模塊啟動完成");
        log.info("?? WebSocket 地址：ws://localhost:{}/ceshi", port);
        log.info("?? 鑒權(quán)方式：HandshakeInterceptor (userId)");
        log.info("?? 跨域策略：AllowedOrigins = *");
        log.info("======================================");
    }
}

2.3 WebSocket 處理器

@Component
@Slf4j
public class MyWebSocketHandler extends TextWebSocketHandler {
    @Resource
    private WebSocketSessionManager sessionManager;
    /**
     * 建立連接成功后回調(diào)
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        // 從攔截器中獲取 userId
        String userId = (String) session.getAttributes().get("userId");
        // 保存連接
        sessionManager.add(userId, session);
        log.info("用戶上線：{}", userId);
    }
    /**
     * 收到客戶端消息時回調(diào)
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        // 心跳消息
        if ("ping".equals(payload)) {
            session.sendMessage(new TextMessage("pong"));
            return;
        }
        log.info("收到客戶端消息：{}", payload);
    }
    /**
     * 連接關(guān)閉時回調(diào)
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        String userId = (String) session.getAttributes().get("userId");
        // 移除連接
        sessionManager.remove(userId);
        log.info("用戶離線：{}", userId);
    }
}

2.4 WebSocket 握手?jǐn)r截器

@Component
public class WebSocketAuthInterceptor implements HandshakeInterceptor {
    @Override
    public boolean beforeHandshake(@NonNull ServerHttpRequest request,
                                   @NonNull ServerHttpResponse response,
                                   @NonNull WebSocketHandler handler,
                                   @NonNull Map<String, Object> attributes) {
        // 獲取 ws://xxx/ws?userId=1001 中的參數(shù)
        String query = request.getURI().getQuery();
        if (query == null) {
            return false;
        }
        // 簡單解析 userId（生產(chǎn)建議用 token / JWT）
        String userId = Arrays.stream(query.split("&"))
                .filter(s -> s.startsWith("userId="))
                .map(s -> s.replace("userId=", ""))
                .findFirst()
                .orElse(null);
        // 校驗失敗，拒絕建立連接
        if (userId == null) {
            return false;
        }
        // 將 userId 保存到 session 屬性中
        attributes.put("userId", userId);
        return true;
    }
    @Override
    public void afterHandshake(@NonNull ServerHttpRequest request,
                               @NonNull ServerHttpResponse response,
                               @NonNull WebSocketHandler handler,
                               Exception ex) {
        // 握手完成后無需處理
    }
}

2.5 WebSocketMessage 會話管理實體類

@Data
@AllArgsConstructor
@NoArgsConstructor
public class WebSocketMessage {
    /**
     * 消息類型
     * HEARTBEAT：心跳
     * NOTICE：通知類消息
     * DATA：業(yè)務(wù)數(shù)據(jù)
     * ERROR：錯誤提示
     */
    private String type;
    /**
     * 業(yè)務(wù)標(biāo)識
     * 如：workOrder、device、alarm
     * 前端可根據(jù) biz 分模塊處理
     */
    private String biz;
    /**
     * 實際業(yè)務(wù)數(shù)據(jù)
     * 可放 Map / DTO / VO
     */
    private Object data;
}

2.6 WebSocket 連接會話管理

@Component
public class WebSocketSessionManager {
    /**
     * 保存在線 WebSocket 連接
     * key：userId
     * value：WebSocketSession
     */
    public static final Map<String, WebSocketSession> SESSION_MAP = new ConcurrentHashMap<>();
    /**
     * 新用戶上線時調(diào)用
     */
    public void add(String userId, WebSocketSession session) {
        SESSION_MAP.put(userId, session);
    }
    /**
     * 用戶下線 / 斷開連接時調(diào)用
     */
    public void remove(String userId) {
        SESSION_MAP.remove(userId);
    }
    /**
     * 獲取指定用戶的連接
     */
    public WebSocketSession get(String userId) {
        return SESSION_MAP.get(userId);
    }
    /**
     * 獲取所有在線連接
     */
    public Collection<WebSocketSession> all() {
        return SESSION_MAP.values();
    }
    /**
     * 給指定用戶推送消息
     */
    public void sendToUser(String userId, String msg) throws IOException {
        WebSocketSession session = SESSION_MAP.get(userId);
        if (session != null && session.isOpen()) {
            session.sendMessage(new TextMessage(msg));
        }
    }
    /**
     * 廣播消息（慎用）
     */
    public void sendToAll(String msg) throws IOException {
        for (WebSocketSession session : SESSION_MAP.values()) {
            if (session.isOpen()) {
                session.sendMessage(new TextMessage(msg));
            }
        }
    }
}

2.7 WebSocket 服務(wù)類

public interface WebSocketService {
    /**
     * 獲取指定的連接信息
     */
    WebSocketSession get(String userId);
    /**
     * 獲取所有在線連接
     */
    Collection<WebSocketSession> all();
    /**
     * 給指定用戶推送消息
     */
    Boolean sendToUser(String userId, String msg);
    /**
     * 廣播消息（慎用）
     */
    Integer sendToAll(String msg);
}

2.7 WebSocket 服務(wù)實現(xiàn)類

@Service
@Slf4j
public class WebSocketServiceImpl implements WebSocketService {
    @Resource
    WebSocketSessionManager webSocketSessionManager;
    @Override
    public WebSocketSession get(String userId) {
        return webSocketSessionManager.get(userId);
    }
    @Override
    public Collection<WebSocketSession> all() {
        return webSocketSessionManager.all();
    }
    @Override
    public Boolean sendToUser(String userId, String msg) {
        WebSocketSession session = SESSION_MAP.get(userId);
        if (session == null || !session.isOpen()) {
            return false;
        }
        try {
            webSocketSessionManager.sendToUser(userId, msg);
            return true;
        } catch (IOException e) {
            log.error("WebSocket 推送失敗 userId={}", userId, e);
            return false;
        }
    }
    @Override
    public Integer sendToAll(String msg) {
        int count = 0;
        for (WebSocketSession session : SESSION_MAP.values()) {
            if (session.isOpen()) {
                try {
                    webSocketSessionManager.sendToAll(msg);
                    count++;
                } catch (IOException e) {
                    log.warn("廣播發(fā)送失敗", e);
                }
            }
        }
        return count;
    }
}

2.8 Controller 對外接口

@Api(tags = "WebSocket API")
@RestController
@RequestMapping("/websocket")
public class WebSocketController {
    @Resource
    private WebSocketService webSocketService;
    @ApiOperation("獲取指定用戶的 WebSocket 連接信息")
    @GetMapping("/connection")
    public AjaxResult getConnection(@Parameter(name = "userId", description = "用戶ID", required = true, example = "1001") String userId) {
        return AjaxResult.success(webSocketService.get(userId));
    }
    @ApiOperation("獲取所有在線 WebSocket 連接")
    @GetMapping("/connections")
    public AjaxResult getAllConnections() {
        return AjaxResult.success(webSocketService.all());
    }
    @ApiOperation("給指定用戶推送 WebSocket 消息")
    @PostMapping("/send/user")
    public AjaxResult sendToUser(
            @Parameter(name = "userId", description = "用戶ID", required = true, example = "1001") String userId,
            @Parameter(name = "msg", description = "消息內(nèi)容", required = true, example = "這是一條通知") String msg) {
        return webSocketService.sendToUser(userId, msg) ? AjaxResult.success("發(fā)送成功") : AjaxResult.error("用戶不在線或發(fā)送失敗");
    }
    @ApiOperation("廣播 WebSocket 消息（慎用）")
    @PostMapping("/send/all")
    public AjaxResult sendToAll(
            @Parameter(name = "msg", description = "消息內(nèi)容", required = true, example = "系統(tǒng)廣播消息") String msg) {
        int count = webSocketService.sendToAll(msg);
        return AjaxResult.success("廣播完成，發(fā)送給 " + count + " 個在線用戶");
    }
}