這篇文章主要介紹了springboot websocket集群連接時候傳遞參數(shù)的示例分析,具有一定借鑒價值,感興趣的朋友可以參考下,希望大家閱讀完這篇文章之后大有收獲,下面讓小編帶著大家一起了解一下。
創(chuàng)新互聯(lián)專注于開封網(wǎng)站建設服務及定制,我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗。 熱誠為您提供開封營銷型網(wǎng)站建設,開封網(wǎng)站制作、開封網(wǎng)頁設計、開封網(wǎng)站官網(wǎng)定制、微信小程序開發(fā)服務,打造開封網(wǎng)絡公司原創(chuàng)品牌,更為您提供開封網(wǎng)站排名全網(wǎng)營銷落地服務。
最近在公司項目中接到個需求。就是后臺跟前端瀏覽器要保持長連接,后臺主動往前臺推數(shù)據(jù)。
網(wǎng)上查了下,websocket stomp協(xié)議處理這個很簡單。尤其是跟springboot 集成。
但是由于開始是單機玩的,很順利。
但是后面部署到生產(chǎn)搞集群的話,就會出問題了。
假如集群兩個節(jié)點,瀏覽器A與節(jié)點A建立連接,A節(jié)點發(fā)的消息瀏覽器A節(jié)點肯定能收到。但是B節(jié)點由于沒有跟瀏覽器A建立連接。B節(jié)點發(fā)的消息瀏覽器就收不到了。
網(wǎng)上也查了好多,但是沒有一個說的很清楚的,也很多都是理論層面的。
還有很多思路都是通過session獲取信息的。但是這都不是我需要的。我需要的是從前臺傳遞參數(shù),連接的時候每個節(jié)點保存下。然后通過SimpleUserRegistry.getUser獲取。
代碼:
<script type="text/javascript" src="${request.contextPath}/scripts/sockjs.min.js"></script> <script type="text/javascript" src="${request.contextPath}/scripts/stomp.min.js"></script> var WEB_SOCKET = { topic : "", url : "", stompClient : null, connect : function(url, topic, callback,userid) { this.url = url; this.topic = topic; var socket = new SockJS(url); //連接SockJS的endpoint名稱為"endpointOyzc" WEB_SOCKET.stompClient = Stomp.over(socket);//使用STMOP子協(xié)議的WebSocket客戶端 WEB_SOCKET.stompClient.connect({userid:userid},function(frame){//連接WebSocket服務端 // console.log('Connected:' + frame); //通過stompClient.subscribe訂閱/topic/getResponse 目標(destination)發(fā)送的消息 WEB_SOCKET.stompClient.subscribe(topic, callback); }); } };
這是響應的前端代碼。只需要引入兩個js。調用new SockJS(url) 就代表跟服務器建立連接了。
@Configuration //注解開啟使用STOMP協(xié)議來傳輸基于代理(message broker)的消息,這時控制器支持使用@MessageMapping,就像使用@RequestMapping一樣 @EnableWebSocketMessageBroker public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer { @Autowired private GetHeaderParamInterceptor getHeaderParamInterceptor; @Override //注冊STOMP協(xié)議的節(jié)點(endpoint),并映射指定的url public void registerStompEndpoints(StompEndpointRegistry registry) { //注冊一個STOMP的endpoint,并指定使用SockJS協(xié)議 registry.addEndpoint("/endpointOyzc") .setAllowedOrigins("*") .withSockJS(); /* registry.addEndpoint("/endpointOyzc") .setAllowedOrigins("*") .setHandshakeHandler(xlHandshakeHandler) .withSockJS();*/ } @Override //配置消息代理(Message Broker) public void configureMessageBroker(MessageBrokerRegistry registry) { //點對點應配置一個/user消息代理,廣播式應配置一個/topic消息代理 registry.enableSimpleBroker("/topic", "/user"); // 全局使用的消息前綴(客戶端訂閱路徑上會體現(xiàn)出來) //registry.setApplicationDestinationPrefixes("/app"); //點對點使用的訂閱前綴(客戶端訂閱路徑上會體現(xiàn)出來),不設置的話,默認也是/user/ registry.setUserDestinationPrefix("/user"); } /** * 采用自定義攔截器,獲取connect時候傳遞的參數(shù) * * @param registration */ @Override public void configureClientInboundChannel(ChannelRegistration registration) { registration.interceptors(getHeaderParamInterceptor); } }
注:上面的endpointOyzc就是前端的url。后面注冊端點,前臺鏈接。
然后注意下configureClientInboundChannel這個方法,這個方法里面注入攔截器就是為了鏈接時候接收參數(shù)的。
/** * @author : hao * @description : websocket建立鏈接的時候獲取headeri里認證的參數(shù)攔截器。 * @time : 2019/7/3 20:42 */ @Component public class GetHeaderParamInterceptor extends ChannelInterceptorAdapter { @Override public Message<?> preSend(Message<?> message, MessageChannel channel) { StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); if (StompCommand.CONNECT.equals(accessor.getCommand())) { Object raw = message.getHeaders().get(SimpMessageHeaderAccessor.NATIVE_HEADERS); if (raw instanceof Map) { Object name = ((Map) raw).get("userid"); if (name instanceof LinkedList) { // 設置當前訪問的認證用戶 accessor.setUser(new JqxxPrincipal(((LinkedList) name).get(0).toString())); } } } return message; } } /** * @author : hao * @description : 自定義的java.security.Principal * @time : 2019/7/3 20:42 */ public class JqxxPrincipal implements Principal { private String loginName; public JqxxPrincipal(String loginName) { this.loginName = loginName; } @Override public String getName() { return loginName; } }
這樣就存入的前臺傳的參數(shù)。
后臺發(fā)消息的時候怎么發(fā)呢?
/** * @author : hao * @description : websocket發(fā)送代理,負責發(fā)送消息 * @time : 2019/7/4 11:01 */ @Component @Slf4j public class WebsocketSendProxy<T> { @Autowired private SimpMessagingTemplate template; @Autowired private SimpUserRegistry userRegistry; @Resource(name = "redisServiceImpl") private RedisService redisService; @Value("spring.redis.message.topic-name") private String topicName; public void sendMsg(RedisWebsocketMsg<T> redisWebsocketMsg) { SimpUser simpUser = userRegistry.getUser(redisWebsocketMsg.getReceiver()); log.info("發(fā)送消息前獲取接收方為{},根據(jù)Registry獲取本節(jié)點上這個用戶{}", redisWebsocketMsg.getReceiver(), simpUser); if (simpUser != null && StringUtils.isNotBlank(simpUser.getName())) { //2. 獲取WebSocket客戶端的訂閱地址 WebSocketChannelEnum channelEnum = WebSocketChannelEnum.fromCode(redisWebsocketMsg.getChannelCode()); if (channelEnum != null) { //3. 給WebSocket客戶端發(fā)送消息 template.convertAndSendToUser(redisWebsocketMsg.getReceiver(), channelEnum.getSubscribeUrl(), redisWebsocketMsg.getContent()); } } else { //給其他訂閱了主題的節(jié)點發(fā)消息,因為本節(jié)點沒有 redisService.convertAndSend(topicName, redisWebsocketMsg); } } }
可以發(fā)現(xiàn)上面代碼利用了redis監(jiān)聽模型,也就是redis模型的消息隊列
/** * @author : hao * @description : redis消息監(jiān)聽實現(xiàn)類,接收處理類 * @time : 2019/7/3 14:00 */ @Component @Slf4j public class MessageReceiver { @Autowired private SimpMessagingTemplate messagingTemplate; @Autowired private SimpUserRegistry userRegistry; /** * 處理WebSocket消息 */ public void receiveMessage(RedisWebsocketMsg redisWebsocketMsg) { log.info(MessageFormat.format("Received Message: {0}", redisWebsocketMsg)); //1. 取出用戶名并判斷是否連接到當前應用節(jié)點的WebSocket SimpUser simpUser = userRegistry.getUser(redisWebsocketMsg.getReceiver()); if (simpUser != null && StringUtils.isNotBlank(simpUser.getName())) { //2. 獲取WebSocket客戶端的訂閱地址 WebSocketChannelEnum channelEnum = WebSocketChannelEnum.fromCode(redisWebsocketMsg.getChannelCode()); if (channelEnum != null) { //3. 給WebSocket客戶端發(fā)送消息 messagingTemplate.convertAndSendToUser(redisWebsocketMsg.getReceiver(), channelEnum.getSubscribeUrl(), redisWebsocketMsg.getContent()); } } } }
redis消息模型只貼部分代碼就好了
/** * 消息監(jiān)聽器 */ @Bean MessageListenerAdapter messageListenerAdapter(MessageReceiver messageReceiver, Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer){ //消息接收者以及對應的默認處理方法 MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(messageReceiver, "receiveMessage"); //消息的反序列化方式 messageListenerAdapter.setSerializer(jackson2JsonRedisSerializer); return messageListenerAdapter; } /** * message listener container */ @Bean RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory , MessageListenerAdapter messageListenerAdapter){ RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); //添加消息監(jiān)聽器 container.addMessageListener(messageListenerAdapter, new PatternTopic(topicName)); return container; }
上面的思路大體如下:客戶端簡歷鏈接時候,傳過來userid保存起來。發(fā)消息的時候 通過userRegistry獲取,能獲取到就證明是跟本節(jié)點建立的鏈接,直接用本節(jié)點發(fā)消息就好了。
如果不是就利用redis消息隊列,把消息推出去。每個節(jié)點去判斷獲取看下是不是本節(jié)點的userid。這樣就實現(xiàn)了集群的部署。
感謝你能夠認真閱讀完這篇文章,希望小編分享的“springboot websocket集群連接時候傳遞參數(shù)的示例分析”這篇文章對大家有幫助,同時也希望大家多多支持創(chuàng)新互聯(lián),關注創(chuàng)新互聯(lián)行業(yè)資訊頻道,更多相關知識等著你來學習!
標題名稱:springbootwebsocket集群連接時候傳遞參數(shù)的示例分析
當前網(wǎng)址:http://muchs.cn/article40/jcpeeo.html
成都網(wǎng)站建設公司_創(chuàng)新互聯(lián),為您提供用戶體驗、搜索引擎優(yōu)化、App開發(fā)、手機網(wǎng)站建設、移動網(wǎng)站建設、定制網(wǎng)站
聲明:本網(wǎng)站發(fā)布的內容(圖片、視頻和文字)以用戶投稿、用戶轉載內容為主,如果涉及侵權請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內容未經(jīng)允許不得轉載,或轉載時需注明來源: 創(chuàng)新互聯(lián)