我們平時前后臺請求用的最多的就是 HTTP/1.1協(xié)議，它有一個缺陷， 通信只能由客戶端發(fā)起，如果想要不斷獲取服務(wù)器信息就要不斷輪詢發(fā)出請求，那么如果我們需要服務(wù)器狀態(tài)變化的時候能夠主動通知客戶端就需要用到WebSocket了， WebSocket是一種網(wǎng)絡(luò)傳輸協(xié)議，同樣也位于 OSI 模型的應(yīng)用層，建立在傳輸層協(xié)議TCP之上。主要特點(diǎn)是 全雙工 通信允許數(shù)據(jù)在兩個方向上同時傳輸，它在能力上相當(dāng)于兩個單工通信方式的結(jié)合 例如指 A→B 的同時 B→A ，是瞬時同步的 二進(jìn)制幀 采用了二進(jìn)制幀結(jié)構(gòu)，語法、語義與 HTTP 完全不兼容

? ? ? ? 相比 http/2，WebSocket 更側(cè)重于“實(shí)時通信”，而 HTTP/2 更側(cè)重于提高傳輸效率，所以兩者的幀結(jié)構(gòu)也有很大的區(qū)別 不像 HTTP/2 那樣定義流，也就不存在多路復(fù)用、優(yōu)先級等特性 自身就是全雙工，也不需要服務(wù)器推送 協(xié)議名 引入 ws 和 wss 分別代表明文和密文的 websocket 協(xié)議，且默認(rèn)端口使用 80 或 443，幾乎與 http 一致

? ? 如果只是服務(wù)單純的向客戶端推送消息，不涉及到客戶端發(fā)送消息到服務(wù)端，也可以使用Spring?WebFlux技術(shù)實(shí)現(xiàn)，直接建立在當(dāng)前http連接上，本質(zhì)上是保持一個http長連接，適合 簡單的服務(wù)器數(shù)據(jù)推送的場景，使用服務(wù)器推送事件，更輕量更便捷

ps:關(guān)于OSI七層模型可以看我的另一篇文章https://stronger.blog.csdn.net/article/details/127725957

在若依代碼倉庫gitee上的Issues，多次有人提到這個事，例如

• RUOYI-VUE 集成websocket后 使用SecurityUtils獲取用戶信息報錯 · Issue #I49SKT · 若依/RuoYi - Gitee.com
• webSocket接口中無法使用@Autowired和@Resource自動裝配 · Issue #I5VT2N · 若依/RuoYi-Vue - Gitee.com
• 請問vue websocket握手如何在請求頭中攜帶token · Issue #I5KAKK · 若依/RuoYi-Cloud - Gitee.com

作者也在里面明確表示不會將WebSocket集成到框架里來，將會以擴(kuò)展插件的形式給出，傳送門 若依網(wǎng)址 ；下載作者網(wǎng)盤代碼大致如下

配置類WebSocketConfig

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** websocket 配置* * @author ruoyi*/
@Configuration
public class WebSocketConfig
{@Beanpublic ServerEndpointExporter serverEndpointExporter(){return new ServerEndpointExporter();}
}

工具類?SemaphoreUtils

import java.util.concurrent.Semaphore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/*** 信號量相關(guān)處理* * @author ruoyi*/
public class SemaphoreUtils{/*** SemaphoreUtils 日志控制器*/private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreUtils.class);/*** 獲取信號量* * @param semaphore* @return*/public static boolean tryAcquire(Semaphore semaphore){boolean flag = false;try{flag = semaphore.tryAcquire();}catch (Exception e){LOGGER.error("獲取信號量異常", e);}return flag;}/*** 釋放信號量* * @param semaphore*/public static void release(Semaphore semaphore){try{semaphore.release();}catch (Exception e){LOGGER.error("釋放信號量異常", e);}}
}

服務(wù)端類WebSocketServer

