文章詳情頁

Springboot Websocket Stomp 消息訂閱推送

瀏覽：2日期：2023-02-26 13:46:21

目錄需求背景websocket協議stomp協議需求背景

閑話不扯，直奔主題。需要和web前端建立長鏈接，互相實時通訊，因此想到了websocket，后面隨著需求的變更，需要用戶訂閱主題，實現消息的精準推送，發布訂閱等，則想到了STOMP(Simple Text-Orientated Messaging Protocol) 面向消息的簡單文本協議。

websocket協議

想到了之前寫的一個websocket長鏈接的demo，也貼上代碼供大家參考。

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

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId></dependency>

聲明websocket endpoint

import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.web.socket.server.standard.ServerEndpointExporter;/** * @ClassName WebSocketConfig * @Author scott * @Date 2021/6/16 * @Version V1.0 **/@Configurationpublic class WebSocketConfig { /** * 注入一個ServerEndpointExporter,該Bean會自動注冊使用@ServerEndpoint注解申明的websocket endpoint */ @Bean public ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter(); }}

websocket實現類，其中通過注解監聽了各種事件，實現了推送消息等相關邏輯

import com.google.common.cache.Cache;import com.google.common.cache.CacheBuilder;import com.ruoyi.common.core.domain.AjaxResult;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;import javax.websocket.*;import javax.websocket.server.PathParam;import javax.websocket.server.ServerEndpoint;import java.util.Objects;import java.util.Set;import java.util.concurrent.ConcurrentMap;import java.util.concurrent.TimeUnit;import java.util.concurrent.atomic.AtomicInteger;/** * @ClassName: DataTypePushWebSocket * @Author: scott * @Date: 2021/6/16**/@ServerEndpoint(value = '/ws/dataType/push/{token}')@Componentpublic class DataTypePushWebSocket { private static final Logger log = LoggerFactory.getLogger(DataTypePushWebSocket.class); /** * 記錄當前在線連接數 */ private static AtomicInteger onlineCount = new AtomicInteger(0); private static Cache<String, Session> SESSION_CACHE = CacheBuilder.newBuilder() .initialCapacity(10) .maximumSize(300) .expireAfterWrite(10, TimeUnit.MINUTES) .build(); /** * 連接建立成功調用的方法 */ @OnOpen public void onOpen(Session session, @PathParam('token')String token) {String sessionId = session.getId();onlineCount.incrementAndGet(); // 在線數加1this.sendMessage('sessionId:' + sessionId +',已經和server建立連接', session);SESSION_CACHE.put(sessionId,session);log.info('有新連接加入：{}，當前在線連接數為：{}', session.getId(), onlineCount.get()); } /** * 連接關閉調用的方法 */ @OnClose public void onClose(Session session,@PathParam('token')String token) {onlineCount.decrementAndGet(); // 在線數減1SESSION_CACHE.invalidate(session.getId());log.info('有一連接關閉：{}，當前在線連接數為：{}', session.getId(), onlineCount.get()); } /** * 收到客戶端消息后調用的方法 * * @param message 客戶端發送過來的消息 */ @OnMessage public void onMessage(String message, Session session,@PathParam('token')String token) {log.info('服務端收到客戶端[{}]的消息:{}', session.getId(), message);this.sendMessage('服務端已收到推送消息:' + message, session); } @OnError public void onError(Session session, Throwable error) {log.error('發生錯誤');error.printStackTrace(); } /** * 服務端發送消息給客戶端 */ private static void sendMessage(String message, Session toSession) {try { log.info('服務端給客戶端[{}]發送消息{}', toSession.getId(), message); toSession.getBasicRemote().sendText(message);} catch (Exception e) { log.error('服務端發送消息給客戶端失敗：{}', e);} } public static AjaxResult sendMessage(String message, String sessionId){Session session = SESSION_CACHE.getIfPresent(sessionId);if(Objects.isNull(session)){ return AjaxResult.error('token已失效');}sendMessage(message,session);return AjaxResult.success(); } public static AjaxResult sendBroadcast(String message){long size = SESSION_CACHE.size();if(size <=0){ return AjaxResult.error('當前沒有在線客戶端，無法推送消息');}ConcurrentMap<String, Session> sessionConcurrentMap = SESSION_CACHE.asMap();Set<String> keys = sessionConcurrentMap.keySet();for (String key : keys) { Session session = SESSION_CACHE.getIfPresent(key); DataTypePushWebSocket.sendMessage(message,session);}return AjaxResult.success(); }}

至此websocket服務端代碼已經完成。

stomp協議

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

import Stomp from ’stompjs’import Settings from ’@/settings.js’export default { // 是否啟用日志默認啟用 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 let timestamp= new Date().getTime() + address this.console('訂閱成功 -> '+address) this.stompClient.subscribe(address,message => { this.console(address+' 訂閱消息通知,信息如下 ↓') this.console(message) let data = message.body callBack(data) },{ id: timestamp }) }, unSub(address){ if(!this.stompClient.connected){ this.console('沒有連接,無法取消訂閱 -> '+address) return } let id = '' for(let item in this.stompClient.subscriptions){ if(item.endsWith(address)){id = itembreak } } 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) } }, // 向訂閱發送消息 send(address,msg) { this.stompClient.send(address,{},msg) }}

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

package com.cn.scott.config;import org.springframework.context.annotation.Configuration;import org.springframework.messaging.simp.config.MessageBrokerRegistry;import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;import org.springframework.web.socket.config.annotation.StompEndpointRegistry;import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;/** * @ClassName: WebSocketStompConfig * @Author: scott * @Date: 2021/7/8**/@Configuration@EnableWebSocketMessageBrokerpublic class WebSocketStompConfig implements WebSocketMessageBrokerConfigurer { private static long HEART_BEAT=10000; @Override public void registerStompEndpoints(StompEndpointRegistry registry) {//允許使用socketJs方式訪問，訪問點為webSocket，允許跨域//在網頁上我們就可以通過這個鏈接//ws://127.0.0.1:port/ws來和服務器的WebSocket連接registry.addEndpoint('/ws').setAllowedOrigins('*'); } @Override public void configureMessageBroker(MessageBrokerRegistry registry) {ThreadPoolTaskScheduler te = new ThreadPoolTaskScheduler();te.setPoolSize(1);te.setThreadNamePrefix('wss-heartbeat-thread-');te.initialize();//基于內存的STOMP消息代理來代替mq的消息代理//訂閱Broker名稱,/user代表點對點即發指定用戶，/topic代表發布廣播即群發//setHeartbeatValue 設置心跳及心跳時間registry.enableSimpleBroker('/user', '/topic').setHeartbeatValue(new long[]{HEART_BEAT,HEART_BEAT}).setTaskScheduler(te);//點對點使用的訂閱前綴，不設置的話，默認也是/user/registry.setUserDestinationPrefix('/user/'); }}

后端stomp協議接受、訂閱等動作通知

package com.cn.scott.ws;import com.alibaba.fastjson.JSON;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.messaging.handler.annotation.DestinationVariable;import org.springframework.messaging.handler.annotation.MessageMapping;import org.springframework.messaging.simp.SimpMessagingTemplate;import org.springframework.messaging.simp.annotation.SubscribeMapping;import org.springframework.web.bind.annotation.RestController;/** * @ClassName StompSocketHandler * @Author scott * @Date 2021/6/30 * @Version V1.0 **/@RestControllerpublic class StompSocketHandler { @Autowired private SimpMessagingTemplate simpMessagingTemplate; /** * @MethodName: subscribeMapping * @Description: 訂閱成功通知 * @Param: [id] * @Return: void * @Author: scott * @Date: 2021/6/30 **/ @SubscribeMapping('/user/{id}/listener') public void subscribeMapping(@DestinationVariable('id') final long id) {System.out.println('>>>>>>用戶:'+id +',已訂閱');SubscribeMsg param = new SubscribeMsg(id,String.format('用戶【%s】已訂閱成功', id));sendToUser(param); } /** * @MethodName: test * @Description: 接收訂閱topic消息 * @Param: [id, msg] * @Return: void * @Author: scott * @Date: 2021/6/30 **/ @MessageMapping(value = '/user/{id}/listener') public void UserSubListener(@DestinationVariable long id, String msg) {System.out.println('收到客戶端:' +id+',的消息');SubscribeMsg param = new SubscribeMsg(id,String.format('已收到用戶【%s】發送消息【%s】', id,msg));sendToUser(param); } @GetMapping('/refresh/{userId}') public void refresh(@PathVariable Long userId, String msg) {StompSocketHandler.SubscribeMsg param = new StompSocketHandler.SubscribeMsg(userId,String.format('服務端向用戶【%s】發送消息【%s】', userId,msg));sendToUser(param); } /** * @MethodName: sendToUser * @Description: 推送消息給訂閱用戶 * @Param: [userId] * @Return: void * @Author: scott * @Date: 2021/6/30 **/ public void sendToUser(SubscribeMsg screenChangeMsg){//這里可以控制權限等。。。simpMessagingTemplate.convertAndSendToUser(screenChangeMsg.getUserId().toString(),'/listener', JSON.toJSONString(screenChangeMsg)); } /** * @MethodName: sendBroadCast * @Description: 發送廣播，需要用戶事先訂閱廣播 * @Param: [topic, msg] * @Return: void * @Author: scott * @Date: 2021/6/30 **/ public void sendBroadCast(String topic,String msg){simpMessagingTemplate.convertAndSend(topic,msg); } /** * @ClassName: SubMsg * @Author: scott * @Date: 2021/6/30 **/ public static class SubscribeMsg {private Long userId;private String msg;public SubscribeMsg(Long UserId, String msg){ this.userId = UserId; this.msg = msg;}public Long getUserId() { return userId;}public String getMsg() { return msg;} }}

連接展示

建立連接成功，這里可以看出是基于websocket協議