Springboot如何集成websocket
1. WebSocket概述
我們再日常的web應用開發(fā)過程中,常見的是前端向后端發(fā)起請求,但有的時候需要后端主動給前端發(fā)送消息,這個時候全雙工的websocket即時通訊就閃亮登場了。
- 全雙工通訊模式允許雙方同時進行雙向通訊,如手機通話。
- 半雙工通訊模式允許雙方交替發(fā)送和接收信息,但不能同時通訊,如對講機。
- 單工通訊模式只能單向傳輸信息,不能回復,如廣播電臺。
2. WebSocket原理
WebSocket是一種在單個TCP連接上進行全雙工通信的協(xié)議。
它通過一個簡單的握手過程來建立連接,然后在連接上進行雙向數(shù)據(jù)傳輸。
與傳統(tǒng)的HTTP請求不同,WebSocket連接一旦建立,就可以在客戶端和服務器之間保持打開狀態(tài),直到被任何一方關閉。
WebSocket協(xié)議的核心特點包括:
- 全雙工通信:客戶端和服務器可以同時發(fā)送和接收消息。
- 持久連接:一旦建立連接,就可以持續(xù)進行數(shù)據(jù)交換,無需像HTTP那樣頻繁地建立新的連接。
- 低延遲:由于連接是持久的,數(shù)據(jù)可以幾乎實時地發(fā)送和接收。
- 輕量級協(xié)議:WebSocket協(xié)議的頭部信息非常簡單,減少了數(shù)據(jù)傳輸?shù)拈_銷。
3. Spring Boot集成WebSocket
在Spring Boot中集成WebSocket非常簡單,Spring提供了對WebSocket的原生支持。
以下是一個基本的集成步驟:
3.1 添加依賴
在pom.xml
中添加Spring Boot的WebSocket依賴:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency>
3.2 創(chuàng)建WebSocket配置類
創(chuàng)建一個配置類來啟用和配置WebSocket:
package com.jiayuan.common.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.server.standard.ServerEndpointExporter; import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean; /** * @Title: WebSocketConfig * @Package: com.jiayuan.common.config * @Description: websocket配置 * @Author: xmc * @Date: 創(chuàng)建時間 2024-04-24 */ @Configuration public class WebSocketConfig { /** * 自動注冊使用了@ServerEndpoint注解聲明的Websocket endpoint * * @return */ @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } /** * 通信文本消息和二進制緩存區(qū)大小 * 避免對接 第三方 報文過大時,Websocket 1009 錯誤 * * @return */ @Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); // 在此處設置bufferSize container.setMaxTextMessageBufferSize(10240000); container.setMaxBinaryMessageBufferSize(10240000); container.setMaxSessionIdleTimeout(15 * 60000L); return container; } }
3.3 創(chuàng)建消息處理器
package com.jiayuan.common.config; import cn.hutool.core.collection.CollectionUtil; import cn.hutool.extra.spring.SpringUtil; import cn.hutool.json.JSONUtil; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.baomidou.mybatisplus.core.toolkit.StringUtils; import com.jiayuan.common.redis.RedisCache; import com.jiayuan.modules.critical.dto.CvRecordExtraDTO; import com.jiayuan.modules.critical.dto.SyncRecordDTO; import com.jiayuan.modules.critical.service.CvRecordService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.annotation.Resource; import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.io.IOException; import java.util.List; import java.util.concurrent.ConcurrentHashMap; /** * @Title: WebSocketServer * @Package: com.jiayuan.common.config * @Description: websocket的服務端 * @Author: xmc * @Date: 創(chuàng)建時間 2024-04-24 */ @Component @Slf4j @ServerEndpoint("/api/pushMessage/{userId}") public class WebSocketServer { /** * 靜態(tài)變量,用來記錄當前在線連接數(shù)。應該把它設計成線程安全的。 */ private static int onlineCount = 0; /** * concurrent包的線程安全Set,用來存放每個客戶端對應的WebSocket對象。 */ private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>(); /** * 與某個客戶端的連接會話,需要通過它來給客戶端發(fā)送數(shù)據(jù) */ private Session session; /** * 接收userId */ private String userId = ""; /** * 連接建立成 * 功調(diào)用的方法 */ @OnOpen public void onOpen(Session session, @PathParam("userId") String userId) { this.session = session; this.userId = userId; if (webSocketMap.containsKey(userId)) { webSocketMap.remove(userId); //加入set中 webSocketMap.put(userId, this); } else { //加入set中 webSocketMap.put(userId, this); //在線數(shù)加1 addOnlineCount(); } log.info("用戶連接:" + userId + ",當前在線人數(shù)為:" + getOnlineCount()); sendMessage("連接成功"); } /** * 連接關閉 * 調(diào)用的方法 */ @OnClose public void onClose() { if (webSocketMap.containsKey(userId)) { webSocketMap.remove(userId); //從set中刪除 subOnlineCount(); } log.info("用戶退出:" + userId + ",當前在線人數(shù)為:" + getOnlineCount()); } /** * 收到客戶端消 * 息后調(diào)用的方法 * * @param message 客戶端發(fā)送過來的消息 **/ @OnMessage public void onMessage(String message, Session session) { log.info("用戶消息:" + userId + ",報文:" + message); //可以群發(fā)消息 //消息保存到數(shù)據(jù)庫、redis if (StringUtils.isNotBlank(message)) { try { //解析發(fā)送的報文 JSONObject jsonObject = JSON.parseObject(message); //追加發(fā)送人(防止串改) jsonObject.put("fromUserId", this.userId); String toUserId = jsonObject.getString("toUserId"); //傳送給對應toUserId用戶的websocket if (StringUtils.isNotBlank(toUserId) && webSocketMap.containsKey(toUserId)) { webSocketMap.get(toUserId).sendMessage(message); } else { //否則不在這個服務器上,發(fā)送到mysql或者redis log.error("請求的userId:" + toUserId + "不在該服務器上"); } } catch (Exception e) { e.printStackTrace(); } } } /** * @param session * @param error */ @OnError public void onError(Session session, Throwable error) { log.error("用戶錯誤:" + this.userId + ",原因:" + error.getMessage()); error.printStackTrace(); } /** * 實現(xiàn)服務 * 器主動推送 */ public void sendMessage(String message) { try { this.session.getBasicRemote().sendText(message); } catch (IOException e) { e.printStackTrace(); } } /** * 發(fā)送自定 * 義消息 **/ public static void sendInfo(String message, String userId) { log.info("發(fā)送消息到:" + userId + ",報文:" + message); if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) { webSocketMap.get(userId).sendMessage(message); } else { log.error("用戶" + userId + ",不在線!"); } } /** * 獲得此時的 * 在線人數(shù) * * @return */ public static synchronized int getOnlineCount() { return onlineCount; } /** * 在線人 * 數(shù)加1 */ public static synchronized void addOnlineCount() { WebSocketServer.onlineCount++; } /** * 在線人 * 數(shù)減1 */ public static synchronized void subOnlineCount() { WebSocketServer.onlineCount--; } }
上述代碼可能會有疑問:session
和userId
兩個字段是不是安全的?
多人連接,后面的會不會覆蓋掉前面session
和userId
。
答案是不會的,關于處理器WebSocketServer
我們要明確以下幾點:
- 1.
websocket
是原型模式,@ServerEndpoint
每次建立雙向通信的時候都會創(chuàng)建一個實例 - 2.為什么每次都
@OnOpen
都要檢查webSocketMap.containsKey(userId)
,實際使用的時候發(fā)現(xiàn)偶爾會出現(xiàn)重連失敗或者其他原因?qū)е轮暗?code>session還存在,這時候就需要一個清除動作
3.4 服務器主動給客戶端發(fā)送消息
WebSocketServer.sendInfo("服務器主動給客戶端發(fā)送消息test", "zhangsan");
4. 使用ApiPost測試WebSocket
以下是如何使用ApiPost進行測試的步驟:
1.新建一個websocket測試
2.填寫URL
ws://localhost:8080/cvms-api/api/pushMessage/3
注意以下幾點:
- 協(xié)議是
ws
,加密方式請選擇wss
- 選擇
Raw
- URL的拼接公式如下
# servlet.context-path 這個是application.yml中的配置 ws://ip:port//${servlet.context-path}/注解@ServerEndpoint的值
- 有權限驗證的,比如說
shiro
權限驗證,URL就需要加入白名單
filterMap.put("/api/pushMessage/*", "anon");
總結(jié)
以上為個人經(jīng)驗,希望能給大家一個參考,也希望大家多多支持腳本之家。
相關文章
Java List接口與Iterator接口及foreach循環(huán)使用解析
這篇文章主要介紹了Java List接口與Iterator接口及foreach循環(huán),主要包括List接口與Iterator接口及foreach循環(huán)具體的使用方法和代碼,需要的朋友可以參考下2022-04-04統(tǒng)一建模語言_動力節(jié)點Java學院整理
這篇文章主要介紹了統(tǒng)一建模語言的相關知識,非常不錯,具有參考借鑒價值,需要的的朋友參考下吧2017-06-06如何將Spring Session存儲到Redis中實現(xiàn)持久化
這篇文章主要介紹了如何將Spring Session存儲到Redis中實現(xiàn)持久化,本文通過實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2023-07-07Java線程安全問題小結(jié)_動力節(jié)點Java學院整理
這篇文章主要介紹了Java線程安全問題小結(jié)的相關資料,需要的朋友可以參考下2017-05-05