集成 websocket 的四种方案
生活随笔
收集整理的這篇文章主要介紹了
集成 websocket 的四种方案
小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.
集成 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 自帶的，下面是他們的具體作用。
另外一點(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è)注解連起來看
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，它的方法功能如下
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/");} }說明
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";} }說明
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ā)送出去。
如何選擇
其它
寫完服務(wù)端代碼后想調(diào)試，但是不會(huì)前端代碼怎么辦，點(diǎn)這里，這是一個(gè)在線的 websocket 客戶端，功能完全夠我們調(diào)試了。
參考鏈接
總結(jié)
以上是生活随笔為你收集整理的集成 websocket 的四种方案的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 【Python】Pandas 数据类型概
- 下一篇： 【机器学习】九种顶流回归算法及实例总结