import java.util.concurrent.Semaphore;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import com.lxh.demo.util.SemaphoreUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;/*** websocket 消息處理* * @author ruoyi*/
@Component
@ServerEndpoint("/websocket/message")
public class WebSocketServer
{/*** WebSocketServer 日志控制器*/private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);/*** 默認(rèn)最多允許同時在線人數(shù)100*/public static int socketMaxOnlineCount = 100;private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);/*** 連接建立成功調(diào)用的方法*/@OnOpenpublic void onOpen(Session session) throws Exception{boolean semaphoreFlag = false;// 嘗試獲取信號量semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);if (!semaphoreFlag){// 未獲取到信號量LOGGER.error("\n 當(dāng)前在線人數(shù)超過限制數(shù)- {}", socketMaxOnlineCount);WebSocketUsers.sendMessageToUserByText(session, "當(dāng)前在線人數(shù)超過限制數(shù)：" + socketMaxOnlineCount);session.close();}else{// 添加用戶WebSocketUsers.put(session.getId(), session);LOGGER.info("\n 建立連接 - {}", session);LOGGER.info("\n 當(dāng)前人數(shù) - {}", WebSocketUsers.getUsers().size());WebSocketUsers.sendMessageToUserByText(session, "連接成功");}}/*** 連接關(guān)閉時處理*/@OnClosepublic void onClose(Session session){LOGGER.info("\n 關(guān)閉連接 - {}", session);// 移除用戶WebSocketUsers.remove(session.getId());// 獲取到信號量則需釋放SemaphoreUtils.release(socketSemaphore);}/*** 拋出異常時處理*/@OnErrorpublic void onError(Session session, Throwable exception) throws Exception{if (session.isOpen()){// 關(guān)閉連接session.close();}String sessionId = session.getId();LOGGER.info("\n 連接異常 - {}", sessionId);LOGGER.info("\n 異常信息 - {}", exception);// 移出用戶WebSocketUsers.remove(sessionId);// 獲取到信號量則需釋放SemaphoreUtils.release(socketSemaphore);}/*** 服務(wù)器接收到客戶端消息時調(diào)用的方法*/@OnMessagepublic void onMessage(String message, Session session){String msg = message.replace("你", "我").replace("嗎", "");WebSocketUsers.sendMessageToUserByText(session, msg);}
}

WebSocketUsers工具類

import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import javax.websocket.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/*** websocket 客戶端用戶集* * @author ruoyi*/
public class WebSocketUsers
{/*** WebSocketUsers 日志控制器*/private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);/*** 用戶集*/private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();/*** 存儲用戶** @param key 唯一鍵* @param session 用戶信息*/public static void put(String key, Session session){USERS.put(key, session);}/*** 移除用戶** @param session 用戶信息** @return 移除結(jié)果*/public static boolean remove(Session session){String key = null;boolean flag = USERS.containsValue(session);if (flag){Set<Map.Entry<String, Session>> entries = USERS.entrySet();for (Map.Entry<String, Session> entry : entries){Session value = entry.getValue();if (value.equals(session)){key = entry.getKey();break;}}}else{return true;}return remove(key);}/*** 移出用戶** @param key 鍵*/public static boolean remove(String key){LOGGER.info("\n 正在移出用戶 - {}", key);Session remove = USERS.remove(key);if (remove != null){boolean containsValue = USERS.containsValue(remove);LOGGER.info("\n 移出結(jié)果 - {}", containsValue ? "失敗" : "成功");return containsValue;}else{return true;}}/*** 獲取在線用戶列表** @return 返回用戶集合*/public static Map<String, Session> getUsers(){return USERS;}/*** 群發(fā)消息文本消息** @param message 消息內(nèi)容*/public static void sendMessageToUsersByText(String message){Collection<Session> values = USERS.values();for (Session value : values){sendMessageToUserByText(value, message);}}/*** 發(fā)送文本消息** @param session 緩存* @param message 消息內(nèi)容*/public static void sendMessageToUserByText(Session session, String message){if (session != null){try{session.getBasicRemote().sendText(message);}catch (IOException e){LOGGER.error("\n[發(fā)送消息異常]", e);}}else{LOGGER.info("\n[你已離線]");}}
}

Html 頁面代碼

