springboot基于Redis發(fā)布訂閱集群下WebSocket的解決方案
更新時間：2021年01月28日 11:15:20 作者：毅大師
這篇文章主要介紹了springboot基于Redis發(fā)布訂閱集群下WebSocket的解決方案，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
一、背景
單機(jī)節(jié)點下，WebSocket連接成功后，可以直接發(fā)送消息。而多節(jié)點下，連接時通過nginx會代理到不同節(jié)點。
假設(shè)一開始用戶連接了node1的socket服務(wù)。觸發(fā)消息發(fā)送的條件的時候也通過nginx進(jìn)行代理，假如代理轉(zhuǎn)到了node2節(jié)點上，那么node2節(jié)點的socket服務(wù)就發(fā)送不了消息，因為一開始用戶注冊的是node1節(jié)點。這就導(dǎo)致了消息發(fā)送失敗。
為了解決這一方案，消息發(fā)送時，就需要一個中間件來記錄，這樣，三個節(jié)點都可以獲取消息，然后在根據(jù)條件進(jìn)行消息推送。
二、解決方案(springboot 基于 Redis發(fā)布訂閱)
1、依賴
<!-- redis --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <!-- websocket --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency>
2、創(chuàng)建業(yè)務(wù)處理類 Demo.class，該類可以實現(xiàn)MessageListener接口后重寫onMessage方法，也可以不實現(xiàn)，自己寫方法。
import com.alibaba.fastjson.JSON;
import com.dy.service.impl.OrdersServiceImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
import java.util.HashMap;
/**
* @program:
* @description: redis消息訂閱-業(yè)務(wù)處理
* @author: zhang yi
* @create: 2021-01-25 16:46
*/
@Component
public class Demo implements MessageListener {
Logger logger = LoggerFactory.getLogger(this.getClass());
@Override
public void onMessage(Message message, byte[] pattern) {
logger.info("消息訂閱成功---------");
logger.info("內(nèi)容："+message.getBody());
logger.info("交換機(jī)："+message.getChannel());
}
}
3、創(chuàng)建PubSubConfig配置類
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
/**
* @program:
* @description: redis發(fā)布訂閱配置
* @author: zhang yi
* @create: 2021-01-25 16:49
*/
@Configuration
@EnableCaching
public class PubSubConfig {
Logger logger = LoggerFactory.getLogger(this.getClass());
//如果是多個交換機(jī)，則參數(shù)為(RedisConnectionFactory connectionFactory,
// MessageListenerAdapter listenerAdapter,
// MessageListenerAdapter listenerAdapter2)
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
// 可以添加多個 messageListener，配置不同的交換機(jī)
container.addMessageListener(listenerAdapter, new PatternTopic("channel:demo"));
//container.addMessageListener(listenerAdapter2, new PatternTopic("channel:demo2"));
return container;
}
/**
* 消息監(jiān)聽器適配器，綁定消息處理器，利用反射技術(shù)調(diào)用消息處理器的業(yè)務(wù)方法
* @param demo 第一步的業(yè)務(wù)處理類
* @return
*/
@Bean
MessageListenerAdapter listenerAdapter(Demo demo) {
logger.info("----------------消息監(jiān)聽器加載成功----------------");
// onMessage 就是方法名，基于反射調(diào)用
return new MessageListenerAdapter(demo, "onMessage");
}
/**
* 多個交換機(jī)就多寫一個
* @param subCheckOrder
* @return
*/
//@Bean
//MessageListenerAdapter listenerAdapter2(SubCheckOrder subCheckOrder) {
// logger.info("----------------消息監(jiān)聽器加載成功----------------");
// return new MessageListenerAdapter(subCheckOrder, "onMessage");
//}
@Bean
StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
return new StringRedisTemplate(connectionFactory);
}
}
4、消息發(fā)布
@Autowired
private RedisTemplate<String, Object> redisTemplate;
redisTemplate.convertAndSend("channel:demo", "我是內(nèi)容");
三、具體用法
- socket連接成功。
- socket消息推送時，把信息發(fā)布到redis中。socket服務(wù)訂閱redis的消息，訂閱成功后進(jìn)行推送。集群下的socket都能訂閱到消息，但是只有之前連接成功的節(jié)點能推送成功，其余的無法推送。
相關(guān)文章
簡單地說，在keyset方法返回的set上做修改會改變原來hashmap，這也許不是你想要的，于是形成一個隱藏的bug2013-04-04
關(guān)于?Safe?Point?是?JVM?中很關(guān)鍵的一個概念，但我估計有不少同學(xué)不是很懂，于是今天跟大家來深入聊聊?Safe?Point，通過本文學(xué)習(xí)你會了解什么是?Safe?Point？為啥需要?Safe?Point？Safe?Point?與?Stop?the?World?的關(guān)系？感興趣的朋友一起看看吧2022-10-10
這篇文章給大家介紹了代碼縮進(jìn)用空格還是Tab？（IDEA中設(shè)置Tab健為4個空格）的相關(guān)知識，本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友參考下吧2021-03-03
解決跨域請求,主要有JSONP,iframe,window.name,CORS等方式,其中CORS方式是最常用的跨域?qū)崿F(xiàn)方式,而且是對各種請求方法、各種數(shù)據(jù)請求類型都是完美支持的,本文介紹了SpringBoot3解決跨域請求的方案小結(jié),需要的朋友可以參考下2024-07-07
這篇文章主要介紹了前端向后端傳遞參數(shù)的多種方式,包括URL參數(shù)（查詢參數(shù)、路徑參數(shù)）、請求體（JSON數(shù)據(jù)、表單數(shù)據(jù)、文件上傳）、請求頭和Cookie,并總結(jié)了每種方式的適用場景,需要的朋友可以參考下2025-03-03
關(guān)于break和continue以及l(fā)abel的區(qū)別和作用(詳解)
下面小編就為大家?guī)硪黄P(guān)于break和continue以及l(fā)abel的區(qū)別和作用(詳解)。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-05-05
IDEA “Cannot resolve symbol”爆紅問題解決
最近發(fā)現(xiàn)個問題,IDEA 無法識別同一個 package 里的其他類,將其顯示為紅色,本文就來介紹一下IDEA “Cannot resolve symbol”爆紅問題解決,感興趣的可以了解一下2023-10-10
這篇文章主要介紹了Java CompletableFuture的使用詳解，幫助大家更好的理解和學(xué)習(xí)使用Java，感興趣的朋友可以了解下2021-03-03
這篇文章主要介紹了Java線程中的線程本地變量ThreadLocal詳解,ThreadLocal存放的值是線程內(nèi)共享的,線程間互斥的,主要用于線程內(nèi)共享一些數(shù)據(jù),避免通過參數(shù)來傳遞,這樣處理后,能夠優(yōu)雅的解決一些實際問題,需要的朋友可以參考下2023-11-11
工廠方法模式(FACTORY METHOD)是一種常用的類創(chuàng)建型設(shè)計模式,此模式的核心精神是封裝類中變化的部分，提取其中個性化善變的部分為獨立類，通過依賴注入以達(dá)到解耦、復(fù)用和方便后期維護(hù)拓展的目的。它的核心結(jié)構(gòu)有四個角色，分別是抽象工廠、具體工廠、抽象產(chǎn)品、具體產(chǎn)品2022-04-04