SpringBoot實(shí)現(xiàn)websocket服務(wù)端及客戶端的詳細(xì)過程
更新時(shí)間：2024年12月30日 11:32:30 作者：uni可樂
文章介紹了WebSocket通信過程、服務(wù)端和客戶端的實(shí)現(xiàn),以及可能遇到的問題及解決方案,感興趣的朋友一起看看吧
一、WebSocket通信過程
客戶端構(gòu)建一個(gè)websocket實(shí)例，并且為它綁定一個(gè)需要連接到的服務(wù)器地址，當(dāng)客戶端連接服務(wù)端的候，會(huì)向服務(wù)端發(fā)送一個(gè)http get報(bào)文，告訴服務(wù)端需要將通信協(xié)議切換到websocket，服務(wù)端收到http請求后將通信協(xié)議切換到websocket,同時(shí)發(fā)給客戶端一個(gè)響應(yīng)報(bào)文，返回的狀態(tài)碼為101，表示同意客戶端協(xié)議轉(zhuǎn)請求，并轉(zhuǎn)換為websocket協(xié)議。以上過程都是利用http通信完成的，稱之為websocket協(xié)議握手(websocket Protocol handshake)，經(jīng)過握手之后，客戶端和服務(wù)端就建立了websocket連接，以后的通信走的都是websocket協(xié)議了。
二、服務(wù)端實(shí)現(xiàn)
1.pom文件添加依賴
<!--webSocket-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>2.啟用Springboot對WebSocket的支持
package com.lby.websocket.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* @author Liby
* @date 2022-04-25 16:18
* @description:
* @version:
*/
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
3.核心配置：WebSocketServer
因?yàn)閃ebSocket是類似客戶端服務(wù)端的形式(采用ws協(xié)議)，那么這里的WebSocketServer其實(shí)就相當(dāng)于一個(gè)ws協(xié)議的Controller
@ ServerEndpoint 注解是一個(gè)類層次的注解，它的功能主要是將目前的類定義成一個(gè)websocket服務(wù)器端, 注解的值將被用于監(jiān)聽用戶連接的終端訪問URL地址,客戶端可以通過這個(gè)URL來連接到WebSocket服務(wù)器端
新建一個(gè)ConcurrentHashMap webSocketMap 用于接收當(dāng)前userId的WebSocket，方便傳遞之間對userId進(jìn)行推送消息。
下面是具體業(yè)務(wù)代碼：
package org.example.server;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
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.concurrent.ConcurrentHashMap;
/*
啟動(dòng)項(xiàng)目后可通過 http://websocket.jsonin.com/ 進(jìn)行測試
輸入網(wǎng)址ws://localhost:8080/websocket/1211，檢測是否能進(jìn)行連接
*/
@ServerEndpoint(value = "/websocket/{userId}")
@Component
public class WebSocket {
private final static Logger logger = LogManager.getLogger(WebSocket.class);
/**
* 靜態(tài)變量，用來記錄當(dāng)前在線連接數(shù)。應(yīng)該把它設(shè)計(jì)成線程安全的
*/
private static int onlineCount = 0;
/**
* concurrent包的線程安全Map，用來存放每個(gè)客戶端對應(yīng)的MyWebSocket對象
*/
private static ConcurrentHashMap<String, WebSocket> webSocketMap = new ConcurrentHashMap<>();
/**
* 與某個(gè)客戶端的連接會(huì)話，需要通過它來給客戶端發(fā)送數(shù)據(jù)
*/
private Session session;
private String userId;
/**
* 連接建立成功調(diào)用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId) {
this.session = session;
this.userId = userId;
//加入map
webSocketMap.put(userId, this);
addOnlineCount(); //在線數(shù)加1
logger.info("用戶{}連接成功,當(dāng)前在線人數(shù)為{}", userId, getOnlineCount());
try {
sendMessage(String.valueOf(this.session.getQueryString()));
} catch (IOException e) {
logger.error("IO異常");
}
}
/**
* 連接關(guān)閉調(diào)用的方法
*/
@OnClose
public void onClose() {
//從map中刪除
webSocketMap.remove(userId);
subOnlineCount(); //在線數(shù)減1
logger.info("用戶{}關(guān)閉連接！當(dāng)前在線人數(shù)為{}", userId, getOnlineCount());
}
/**
* 收到客戶端消息后調(diào)用的方法
*
* @param message 客戶端發(fā)送過來的消息
*/
@OnMessage
public void onMessage(String message, Session session) {
logger.info("來自客戶端用戶：{} 消息:{}",userId, message);
//群發(fā)消息
for (String item : webSocketMap.keySet()) {
try {
webSocketMap.get(item).sendMessage(message);
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 發(fā)生錯(cuò)誤時(shí)調(diào)用
*
*/
@OnError
public void onError(Session session, Throwable error) {
logger.error("用戶錯(cuò)誤:" + this.userId + ",原因:" + error.getMessage());
error.printStackTrace();
}
/**
* 向客戶端發(fā)送消息
*/
public void sendMessage(String message) throws IOException {
this.session.getBasicRemote().sendText(message);
//this.session.getAsyncRemote().sendText(message);
}
/**
* 通過userId向客戶端發(fā)送消息
*/
public void sendMessageByUserId(String userId, String message) throws IOException {
logger.info("服務(wù)端發(fā)送消息到{},消息：{}",userId,message);
if(StrUtil.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
webSocketMap.get(userId).sendMessage("hello");
}else{
logger.error("用戶{}不在線",userId);
}
}
/**
* 群發(fā)自定義消息
*/
public void sendInfo(String message) throws IOException {
for (String item : webSocketMap.keySet()) {
try {
webSocketMap.get(item).sendMessage(message);
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static synchronized int getOnlineCount() {
return onlineCount;
}
public static synchronized void addOnlineCount() {
WebSocket.onlineCount++;
}
public static synchronized void subOnlineCount() {
WebSocket.onlineCount--;
}
}
三、客戶端實(shí)現(xiàn)
1.pom文件添加依賴
<dependency>
<groupId>org.java-websocket</groupId>
<artifactId>Java-WebSocket</artifactId>
<version>1.5.3</version>
</dependency>2.客戶端實(shí)現(xiàn)代碼
package org.example;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import java.net.URI;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* @author kele
* @date 2024/2/19
**/
@Slf4j
public class MyWebSocketClient extends WebSocketClient {
public MyWebSocketClient(URI serverUri) {
super(serverUri);
}
@SneakyThrows
@Override
public void onOpen(ServerHandshake data) {
try {
log.info("WebSocket連接已打開。");
}catch (Exception e){
log.error("onOpen error :{}",e.getMessage());
}
}
@SneakyThrows
@Override
public void onMessage(String message) {
try {
if (message != null && !message.isEmpty()) {
log.info("收到消息: {}",message);
}
}catch (Exception e){
log.error("onMessage error : {}",message);
}
}
@Override
public void onClose(int code, String reason, boolean remote) {
log.info("WebSocket連接已關(guān)閉。");
}
@Override
public void onError(Exception ex) {
log.info("WebSocket連接發(fā)生錯(cuò)誤:{}", ex.getMessage());
}
/**
* 連接定時(shí)檢查
*/
public void startReconnectTask(long delay, TimeUnit unit) {
System.out.println("WebSocket 心跳檢查");
log.info("WebSocket 心跳檢查");
// 以下為定時(shí)器,建議使用自定義線程池，或交給框架處理（spring）
ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
executorService.scheduleWithFixedDelay(() -> {
// 檢查邏輯：判斷當(dāng)前連接是否連通
if (!this.isOpen()) {
System.out.println("WebSocket 開始重連......");
log.info("WebSocket 開始重連......");
// 重置連接
this.reconnect();
// 以下為錯(cuò)誤示范
//this.close();
//this.connect();
}
}, 0, delay, unit);
}
}3.開啟客戶端連接服務(wù)，并開啟定時(shí)檢查websocket連接
package org.example;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.net.URI;
import java.util.concurrent.TimeUnit;
@Slf4j
@Component
public class Init implements Runnable {
public static MyWebSocketClient myWebSocketClient;
@PostConstruct
public void run () {
try {
//啟動(dòng)連接
log.info("連接websocket服務(wù)端");
log.info("項(xiàng)目啟動(dòng)");
// 服務(wù)地址
URI uri = new URI("ws://127.0.0.1:8077/websocket/123");
log.info("服務(wù)地址 -{}", uri);
// 創(chuàng)建客戶端
myWebSocketClient = new MyWebSocketClient(uri);
// 建立連接
myWebSocketClient.connect();
// 開啟 定時(shí)檢查
myWebSocketClient.startReconnectTask(5, TimeUnit.SECONDS);
}catch (Exception e){
e.printStackTrace();
}
}
}四、可能遇到的問題及解決方案
1.websocket客戶端重連機(jī)制
注意引入Java-websocket的版本要高于1.5.0才有reconnect（）方法
2.在使用websocket連接傳輸Base64編碼時(shí)出現(xiàn)錯(cuò)誤
原因：Websocket中maxMessageSize默認(rèn)是8K，接收的圖片大于8K導(dǎo)致接收失敗，解決方法：適當(dāng)加大maxMessageSize。
參考資料：
@OnMessage(maxMessageSize = 10240000)
public void onMessage(byte[] message) throws IOException {
log.info("收到圖片了");
String str = Base64.encodeBase64String(message); // 圖片base64
log.info(str.substring(0, 100));
}到此這篇關(guān)于SpringBoot實(shí)現(xiàn)websocket服務(wù)端及客戶端的文章就介紹到這了,更多相關(guān)SpringBoot websocket服務(wù)端及客戶端內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Spring事務(wù)&Spring整合MyBatis的兩種方式
這篇文章主要介紹了Spring事務(wù)&Spring整合MyBatis的兩種方式,本文結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2023-02-02
這篇文章主要介紹了關(guān)于Netty--Http請求處理方式，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-06-06
Java實(shí)現(xiàn)將CSV轉(zhuǎn)為Excel的示例代碼
CSV(Comma?Separated?Values)文件是一種純文本文件，包含用逗號(hào)分隔的數(shù)據(jù)，常用于將數(shù)據(jù)從一個(gè)應(yīng)用程序?qū)牖驅(qū)С龅搅硪粋€(gè)應(yīng)用程序。本文將利用Java實(shí)現(xiàn)CSV轉(zhuǎn)為Excel，感興趣的可以了解一下2022-03-03
文章簡要介紹Nacos作為阿里巴巴開源的微服務(wù)管理組件,涵蓋其核心功能及單機(jī)模式部署步驟：下載穩(wěn)定版、配置MySQL、執(zhí)行數(shù)據(jù)庫腳本、啟動(dòng)服務(wù)并查看日志,最后通過指定地址和賬號(hào)登錄控制臺(tái)2025-07-07
MyBatisCodeHelperPro插件下載及使用教程詳解
這篇文章主要介紹了MyBatis CodeHelperPro插件使用教程，本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友參考下吧2020-09-09
java操作mongodb之多表聯(lián)查的實(shí)現(xiàn)($lookup)
這篇文章主要介紹了java操作mongodb之多表聯(lián)查的實(shí)現(xiàn)($lookup)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
Intellij無法創(chuàng)建java文件解決方案
這篇文章主要介紹了Intellij無法創(chuàng)建java文件解決方案,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-10-10
SpringBoot整合Mybatis-Plus實(shí)現(xiàn)微信注冊登錄的示例代碼
微信是不可或缺的通訊工具,本文主要介紹了SpringBoot整合Mybatis-Plus實(shí)現(xiàn)微信注冊登錄的示例代碼,文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的可以了解一下2024-02-02
Redis 是一個(gè)高性能的key-value數(shù)據(jù)庫。在部分場合可以對關(guān)系數(shù)據(jù)庫起到很好的補(bǔ)充作用。它提供了Java，C/C++，C#，PHP，JavaScript，Perl，Object-C，Python，Ruby，Erlang等客戶端使用很方便2021-06-06
這篇文章主要為大家介紹了resty緩存技術(shù)的設(shè)計(jì)及使用示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步2022-03-03