<!DOCTYPE html>
<html lang="zh" xmlns:th="http://www.thymeleaf.org">
<head><meta charset="utf-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><title>測試界面</title>
</head><body><div><input type="text" style="width: 20%" value="ws://127.0.0.1/websocket/message" id="url"><button id="btn_join">連接</button><button id="btn_exit">斷開</button>
</div>
<br/>
<textarea id="message" cols="100" rows="9"></textarea> <button id="btn_send">發(fā)送消息</button>
<br/>
<br/>
<textarea id="text_content" readonly="readonly" cols="100" rows="9"></textarea>返回內(nèi)容
<br/>
<br/>
<script th:src="@{/js/jquery.min.js}" ></script>
<script type="text/javascript">$(document).ready(function(){var ws = null;// 連接$('#btn_join').click(function() {var url = $("#url").val();ws = new WebSocket(url);ws.onopen = function(event) {$('#text_content').append('已經(jīng)打開連接!' + '\n');}ws.onmessage = function(event) {$('#text_content').append(event.data + '\n');}ws.onclose = function(event) {$('#text_content').append('已經(jīng)關(guān)閉連接!' + '\n');}});// 發(fā)送消息$('#btn_send').click(function() {var message = $('#message').val();if (ws) {ws.send(message);} else {alert("未連接到服務(wù)器");}});//斷開$('#btn_exit').click(function() {if (ws) {ws.close();ws = null;}});})
</script>
</body>
</html>

成功運(yùn)行后，頁面如下

注意此時沒有走用戶認(rèn)證，那么就要對路徑放行，因?yàn)槿粢揽蚣苡玫氖荢pringSecurity，所以找到文件SecurityConfig.java ，進(jìn)行路徑放行

三、用戶認(rèn)證問題

雖然按著上述步驟我們完成了瀏覽器(客戶端)和Java(服務(wù)端)的WebSocket通信，但是我們不能限定哪些用戶可以連接我們的服務(wù)端獲取數(shù)據(jù)，服務(wù)端也不知道應(yīng)該具體給哪些用戶發(fā)送消息，在我們框架之前交互我們是通過瀏覽器傳遞toke 值來實(shí)現(xiàn)用戶身份確認(rèn)的，那么我們的WebSocket可不可以也這樣呢？

很不幸的是 ws連接是無法像http一樣完全自主定義請求頭的,給token認(rèn)證帶來了不便，我們大致可以通過以下集中方式完成用戶認(rèn)證

1、將?token?明文攜帶在?url?中，例如ws://localhost:8080/weggo/websocket/message?Authorization=Bearer+token

2、通過websocket下的子協(xié)議來實(shí)現(xiàn)，Stomp這個協(xié)議來實(shí)現(xiàn)，前端采用SocketJs框架來實(shí)現(xiàn)對應(yīng)定制請求頭。實(shí)現(xiàn)攜帶authorization=Bearer +token 的需求，這樣就可以正常建立連接

3、利用子協(xié)議數(shù)組，將 token 攜帶在 protocols 里，var ws = new WebSocket(url, ["token"]);

這樣后端在 onOpen 事件中，就可以從 server 中讀取 Sec-WebSocket-Protocol 屬性來進(jìn)行 token 的獲取，具體可以參考WebScoket構(gòu)造函數(shù)官方文檔

var aWebSocket = new WebSocket(url [, protocols]);
url
要連接的URL；這應(yīng)該是WebSocket服務(wù)器將響應(yīng)的URL。
protocols 可選
一個協(xié)議字符串或者一個包含協(xié)議字符串的數(shù)組。這些字符串用于指定子協(xié)議，這樣單個服務(wù)器可以實(shí)現(xiàn)多個WebSocket子協(xié)議
（例如，您可能希望一臺服務(wù)器能夠根據(jù)指定的協(xié)議（protocol）處理不同類型的交互）。如果不指定協(xié)議字符串，則假定為空字符串。

protocols對應(yīng)的就是發(fā)起ws連接時, 攜帶在請求頭中的Sec-WebSocket-Protocol屬性, 服務(wù)端可以獲取到此屬性的值用于通信邏輯(即通信子協(xié)議,當(dāng)然用來進(jìn)行token認(rèn)證也是完全沒問題的)，前端人員在請求頭上攜帶sec-websocket-protocol=Bearer +token后臺在請求到達(dá)oauth2之前進(jìn)行攔截，然后將在請求頭上添加Authorization=Bearer +token（key首字母大寫），然后在響應(yīng)頭(respone)上添加sec-websocket-protocol=Bearer +token(不添加會報錯)

方法3部分代碼示例

//前端
var aWebSocket = new WebSocket(url ['用戶token']);//后端
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {//這里就是我們所提交的tokenString submitedToken=session.getHandshakeHeaders().get("sec-websocket-protocol").get(0);//根據(jù)token取得登錄用戶信息（業(yè)務(wù)邏輯根據(jù)你自己的來處理）
}

另外，如果需要在第一次握手前的時候就取得token，只需要在header里面取得就可以啦

@Override
public boolean beforeHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Map<String, Object> map) throws Exception {System.out.println("準(zhǔn)備握手");String submitedToken = serverHttpRequest.getHeaders().get("sec-websocket-protocol")return true;
}

