SpringBoot+WebSocket實(shí)現(xiàn)IM及時(shí)通訊的代碼示例
1、Springboot集成Websocket
集成分為三步：添加依賴(lài)、增加配置類(lèi)和消息核心類(lèi)、前端集成。
1.1、添加依賴(lài)
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
<version>2.1.13.RELEASE</version>
</dependency>
1.2、增加WebSocket配置類(lèi)
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* WebSocket配置
*/
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
1.3、增加消息核心類(lèi)WebSocketServer
@ServerEndpoint("/trtc/websocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {
// 消息存儲(chǔ)
private static MessageStore messageStore;
// 消息發(fā)送
private static MessageSender messageSender;
public static void setMessageStore(MessageStore messageStore) {
WebSocketServer.messageStore = messageStore;
}
public static void setMessageSender(MessageSender messageSender) {
WebSocketServer.messageSender = messageSender;
}
/**
* 連接建立成功調(diào)用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId) {
messageStore.saveSession(session);
}
/**
* 連接關(guān)閉調(diào)用的方法
*/
@OnClose
public void onClose(Session session, @PathParam("userId") String userId) {
messageStore.deleteSession(session);
}
/**
* 收到客戶(hù)端消息后調(diào)用的方法
*
* @ Param message 客戶(hù)端發(fā)送過(guò)來(lái)的消息
*/
@OnMessage
public void onMessage(String message, Session session) throws Exception {
log.warn("=========== 收到來(lái)自窗口" + session.getId() + "的信息:" + message);
handleTextMessage(session, new TextMessage(message));
}
/**
* @param session
* @param error
*/
@OnError
public void onError(Session session, @PathParam("userId") String userId, Throwable error) {
log.error("=========== 發(fā)生錯(cuò)誤");
error.printStackTrace();
// msgStore.deleteSession(session);
}
public void handleTextMessage(Session session, TextMessage message) throws Exception {
log.warn("=========== Received message: {}", message.getPayload());
}
}
1.4、前端頁(yè)面加入socket
<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/html">
<head>
<title>WebSocket Example</title>
</head>
<body>
登錄用戶(hù)ID：<input type="text" id="sendUserId" /></br>
接受用戶(hù)ID：<input type="text" id="receivedUserId" /></br>
發(fā)送消息內(nèi)容：<input type="text" id="messageInput" /></br>
接受消息內(nèi)容：<input type="text" id="messageReceive" /></br>
<button onclick="sendMessage()">Send</button>
<script>
var socket = new WebSocket("ws://localhost:8080/trtc/websocket/aaa");
var roomId = "123456";
// 隨機(jī)產(chǎn)出六位數(shù)字
var sendUserId = Math.floor(Math.random() * 1000000);
document.getElementById("sendUserId").value = sendUserId;
var messageReceive = document.getElementById("messageReceive");
socket.onopen = function (event) {
console.log("WebSocket is open now.");
let loginInfo = {
msgType: 2, //登錄消息
sendUserId: sendUserId,
bizType: 1, //業(yè)務(wù)類(lèi)型
bizOptModule: 1, //業(yè)務(wù)模塊
roomId: roomId,
msgBody: {},
};
socket.send(JSON.stringify(loginInfo));
};
socket.onmessage = function (event) {
var message = event.data;
console.log("Received message: " + message);
messageReceive.value = message;
};
socket.onclose = function (event) {
console.log("WebSocket is closed now.");
};
function sendMessage() {
var message = document.getElementById("messageInput").value;
var receivedUserId = document.getElementById("receivedUserId").value;
let operateInfo = {
msgType: 100, //業(yè)務(wù)消息
sendUserId: sendUserId,
bizType: 1, //業(yè)務(wù)類(lèi)型
bizOptModule: 1, //業(yè)務(wù)模塊
roomId: roomId,
receivedUserId: receivedUserId,
msgBody: {
operateType: 1, //操作類(lèi)型：禁言
operateContent: "1",
},
};
socket.send(JSON.stringify(operateInfo));
}
setInterval(() => {
socket.send("ping");
}, 30000);
</script>
</body>
</html>
2、小型及時(shí)通訊包含的模塊
以上只是集成了websocket框架，實(shí)現(xiàn)了基本的全雙工通信，服務(wù)器和客戶(hù)端都可以同時(shí)發(fā)送和接收數(shù)據(jù)。要想實(shí)現(xiàn)一些小型完整的及時(shí)通訊，還需要具備以下幾個(gè)核心模塊。也通過(guò)一張流程圖展示了其中過(guò)程。
2.1、消息對(duì)象模型
組織消息內(nèi)容，比如消息類(lèi)型、發(fā)送者用戶(hù)ID、接受者用戶(hù)ID、具體的消息體等。比如：
public class SocketMsg<T> {
/**
* 消息類(lèi)型：1心跳 2登錄 3業(yè)務(wù)操作
*/
private Integer msgType;
/**
* 發(fā)送者用戶(hù)ID
*/
private String sendUserId;
/**
* 接受者用戶(hù)ID
*/
private String receivedUserId;
/**
* 業(yè)務(wù)類(lèi)型
*/
private Integer bizType;
/**
* 業(yè)務(wù)操作模塊
*/
private Integer bizOptModule;
/**
* 消息內(nèi)容
*/
private T msgBody;
}
2.2、消息存儲(chǔ)模塊
負(fù)責(zé)存儲(chǔ)消息內(nèi)容、用戶(hù)ID和sessionID的關(guān)系，防止數(shù)據(jù)丟失或者服務(wù)器重啟等。
2.3、消息發(fā)送模塊
功能開(kāi)發(fā)完畢，一般部署到分布式集群環(huán)境，所以通訊時(shí)session會(huì)分布在多臺(tái)服務(wù)器。比如用戶(hù)A的session在機(jī)器1，用戶(hù)B的session在機(jī)器2，此時(shí)A發(fā)送給B，就無(wú)法單獨(dú)通過(guò)機(jī)器1完成。
因?yàn)闄C(jī)器1拿不到機(jī)器2里的session，所以消息發(fā)不過(guò)去。此時(shí)只能借助別的中間件來(lái)實(shí)現(xiàn)，比如借助消息中間件kafka實(shí)現(xiàn)。
機(jī)器1將消息發(fā)送給kafka，然后機(jī)器1和機(jī)器2都監(jiān)聽(tīng)kafka，然后查看用戶(hù)對(duì)應(yīng)的session是否在本機(jī)，如果在本機(jī)則發(fā)送出去。
2.4、消息推送模塊
模塊3提到的消息發(fā)送流程中，消息發(fā)送給 消息中間件，然后服務(wù)器消費(fèi)到消費(fèi)，在通過(guò)本機(jī)的session推送給客戶(hù)端。
2.5、架構(gòu)圖
3、遇到的幾個(gè)問(wèn)題
3.1、連接自動(dòng)斷開(kāi)
webSocket連接之后，發(fā)現(xiàn)一個(gè)問(wèn)題：就是每隔一段時(shí)間如果不傳送數(shù)據(jù)的話(huà)，與前端的連接就會(huì)自動(dòng)斷開(kāi)。采用心跳消息的方式，就可以解決這個(gè)問(wèn)題。比如客服端每隔30秒自動(dòng)發(fā)送ping消息給服務(wù)端，服務(wù)端返回pong。
3.2、Session無(wú)法被序列化
分布式場(chǎng)景會(huì)存在這樣一個(gè)問(wèn)題，當(dāng)一次請(qǐng)求負(fù)載到第一臺(tái)服務(wù)器時(shí)，session在第一臺(tái)服務(wù)器線(xiàn)程上，第二次請(qǐng)求，負(fù)載到第二臺(tái)服務(wù)器上，此時(shí)通過(guò)userId查找當(dāng)前用戶(hù)的session時(shí)，是查找不到的。
本來(lái)想著把session存入到redis中，就可以從redis獲取用戶(hù)的session，希望用這種方式來(lái)解決分布式場(chǎng)景下消息發(fā)送的問(wèn)題。結(jié)果出現(xiàn)如下錯(cuò)誤：
The remote endpoint was in state [STREAM_WRITING] which is an invalid state for called method
翻看了session源碼，發(fā)現(xiàn)session無(wú)法被序列化。所以這個(gè)方案只能放棄。解決方案請(qǐng)看下面的問(wèn)題5或者上面的架構(gòu)圖。
3.3、對(duì)象無(wú)法自動(dòng)注入
使用了@ServerEndpoint注解的類(lèi)中使用@Resource或@Autowired注入對(duì)象都會(huì)失敗，并且報(bào)空指針異常。
原因是WebSocket服務(wù)是線(xiàn)程安全的，那么當(dāng)我們?nèi)グl(fā)起一個(gè)ws連接時(shí)，就會(huì)創(chuàng)建一個(gè)端點(diǎn)對(duì)象。WebSocket服務(wù)是多對(duì)象的，不是單例的。而我們的Spring的Bean默認(rèn)就是單例的，在非單例類(lèi)中注入一個(gè)單例的Bean是沖突的。
或者說(shuō)：
Spring管理采用單例模式（singleton），而 WebSocket 是多對(duì)象的，即每個(gè)客戶(hù)端對(duì)應(yīng)后臺(tái)的一個(gè) WebSocket 對(duì)象，也可以理解成 new 了一個(gè) WebSocket，這樣當(dāng)然是不能獲得自動(dòng)注入的對(duì)象了，因?yàn)檫@兩者剛好沖突。
@Autowired 注解注入對(duì)象操作是在啟動(dòng)時(shí)執(zhí)行的，而不是在使用時(shí)，而 WebSocket 是只有連接使用時(shí)才實(shí)例化對(duì)象，且有多個(gè)連接就有多個(gè)對(duì)象。所以我們可以得出結(jié)論，這個(gè) Service 根本就沒(méi)有注入到 WebSocket 當(dāng)中。
如何解決呢？
使用靜態(tài)對(duì)象，并且對(duì)外暴露set方法，這樣在對(duì)象初始化的時(shí)候，將其注入到WebSocketServer中。比如說(shuō)這樣：
@ServerEndpoint("/trtc/websocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {
private static MessageStore messageStore;
private static MessageSender messageSender;
public static void setMessageStore(MessageStore messageStore) {
WebSocketServer.messageStore = messageStore;
}
public static void setMessageSender(MessageSender messageSender) {
WebSocketServer.messageSender = messageSender;
}
}
@Slf4j
@Service
public class MessageStore {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
@PostConstruct
public void init() {
WebSocketServer.setMessageStore(this);
}
}
3.4、分布式場(chǎng)景消息如何發(fā)給客戶(hù)端
問(wèn)題2中提到了分布式場(chǎng)景下存在的session不在本機(jī)的問(wèn)題，這種場(chǎng)景可以通過(guò)發(fā)送消息中間件的方式解決。具體這樣解決：
每次連接時(shí)，都將userId和對(duì)應(yīng)的session存入到本機(jī)，發(fā)送消息時(shí)，直接發(fā)送給MQ-Broker，然后每臺(tái)應(yīng)用負(fù)載都去消費(fèi)這個(gè)消息，拿到消息之后，判斷在本機(jī)能根據(jù)userId是否能找到session，找到則推送到客戶(hù)端。
3.5、部署時(shí)Nginx配置問(wèn)題
代碼開(kāi)發(fā)完畢之后，本機(jī)跑通后，然后部署到服務(wù)器之后，還差很重要的一步：配置nginx代理。
3.5.1、給后端應(yīng)用部署獨(dú)立域名
要給后端應(yīng)用部署獨(dú)立域名，nginx代理直接轉(zhuǎn)發(fā)到應(yīng)用的獨(dú)立域名，不要走微服務(wù)的gateway網(wǎng)關(guān)轉(zhuǎn)發(fā)過(guò)去。
3.5.2、多層nginx轉(zhuǎn)發(fā)問(wèn)題
當(dāng)只有一層nginx的時(shí)候，配置比較簡(jiǎn)單，如下：
location ~* ^/api/websocket/* {
proxy_pass http://mangodwsstest.mangod.top;
proxy_read_timeout 300s;
proxy_send_timeout 300s;
proxy_set_header Host mangodwsstest.mangod.top;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "Upgrade";
proxy_set_header X-Real-IP $remote_addr;
}
但是，當(dāng)有兩層nginx轉(zhuǎn)發(fā)的時(shí)候，問(wèn)題就出現(xiàn)了。
在最外層的nginx需要使用如下配置，不能照抄后面一層的配置。proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for和proxy_set_header X-Forwarded-Proto $scheme這兩個(gè)配置不能少，用來(lái)將協(xié)議和真實(shí)IP傳遞給后面一層的nginx。
location ~* ^/api/websocket/* {
proxy_pass http://mangodwsstest.mangod.top;
proxy_read_timeout 300s;
proxy_send_timeout 300s;
proxy_set_header Host $http_host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
}
4、完整代碼
頁(yè)面效果如下
開(kāi)啟兩個(gè)web頁(yè)面，用戶(hù)1輸入用戶(hù)2的用戶(hù)ID，輸入發(fā)送消息內(nèi)容，點(diǎn)擊發(fā)送。在用戶(hù)2的頁(yè)面的接受消息內(nèi)容可以看到發(fā)送的消息。
代碼結(jié)構(gòu)
代碼地址
https://github.com/yclxiao/spring-websocket.git
總結(jié)：本文介紹springboot如何集成websocket、IM及時(shí)通訊需要哪些模塊、開(kāi)發(fā)和部署過(guò)程中遇到的問(wèn)題、以及實(shí)現(xiàn)小型IM及時(shí)通訊的代碼。
以上就是SpringBoot+WebSocket實(shí)現(xiàn)IM及時(shí)通訊的代碼示例的詳細(xì)內(nèi)容，更多關(guān)于SpringBoot+WebSocket IM及時(shí)通訊的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
- Springboot實(shí)現(xiàn)人臉識(shí)別與WebSocket長(zhǎng)連接的實(shí)現(xiàn)代碼
- SpringBoot+websocket實(shí)現(xiàn)消息對(duì)話(huà)功能
- SpringBoot 整合WebSocket 前端 uniapp 訪(fǎng)問(wèn)的詳細(xì)方法
- Springboot+WebSocket+Netty實(shí)現(xiàn)在線(xiàn)聊天/群聊系統(tǒng)
- SpringBoot集成WebSocket的兩種方式(JDK內(nèi)置版和Spring封裝版)
- SpringBoot中使用WebSocket的教程分享
- SpringBoot實(shí)現(xiàn)WebSocket全雙工通信的項(xiàng)目實(shí)踐
- 使用WebSocket+SpringBoot+Vue搭建簡(jiǎn)易網(wǎng)頁(yè)聊天室的實(shí)現(xiàn)代碼
- SpringBoot整合WebSocket實(shí)現(xiàn)后端向前端發(fā)送消息的實(shí)例代碼
- Springboot+WebSocket實(shí)現(xiàn)在線(xiàn)聊天功能
- springboot中websocket簡(jiǎn)單實(shí)現(xiàn)
- Spring Boot中的WebSocketMessageBrokerConfigurer接口使用
相關(guān)文章
這篇文章主要介紹的是Java抽象類(lèi)和接口的區(qū)別詳情，2021-11-11
Java的Struts框架簡(jiǎn)介與環(huán)境配置教程
這篇文章主要介紹了Java的Struts框架與其環(huán)境配置教程,Struts框架是Java注明的SSH三大web開(kāi)發(fā)框架之一,需要的朋友可以參考下2015-11-11
一個(gè)合格JAVA軟件工程師應(yīng)該具備哪些專(zhuān)業(yè)技能，面試技巧是什么？本文為大家分享了2016版最新Java軟件工程師就業(yè)思維圖，感興趣的小伙伴們可以參考一下2016-11-11
基于springboot和redis實(shí)現(xiàn)單點(diǎn)登錄
這篇文章主要為大家詳細(xì)介紹了基于springboot和redis實(shí)現(xiàn)單點(diǎn)登錄，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-06-06
Spring使用AspectJ注解和XML配置實(shí)現(xiàn)AOP
這篇文章主要介紹了Spring使用AspectJ注解和XML配置實(shí)現(xiàn)AOP的相關(guān)資料,具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2016-10-10
spring和quartz整合,并簡(jiǎn)單調(diào)用(實(shí)例講解)
下面小編就為大家?guī)?lái)一篇spring和quartz整合,并簡(jiǎn)單調(diào)用(實(shí)例講解)。小編覺(jué)得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-07-07
基于序列化存取實(shí)現(xiàn)java對(duì)象深度克隆的方法詳解
本篇文章是對(duì)序列化存取實(shí)現(xiàn)java對(duì)象深度克隆的方法進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-05-05
最近稍微有點(diǎn)空閑，回頭再來(lái)優(yōu)化下基礎(chǔ)框架，也是一種重新學(xué)習(xí)。今天主要寫(xiě)寫(xiě)跟大家分享下springboot使用undertow，廢話(huà)不多說(shuō)2022-07-07
Spring?Boot?內(nèi)置工具類(lèi)ReflectionUtils的實(shí)現(xiàn)
ReflectionUtils是一個(gè)反射工具類(lèi),它封裝了Java反射的操作,使得我們能夠更輕松地操作和訪(fǎng)問(wèn)類(lèi)的方法、字段,本文主要介紹了Spring?Boot?內(nèi)置工具類(lèi)ReflectionUtils的實(shí)現(xiàn),感興趣的可以了解一下2023-11-11
spring對(duì)JDBC和orm的支持實(shí)例詳解
這篇文章主要介紹了spring對(duì)JDBC和orm的支持實(shí)例詳解,需要的朋友可以參考下2017-09-09