集成 websocket 的四種方案

1. 原生注解

pom.xml

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

WebSocketConfig

package cn.coder4j.study.example.websocket.config;import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.server.standard.ServerEndpointExporter;@Configuration @EnableWebSocket public class WebSocketConfig {@Beanpublic ServerEndpointExporter serverEndpoint() {return new ServerEndpointExporter();} }

說明：

這個(gè)配置類很簡(jiǎn)單，通過這個(gè)配置 spring boot 才能去掃描后面的關(guān)于 websocket 的注解

WsServerEndpoint

package cn.coder4j.study.example.websocket.ws;import org.springframework.stereotype.Component;import javax.websocket.OnClose; import javax.websocket.OnMessage; import javax.websocket.OnOpen; import javax.websocket.Session; import javax.websocket.server.ServerEndpoint; import java.io.IOException; import java.util.HashMap; import java.util.Map;@ServerEndpoint("/myWs") @Component public class WsServerEndpoint {/*** 連接成功** @param session*/@OnOpenpublic void onOpen(Session session) {System.out.println("連接成功");}/*** 連接關(guān)閉** @param session*/@OnClosepublic void onClose(Session session) {System.out.println("連接關(guān)閉");}/*** 接收到消息** @param text*/@OnMessagepublic String onMsg(String text) throws IOException {return "servet 發(fā)送：" + text;} }

說明

這里有幾個(gè)注解需要注意一下，首先是他們的包都在 **javax.websocket **下。并不是 spring 提供的，而 jdk 自帶的，下面是他們的具體作用。

@ServerEndpoint

通過這個(gè) spring boot 就可以知道你暴露出去的 ws 應(yīng)用的路徑，有點(diǎn)類似我們經(jīng)常用的@RequestMapping。比如你的啟動(dòng)端口是8080，而這個(gè)注解的值是ws，那我們就可以通過 ws://127.0.0.1:8080/ws 來連接你的應(yīng)用

@OnOpen

當(dāng) websocket 建立連接成功后會(huì)觸發(fā)這個(gè)注解修飾的方法，注意它有一個(gè) Session 參數(shù)

@OnClose

當(dāng) websocket 建立的連接斷開后會(huì)觸發(fā)這個(gè)注解修飾的方法，注意它有一個(gè) Session 參數(shù)

@OnMessage

當(dāng)客戶端發(fā)送消息到服務(wù)端時(shí)，會(huì)觸發(fā)這個(gè)注解修改的方法，它有一個(gè) String 入?yún)⒈砻骺蛻舳藗魅氲闹?/li>

@OnError

當(dāng) websocket 建立連接時(shí)出現(xiàn)異常會(huì)觸發(fā)這個(gè)注解修飾的方法，注意它有一個(gè) Session 參數(shù)

另外一點(diǎn)就是服務(wù)端如何發(fā)送消息給客戶端，服務(wù)端發(fā)送消息必須通過上面說的 Session 類，通常是在@OnOpen 方法中，當(dāng)連接成功后把 session 存入 Map 的 value，key 是與 session 對(duì)應(yīng)的用戶標(biāo)識(shí)，當(dāng)要發(fā)送的時(shí)候通過 key 獲得 session 再發(fā)送，這里可以通過 session.getBasicRemote_()*.sendText*(_) 來對(duì)客戶端發(fā)送消息。

2. Spring封裝

pom.xml

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

HttpAuthHandler

