第SpringbootWebsocketStomp消息訂閱推送

pom文件

直接引入spring-boot-starter-websocket即可。

dependency

groupIdorg.springframework.boot/groupId

artifactIdspring-boot-starter-websocket/artifactId

/dependency

聲明websocketendpoint

importorg.springframework.context.annotation.Bean;

importorg.springframework.context.annotation.Configuration;

importorg.springframework.web.socket.server.standard.ServerEndpointExporter;

*@ClassNameWebSocketConfig

*@Authorscott

*@Date2025/6/16

*@VersionV1.0

@Configuration

publicclassWebSocketConfig{

*注入一個ServerEndpointExporter,該Bean會自動注冊使用@ServerEndpoint注解申明的websocketendpoint

@Bean

publicServerEndpointExporterserverEndpointExporter(){

returnnewServerEndpointExporter();

websocket實現(xiàn)類，其中通過注解監(jiān)聽了各種事件，實現(xiàn)了推送消息等相關邏輯

importmon.cache.Cache;

importmon.cache.CacheBuilder;

importmon.core.domain.AjaxResult;

importorg.slf4j.Logger;

importorg.slf4j.LoggerFactory;

importorg.springframework.stereotype.Component;

importjavax.websocket.*;

importjavax.websocket.server.PathParam;

importjavax.websocket.server.ServerEndpoint;

importjava.util.Objects;

importjava.util.Set;

importjava.util.concurrent.ConcurrentMap;

importjava.util.concurrent.TimeUnit;

importjava.util.concurrent.atomic.AtomicInteger;

*@ClassName:DataTypePushWebSocket

*@Author:scott

*@Date:2025/6/16

@ServerEndpoint(value="/ws/dataType/push/{token}")

@Component

publicclassDataTypePushWebSocket{

privatestaticfinalLoggerlog=LoggerFactory.getLogger(DataTypePushWebSocket.class);

*記錄當前在線連接數(shù)

privatestaticAtomicIntegeronlineCount=newAtomicInteger(0);

privatestaticCacheString,SessionSESSION_CACHE=CacheBuilder.newBuilder()

.initialCapacity(10)

.maximumSize(300)

.expireAfterWrite(10,TimeUnit.MINUTES)

.build();

*連接建立成功調(diào)用的方法

@OnOpen

publicvoidonOpen(Sessionsession,@PathParam("token")Stringtoken){

StringsessionId=session.getId();

onlineCount.incrementAndGet();//在線數(shù)加1

this.sendMessage("sessionId:"+sessionId+",已經(jīng)和server建立連接",session);

SESSION_CACHE.put(sessionId,session);

("有新連接加入：{}，當前在線連接數(shù)為：{}",session.getId(),onlineCount.get());

*連接關閉調(diào)用的方法

@OnClose

publicvoidonClose(Sessionsession,@PathParam("token")Stringtoken){

onlineCount.decrementAndGet();//在線數(shù)減1

SESSION_CACHE.invalidate(session.getId());

("有一連接關閉：{}，當前在線連接數(shù)為：{}",session.getId(),onlineCount.get());

*收到客戶端消息后調(diào)用的方法

*@parammessage客戶端發(fā)送過來的消息

@OnMessage

publicvoidonMessage(Stringmessage,Sessionsession,@PathParam("token")Stringtoken){

("服務端收到客戶端[{}]的消息:{}",session.getId(),message);

this.sendMessage("服務端已收到推送消息:"+message,session);

@OnError

publicvoidonError(Sessionsession,Throwableerror){

log.error("發(fā)生錯誤");

error.printStackTrace();

*服務端發(fā)送消息給客戶端

privatestaticvoidsendMessage(Stringmessage,SessiontoSession){

try{

("服務端給客戶端[{}]發(fā)送消息{}",toSession.getId(),message);

toSession.getBasicRemote().sendText(message);

}catch(Exceptione){

log.error("服務端發(fā)送消息給客戶端失敗：{}",e);

publicstaticAjaxResultsendMessage(Stringmessage,StringsessionId){

Sessionsession=SESSION_CACHE.getIfPresent(sessionId);

if(Objects.isNull(session)){

returnAjaxResult.error("token已失效");

sendMessage(message,session);

returnAjaxResult.success();

publicstaticAjaxResultsendBroadcast(Stringmessage){

longsize=SESSION_CACHE.size();

if(size=0){

returnAjaxResult.error("當前沒有在線客戶端，無法推送消息");

ConcurrentMapString,SessionsessionConcurrentMap=SESSION_CACHE.asMap();

SetStringkeys=sessionConcurrentMap.keySet();

for(Stringkey:keys){

Sessionsession=SESSION_CACHE.getIfPresent(key);

DataTypePushWebSocket.sendMessage(message,session);

returnAjaxResult.success();

至此websocket服務端代碼已經(jīng)完成。

stomp協(xié)議

前端代碼.這個是在某個vue工程中寫的js，各位大佬自己動手改改即可。其中Settings.wsPath是后端定義的ws地址例如ws://localhost:9003/ws

importStompfrom'stompjs'

importSettingsfrom'@/settings.js'

exportdefault{

//是否啟用日志默認啟用

debug:true,

//客戶端連接信息

stompClient:{},

//初始化

init(callBack){

this.stompClient=Stomp.client(Settings.wsPath)

this.stompClient.hasDebug=this.debug

this.stompClient.connect({},suce={

this.console("連接成功,信息如下↓")

this.console(this.stompClient)

if(callBack){

callBack()

},err={

if(err){

this.console("連接失敗,信息如下↓")

this.console(err)

//訂閱

sub(address,callBack){

if(!this.stompClient.connected){

this.console("沒有連接,無法訂閱")

return

//生成id

lettimestamp=newDate().getTime()+address

this.console("訂閱成功-"+address)

this.stompClient.subscribe(address,message={

this.console(address+"訂閱消息通知,信息如下↓")

this.console(message)

letdata=message.body

callBack(data)

id:timestamp

unSub(address){

if(!this.stompClient.connected){

this.console("沒有連接,無法取消訂閱-"+address)

return

letid=""

for(letiteminthis.stompClient.subscriptions){

if(item.endsWith(address)){

id=item

break

this.stompClient.unsubscribe(id)

this.console("取消訂閱成功-id:"+id+"address:"+address)

//斷開連接

disconnect(callBack){

if(!this.stompClient.connected){

this.console("沒有連接,無法斷開連接")

return

this.stompClient.disconnect(()={

console.log("斷開成功")

if(callBack){

callBack()

//單位秒

reconnect(time){

setInterval(()={

if(!this.stompClient.connected){

this.console("重新連接中...")

this.init()

},time*1000)

console(msg){

if(this.debug){

console.log(msg)

//向訂閱發(fā)送消息

send(address,msg){

this.stompClient.send(address,{},msg)

后端stompconfig,里面都有注釋，寫的很詳細,并且我加入了和前端的心跳pingpong。

package.scott.config;

importorg.springframework.context.annotation.Configuration;

importorg.springframework.messaging.simp.config.MessageBrokerRegistry;

importorg.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

importorg.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;

importorg.springframework.web.socket.config.annotation.StompEndpointRegistry;

importorg.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

*@ClassName:WebSocketStompConfig

*@Author:scott

*@Date:2025/7/8

@Configuration

@EnableWebSocketMessageBroker

publicclassWebSocketStompConfigimplementsWebSocketMessageBrokerConfigurer{

privatestaticlongHEART_BEAT=10000;

@Override

publicvoidregisterStompEndpoints(StompEndpointRegistryregistry){

//允許使用socketJs方式訪問，訪問點為webSocket，允許跨域

//在網(wǎng)頁上我們就可以通過這個鏈接

//ws://:port/ws來和服務器的WebSocket連接

registry.addEndpoint("/ws").setAllowedOrigins("*");

@Override

publicvoidconfigureMessageBroker(MessageBrokerRegistryregistry){

ThreadPoolTaskSchedulerte=newThreadPoolTaskScheduler();

te.setPoolSize(1);

te.setThreadNamePrefix("wss-heartbeat-thread-");

te.initialize();

//基于內(nèi)存的STOMP消息代理來代替mq的消息代理

//訂閱Broker名稱,/user代表點對點即發(fā)指定用戶，/topic代表發(fā)布廣播即群發(fā)

//setHeartbeatValue設置心跳及心跳時間

registry.enableSimpleBroker("/user","/topic").setHeartbeatValue(newlong[]{HEART_BEAT,HEART_BEAT}).setTaskScheduler(te);

//點對點使用的訂閱前綴，不設置的話，默認也是/user/

registry.setUserDestinationPrefix("/user/");

后端stomp協(xié)議接受、訂閱等動作通知

package.scott.ws;

importcom.alibaba.fastjson.JSON;

importorg.springframework.beans.factory.annotation.Autowired;

importorg.springframework.messaging.handler.annotation.DestinationVariable;

importorg.springframework.messaging.handler.annot