因?yàn)槲业捻?xiàng)目是APP 移動端與服務(wù)端進(jìn)行交互，所以后來選擇了最簡單實(shí)現(xiàn)的方案一

首先要解決的就是在攔截器獲取url 的token 信息，原框架只從head里面獲取，所以需要稍加改動

找到TokenService.java文件里的getToken方法，改成如下，這樣就可以獲取url 中的token 了又不影響原來的Http 請求

 private String getToken(HttpServletRequest request){String token = Optional.ofNullable(request.getHeader(header)).orElse(request.getParameter(header));if (StringUtils.isNotEmpty(token) && token.startsWith(Constants.TOKEN_PREFIX)){token = token.replace(Constants.TOKEN_PREFIX, "");}return token;}

接下來就是需要對我們的WebSocket類進(jìn)行改造了，為了方便閱讀，去除了WebSocketUsers類，添加了類變量webSocketSet來存儲客戶端對象

import com.alibaba.fastjson2.JSON;
import com.tongchuang.common.utils.SecurityUtils;
import com.tongchuang.web.mqtt.domain.DeviceInfo;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Component;import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;/*** websocket 消息處理** @author stronger*/
@Component
@ServerEndpoint("/websocket/message")
public class WebSocketServer {/*========================聲明類變量，意在所有實(shí)例共享=================================================*//*** WebSocketServer 日志控制器*/private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);/*** 默認(rèn)最多允許同時在線人數(shù)100*/public static int socketMaxOnlineCount = 100;private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.SECONDS, 8);/*** concurrent包的線程安全Set，用來存放每個客戶端對應(yīng)的MyWebSocket對象。*/private static final CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();/*** 連接數(shù)*/private static final AtomicInteger count = new AtomicInteger();/*========================聲明實(shí)例變量，意在每個實(shí)例獨(dú)享=======================================================*//*** 與某個客戶端的連接會話，需要通過它來給客戶端發(fā)送數(shù)據(jù)*/private Session session;/*** 用戶id*/private String sid = "";/*** 連接建立成功調(diào)用的方法*/@OnOpenpublic void onOpen(Session session) throws Exception {// 嘗試獲取信號量boolean semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);if (!semaphoreFlag) {// 未獲取到信號量LOGGER.error("\n 當(dāng)前在線人數(shù)超過限制數(shù)- {}", socketMaxOnlineCount);// 給當(dāng)前Session 登錄用戶發(fā)送消息sendMessageToUserByText(session, "當(dāng)前在線人數(shù)超過限制數(shù)：" + socketMaxOnlineCount);session.close();} else {// 返回此會話的經(jīng)過身份驗(yàn)證的用戶，如果此會話沒有經(jīng)過身份驗(yàn)證的用戶,則返回nullAuthentication authentication = (Authentication) session.getUserPrincipal();SecurityUtils.setAuthentication(authentication);String username = SecurityUtils.getUsername();this.session = session;//如果存在就先刪除一個，防止重復(fù)推送消息for (WebSocketServer webSocket : webSocketSet) {if (webSocket.sid.equals(username)) {webSocketSet.remove(webSocket);count.getAndDecrement();}}count.getAndIncrement();webSocketSet.add(this);this.sid = username;LOGGER.info("\n 當(dāng)前人數(shù) - {}", count);sendMessageToUserByText(session, "連接成功");}}/*** 連接關(guān)閉時處理*/@OnClosepublic void onClose(Session session) {LOGGER.info("\n 關(guān)閉連接 - {}", session);// 移除用戶webSocketSet.remove(session);// 獲取到信號量則需釋放SemaphoreUtils.release(socketSemaphore);}/*** 拋出異常時處理*/@OnErrorpublic void onError(Session session, Throwable exception) throws Exception {if (session.isOpen()) {// 關(guān)閉連接session.close();}String sessionId = session.getId();LOGGER.info("\n 連接異常 - {}", sessionId);LOGGER.info("\n 異常信息 - {}", exception);// 移出用戶webSocketSet.remove(session);// 獲取到信號量則需釋放SemaphoreUtils.release(socketSemaphore);}/*** 服務(wù)器接收到客戶端消息時調(diào)用的方法*/@OnMessagepublic void onMessage(String message, Session session) {Authentication authentication = (Authentication) session.getUserPrincipal();LOGGER.info("收到來自" + sid + "的信息:" + message);// 實(shí)時更新this.refresh(sid, authentication);sendMessageToUserByText(session, "我收到了你的新消息哦");}/*** 刷新定時任務(wù)，發(fā)送信息*/private void refresh(String userId, Authentication authentication) {this.start(5000L, task -> {// 判斷用戶是否在線，不在線則不用處理，因?yàn)樵趦?nèi)部無法關(guān)閉該定時任務(wù)，所以通過返回值在外部進(jìn)行判斷。if (WebSocketServer.isConn(userId)) {// 因?yàn)檫@里是長鏈接，不會和普通網(wǎng)頁一樣，每次發(fā)送http 請求可以走攔截器【doFilterInternal】續(xù)約，所以需要手動續(xù)約SecurityUtils.setAuthentication(authentication);// 從數(shù)據(jù)庫或者緩存中獲取信息，構(gòu)建自定義的BeanDeviceInfo deviceInfo = DeviceInfo.builder().Macaddress("de5a735951ee").Imei("351517175516665").Battery("99").Charge("0").Latitude("116.402649").Latitude("39.914859").Altitude("80").Method(SecurityUtils.getUsername()).build();// TODO判斷數(shù)據(jù)是否有更新// 發(fā)送最新數(shù)據(jù)給前端WebSocketServer.sendInfo("JSON", deviceInfo, userId);// 設(shè)置返回值，判斷是否需要繼續(xù)執(zhí)行return true;}return false;});}private void start(long delay, Function<Timeout, Boolean> function) {timer.newTimeout(t -> {// 獲取返回值，判斷是否執(zhí)行Boolean result = function.apply(t);if (result) {timer.newTimeout(t.task(), delay, TimeUnit.MILLISECONDS);}}, delay, TimeUnit.MILLISECONDS);}/*** 判斷是否有鏈接** @return*/public static boolean isConn(String sid) {for (WebSocketServer item : webSocketSet) {if (item.sid.equals(sid)) {return true;}}return false;}/*** 群發(fā)自定義消息* 或者指定用戶發(fā)送消息*/public static void sendInfo(String type, Object data, @PathParam("sid") String sid) {// 遍歷WebSocketServer對象集合，如果符合條件就推送for (WebSocketServer item : webSocketSet) {try {//這里可以設(shè)定只推送給這個sid的，為null則全部推送if (sid == null) {item.sendMessage(type, data);} else if (item.sid.equals(sid)) {item.sendMessage(type, data);}} catch (IOException ignored) {}}}/*** 實(shí)現(xiàn)服務(wù)器主動推送*/private void sendMessage(String type, Object data) throws IOException {Map<String, Object> result = new HashMap<>();result.put("type", type);result.put("data", data);this.session.getAsyncRemote().sendText(JSON.toJSONString(result));}/*** 實(shí)現(xiàn)服務(wù)器主動推送-根據(jù)session*/public static void sendMessageToUserByText(Session session, String message) {if (session != null) {try {session.getBasicRemote().sendText(message);} catch (IOException e) {LOGGER.error("\n[發(fā)送消息異常]", e);}} else {LOGGER.info("\n[你已離線]");}}
}