package cn.coder4j.study.example.websocket.handler;import cn.coder4j.study.example.websocket.config.WsSessionManager; 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.time.LocalDateTime;@Component public class HttpAuthHandler extends TextWebSocketHandler {/*** socket 建立成功事件** @param session* @throws Exception*/@Overridepublic void afterConnectionEstablished(WebSocketSession session) throws Exception {Object token = session.getAttributes().get("token");if (token != null) {// 用戶連接成功，放入在線用戶緩存WsSessionManager.add(token.toString(), session);} else {throw new RuntimeException("用戶登錄已經(jīng)失效!");}}/*** 接收消息事件** @param session* @param message* @throws Exception*/@Overrideprotected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {// 獲得客戶端傳來的消息String payload = message.getPayload();Object token = session.getAttributes().get("token");System.out.println("server 接收到 " + token + " 發(fā)送的 " + payload);session.sendMessage(new TextMessage("server 發(fā)送給 " + token + " 消息 " + payload + " " + LocalDateTime.now().toString()));}/*** socket 斷開連接時(shí)** @param session* @param status* @throws Exception*/@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {Object token = session.getAttributes().get("token");if (token != null) {// 用戶退出，移除緩存WsSessionManager.remove(token.toString());}}}

說明

通過繼承 TextWebSocketHandler 類并覆蓋相應(yīng)方法，可以對(duì) websocket 的事件進(jìn)行處理，這里可以同原生注解的那幾個(gè)注解連起來看

afterConnectionEstablished 方法是在 socket 連接成功后被觸發(fā)，同原生注解里的 @OnOpen 功能

afterConnectionClosed方法是在 socket 連接關(guān)閉后被觸發(fā)，同原生注解里的 @OnClose 功能

handleTextMessage方法是在客戶端發(fā)送信息時(shí)觸發(fā)，同原生注解里的 @OnMessage 功能

WsSessionManager

package cn.coder4j.study.example.websocket.config;import lombok.extern.slf4j.Slf4j; import org.springframework.web.socket.WebSocketSession;import java.io.IOException; import java.util.concurrent.ConcurrentHashMap;@Slf4j public class WsSessionManager {/*** 保存連接 session 的地方*/private static ConcurrentHashMap<String, WebSocketSession> SESSION_POOL = new ConcurrentHashMap<>();/*** 添加 session** @param key*/public static void add(String key, WebSocketSession session) {// 添加 sessionSESSION_POOL.put(key, session);}/*** 刪除 session,會(huì)返回刪除的 session** @param key* @return*/public static WebSocketSession remove(String key) {// 刪除 sessionreturn SESSION_POOL.remove(key);}/*** 刪除并同步關(guān)閉連接** @param key*/public static void removeAndClose(String key) {WebSocketSession session = remove(key);if (session != null) {try {// 關(guān)閉連接session.close();} catch (IOException e) {// todo: 關(guān)閉出現(xiàn)異常處理e.printStackTrace();}}}/*** 獲得 session** @param key* @return*/public static WebSocketSession get(String key) {// 獲得 sessionreturn SESSION_POOL.get(key);} }

說明

這里簡(jiǎn)單通過 **ConcurrentHashMap **來實(shí)現(xiàn)了一個(gè) session 池，用來保存已經(jīng)登錄的 web socket 的 session。前文提過，服務(wù)端發(fā)送消息給客戶端必須要通過這個(gè) session。

MyInterceptor

package cn.coder4j.study.example.websocket.interceptor;import cn.hutool.core.util.StrUtil; import cn.hutool.http.HttpUtil; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; import org.springframework.stereotype.Component; import org.springframework.web.socket.WebSocketHandler; import org.springframework.web.socket.server.HandshakeInterceptor;import java.util.HashMap; import java.util.Map;@Component public class MyInterceptor implements HandshakeInterceptor {/*** 握手前** @param request* @param response* @param wsHandler* @param attributes* @return* @throws Exception*/@Overridepublic boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {System.out.println("握手開始");// 獲得請(qǐng)求參數(shù)HashMap<String, String> paramMap = HttpUtil.decodeParamMap(request.getURI().getQuery(), "utf-8");String uid = paramMap.get("token");if (StrUtil.isNotBlank(uid)) {// 放入屬性域attributes.put("token", uid);System.out.println("用戶 token " + uid + " 握手成功！");return true;}System.out.println("用戶登錄已失效");return false;}/*** 握手后** @param request* @param response* @param wsHandler* @param exception*/@Overridepublic void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {System.out.println("握手完成");}}

說明

通過實(shí)現(xiàn) HandshakeInterceptor 接口來定義握手?jǐn)r截器，注意這里與上面 Handler 的事件是不同的，這里是建立握手時(shí)的事件，分為握手前與握手后，而 Handler 的事件是在握手成功后的基礎(chǔ)上建立 socket 的連接。所以在如果把認(rèn)證放在這個(gè)步驟相對(duì)來說最節(jié)省服務(wù)器資源。它主要有兩個(gè)方法 beforeHandshake 與 **afterHandshake **，顧名思義一個(gè)在握手前觸發(fā)，一個(gè)在握手后觸發(fā)。

WebSocketConfig

package cn.coder4j.study.example.websocket.config;import cn.coder4j.study.example.websocket.handler.HttpAuthHandler; import cn.coder4j.study.example.websocket.interceptor.MyInterceptor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;@Configuration @EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer {@Autowiredprivate HttpAuthHandler httpAuthHandler;@Autowiredprivate MyInterceptor myInterceptor;@Overridepublic void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {registry.addHandler(httpAuthHandler, "myWS").addInterceptors(myInterceptor).setAllowedOrigins("*");} }

說明

通過實(shí)現(xiàn) WebSocketConfigurer 類并覆蓋相應(yīng)的方法進(jìn)行 websocket 的配置。我們主要覆蓋 registerWebSocketHandlers 這個(gè)方法。通過向 WebSocketHandlerRegistry 設(shè)置不同參數(shù)來進(jìn)行配置。其中 addHandler方法添加我們上面的寫的 ws 的 handler 處理類，第二個(gè)參數(shù)是你暴露出的 ws 路徑。addInterceptors添加我們寫的握手過濾器。**setAllowedOrigins("*") **這個(gè)是關(guān)閉跨域校驗(yàn)，方便本地調(diào)試，線上推薦打開。

3. TIO

pom.xml

<dependency><groupId>org.t-io</groupId><artifactId>tio-websocket-spring-boot-starter</artifactId><version>3.5.5.v20191010-RELEASE</version> </dependency>

application.xml

tio:websocket:server:port: 8989

說明

這里只配置了 ws 的啟動(dòng)端口，還有很多配置，可以通過結(jié)尾給的鏈接去尋找

MyHandler

package cn.coder4j.study.example.websocket.handler;import org.springframework.stereotype.Component; import org.tio.core.ChannelContext; import org.tio.http.common.HttpRequest; import org.tio.http.common.HttpResponse; import org.tio.websocket.common.WsRequest; import org.tio.websocket.server.handler.IWsMsgHandler;@Component public class MyHandler implements IWsMsgHandler {/*** 握手** @param httpRequest* @param httpResponse* @param channelContext* @return* @throws Exception*/@Overridepublic HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {return httpResponse;}/*** 握手成功** @param httpRequest* @param httpResponse* @param channelContext* @throws Exception*/@Overridepublic void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {System.out.println("握手成功");}/*** 接收二進(jìn)制文件** @param wsRequest* @param bytes* @param channelContext* @return* @throws Exception*/@Overridepublic Object onBytes(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {return null;}/*** 斷開連接** @param wsRequest* @param bytes* @param channelContext* @return* @throws Exception*/@Overridepublic Object onClose(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {System.out.println("關(guān)閉連接");return null;}/*** 接收消息** @param wsRequest* @param s* @param channelContext* @return* @throws Exception*/@Overridepublic Object onText(WsRequest wsRequest, String s, ChannelContext channelContext) throws Exception {System.out.println("接收文本消息:" + s);return "success";} }

說明

這個(gè)同上個(gè)例子中的 handler 很像，也是通過實(shí)現(xiàn)接口覆蓋方法來進(jìn)行事件處理，實(shí)現(xiàn)的接口是IWsMsgHandler，它的方法功能如下

handshake

在握手的時(shí)候觸發(fā)

onAfterHandshaked

在握手成功后觸發(fā)

onBytes

客戶端發(fā)送二進(jìn)制消息觸發(fā)

onClose

客戶端關(guān)閉連接時(shí)觸發(fā)

onText

客戶端發(fā)送文本消息觸發(fā)

StudyWebsocketExampleApplication#

package cn.coder4j.study.example.websocket;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.tio.websocket.starter.EnableTioWebSocketServer;@SpringBootApplication @EnableTioWebSocketServer public class StudyWebsocketExampleApplication {public static void main(String[] args) {SpringApplication.run(StudyWebsocketExampleApplication.class, args);} }

說明

這個(gè)類的名稱不重要，它其實(shí)是你的 spring boot 啟動(dòng)類，只要記得加上**@EnableTioWebSocketServer**注解就可以了

STOMP

pom.xml

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

WebSocketConfig

package cn.coder4j.study.example.websocket.config;import org.springframework.context.annotation.Configuration; import org.springframework.messaging.simp.config.MessageBrokerRegistry; import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; import org.springframework.web.socket.config.annotation.StompEndpointRegistry; import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;@Configuration @EnableWebSocketMessageBroker public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {@Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {// 配置客戶端嘗試連接地址registry.addEndpoint("/ws").setAllowedOrigins("*").withSockJS();}@Overridepublic void configureMessageBroker(MessageBrokerRegistry registry) {// 設(shè)置廣播節(jié)點(diǎn)registry.enableSimpleBroker("/topic", "/user");// 客戶端向服務(wù)端發(fā)送消息需有/app 前綴registry.setApplicationDestinationPrefixes("/app");// 指定用戶發(fā)送（一對(duì)一）的前綴 /user/registry.setUserDestinationPrefix("/user/");} }

說明

通過實(shí)現(xiàn) WebSocketMessageBrokerConfigurer 接口和加上**@EnableWebSocketMessageBroker**來進(jìn)行 stomp 的配置與注解掃描。

其中覆蓋 registerStompEndpoints 方法來設(shè)置暴露的 stomp 的路徑，其它一些跨域、客戶端之類的設(shè)置。

覆蓋 configureMessageBroker方法來進(jìn)行節(jié)點(diǎn)的配置。

其中 enableSimpleBroker配置的廣播節(jié)點(diǎn)，也就是服務(wù)端發(fā)送消息，客戶端訂閱就能接收消息的節(jié)點(diǎn)。

覆蓋setApplicationDestinationPrefixes方法，設(shè)置客戶端向服務(wù)端發(fā)送消息的節(jié)點(diǎn)。

覆蓋 setUserDestinationPrefix 方法，設(shè)置一對(duì)一通信的節(jié)點(diǎn)。

WSController

package cn.coder4j.study.example.websocket.controller;import cn.coder4j.study.example.websocket.model.RequestMessage; import cn.coder4j.study.example.websocket.model.ResponseMessage; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.handler.annotation.MessageMapping; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.simp.SimpMessagingTemplate; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.ResponseBody;@Controller public class WSController {@Autowiredprivate SimpMessagingTemplate simpMessagingTemplate;@MessageMapping("/hello")@SendTo("/topic/hello")public ResponseMessage hello(RequestMessage requestMessage) {System.out.println("接收消息：" + requestMessage);return new ResponseMessage("服務(wù)端接收到你發(fā)的：" + requestMessage);}@GetMapping("/sendMsgByUser")public @ResponseBodyObject sendMsgByUser(String token, String msg) {simpMessagingTemplate.convertAndSendToUser(token, "/msg", msg);return "success";}@GetMapping("/sendMsgByAll")public @ResponseBodyObject sendMsgByAll(String msg) {simpMessagingTemplate.convertAndSend("/topic", msg);return "success";}@GetMapping("/test")public String test() {return "test-stomp.html";} }

說明

通過 @MessageMapping 來暴露節(jié)點(diǎn)路徑，有點(diǎn)類似 @RequestMapping。注意這里雖然寫的是 hello ，但是我們客戶端調(diào)用的真正地址是**/app/hello**。 因?yàn)槲覀冊(cè)谏厦娴?config 里配置了registry.setApplicationDestinationPrefixes("/app")。

@SendTo這個(gè)注解會(huì)把返回值的內(nèi)容發(fā)送給訂閱了 /topic/hello *的客戶端，與之類似的還有一個(gè)*@SendToUser 只不過他是發(fā)送給用戶端一對(duì)一通信的。這兩個(gè)注解一般是應(yīng)答時(shí)響應(yīng)的，如果服務(wù)端主動(dòng)發(fā)送消息可以通過 simpMessagingTemplate類的convertAndSend方法。注意 simpMessagingTemplate.convertAndSendToUser(token, “/msg”, msg) *，聯(lián)系到我們上文配置的 **registry.setUserDestinationPrefix("/user/"),**這里客戶端訂閱的是*/user/{token}/msg,千萬不要搞錯(cuò)。

Session 共享的問題

上面反復(fù)提到一個(gè)問題就是，服務(wù)端如果要主動(dòng)發(fā)送消息給客戶端一定要用到 session。而大家都知道的是 session 這個(gè)東西是不跨 jvm 的。如果有多臺(tái)服務(wù)器，在 http 請(qǐng)求的情況下，我們可以通過把 session 放入緩存中間件中來共享解決這個(gè)問題，通過 spring session 幾條配置就解決了。但是 web socket 不可以。他的 session 是不能序列化的，當(dāng)然這樣設(shè)計(jì)的目的不是為了為難你，而是出于對(duì) http 與 web socket 請(qǐng)求的差異導(dǎo)致的。
目前網(wǎng)上找到的最簡(jiǎn)單方案就是通過 redis 訂閱廣播的形式，主要代碼跟第二種方式差不多，你要在本地放個(gè) map 保存請(qǐng)求的 session。也就是說每臺(tái)服務(wù)器都會(huì)保存與他連接的 session 于本地。然后發(fā)消息的地方要修改，并不是現(xiàn)在這樣直接發(fā)送，而通過 redis 的訂閱機(jī)制。服務(wù)器要發(fā)消息的時(shí)候，你通過 redis 廣播這條消息，所有訂閱的服務(wù)端都會(huì)收到這個(gè)消息，然后本地嘗試發(fā)送。最后肯定只有有這個(gè)對(duì)應(yīng)用戶 session 的那臺(tái)才能發(fā)送出去。

如何選擇

如果你在使用 tio，那推薦使用 tio 的集成。因?yàn)樗呀?jīng)實(shí)現(xiàn)了很多功能，包括上面說的通過 redis 的 session 共享，只要加幾個(gè)配置就可以了。但是 tio 是半開源，文檔是需要收費(fèi)的。如果沒有使用，那就忘了他。

如果你的業(yè)務(wù)要求比較靈活多變，推薦使用前兩種，更推薦第二種 Spring 封裝的形式。

如果只是簡(jiǎn)單的服務(wù)器雙向通信，推薦 stomp 的形式，因?yàn)樗菀滓?guī)范使用。

其它

websocket 在線驗(yàn)證

寫完服務(wù)端代碼后想調(diào)試，但是不會(huì)前端代碼怎么辦，點(diǎn)這里，這是一個(gè)在線的 websocket 客戶端，功能完全夠我們調(diào)試了。

參考鏈接

SpringBoot 系統(tǒng) - 集成 WebSocket 實(shí)時(shí)通信

WebSocket 的故事（二）—— Spring 中如何利用 STOMP 快速構(gòu)建 WebSocket 廣播式消息模式

[SpringBoot集成WebSocket【基于純H5】進(jìn)行點(diǎn)對(duì)點(diǎn)一對(duì)一]和廣播[一對(duì)多]實(shí)時(shí)推送

Spring Framework 參考文檔（WebSocket STOMP）

Spring Boot中使用WebSocket總結(jié)（一）：幾種實(shí)現(xiàn)方式詳解

Spring Boot 系列 - WebSocket 簡(jiǎn)單使用

tio-websocket-spring-boot-starter

總結(jié)

以上是生活随笔為你收集整理的集成 websocket 的四种方案的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。