websocket+redis動(dòng)態(tài)訂閱和動(dòng)態(tài)取消訂閱的實(shí)現(xiàn)示例
更新時(shí)間：2022年05月17日 14:56:02 作者：柯騰_
本文主要介紹了websocket+redis動(dòng)態(tài)訂閱和動(dòng)態(tài)取消訂閱，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
原理
websocket的訂閱就是在前后端建立ws連接之后，前端通過(guò)發(fā)送一定格式的消息，后端解析出來(lái)去訂閱或者取消訂閱redis頻道。
訂閱頻道消息格式:
{
"cmd":"subscribe",
"topic":[
"topic_name"
]
}
模糊訂閱格式
{
"cmd":"psubscribe",
"topic":[
"topic_name"
]
}
取消訂閱格式
{
"cmd":"unsubscribe",
"topic":[
"topic_name"
]
}
兩個(gè)核心類(lèi)，一個(gè)是redis的訂閱監(jiān)聽(tīng)類(lèi)，一個(gè)是websocket的發(fā)布訂閱類(lèi)。
redis訂閱監(jiān)聽(tīng)類(lèi)
package com.curtain.core;
import com.curtain.config.GetBeanUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;
import java.util.Arrays;
/**
?* @Author Curtain
?* @Date 2021/6/7 14:27
?* @Description
?*/
@Component
@Slf4j
public class RedisPubSub extends JedisPubSub {
? ? private JedisPool jedisPool = GetBeanUtil.getBean(JedisPool.class);
? ? private Jedis jedis;
? ? //訂閱
? ? public void subscribe(String... channels) {
? ? ? ? jedis = jedisPool.getResource();
? ? ? ? try {
? ? ? ? ? ? jedis.subscribe(this, channels);
? ? ? ? } catch (Exception e) {
? ? ? ? ? ? log.error(e.getMessage());
? ? ? ? ? ? if (jedis != null)
? ? ? ? ? ? ? ? jedis.close();
? ? ? ? ? ? //遇到異常后關(guān)閉連接重新訂閱
? ? ? ? ? ? log.info("監(jiān)聽(tīng)遇到異常，四秒后重新訂閱頻道：");
? ? ? ? ? ? Arrays.asList(channels).forEach(s -> {log.info(s);});
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? Thread.sleep(4000);
? ? ? ? ? ? } catch (InterruptedException interruptedException) {
? ? ? ? ? ? ? ? interruptedException.printStackTrace();
? ? ? ? ? ? }
? ? ? ? ? ? subscribe(channels);
? ? ? ? }
? ? }
? ? //模糊訂閱
? ? public void psubscribe(String... channels) {
? ? ? ? Jedis jedis = jedisPool.getResource();
? ? ? ? try {
? ? ? ? ? ? jedis.psubscribe(this, channels);
? ? ? ? } catch (ArithmeticException e) {//取消訂閱故意造成的異常
? ? ? ? ? ? if (jedis != null)
? ? ? ? ? ? ? ? jedis.close();
? ? ? ? } catch (Exception e) {
? ? ? ? ? ? log.error(e.getMessage());
? ? ? ? ? ? if (jedis != null)
? ? ? ? ? ? ? ? jedis.close();
? ? ? ? ? ? //遇到異常后關(guān)閉連接重新訂閱
? ? ? ? ? ? log.info("監(jiān)聽(tīng)遇到異常，四秒后重新訂閱頻道：");
? ? ? ? ? ? Arrays.asList(channels).forEach(s -> {log.info(s);});
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? Thread.sleep(4000);
? ? ? ? ? ? } catch (InterruptedException interruptedException) {
? ? ? ? ? ? ? ? interruptedException.printStackTrace();
? ? ? ? ? ? }
? ? ? ? ? ? psubscribe(channels);
? ? ? ? }
? ? }
? ? public void unsubscribeAndClose(String... channels){
? ? ? ? unsubscribe(channels);
? ? ? ? if (jedis != null && !isSubscribed())
? ? ? ? ? ? jedis.close();
? ? }
? ? public void punsubscribeAndClose(String... channels){
? ? ? ? punsubscribe(channels);
? ? ? ? if (jedis != null && !isSubscribed())
? ? ? ? ? ? jedis.close();
? ? }
? ? @Override
? ? public void onSubscribe(String channel, int subscribedChannels) {
? ? ? ? log.info("subscribe redis channel：" + channel + ", 線程id：" + Thread.currentThread().getId());
? ? }
? ? @Override
? ? public void onPSubscribe(String pattern, int subscribedChannels) {
? ? ? ? log.info("psubscribe redis channel：" + pattern + ", 線程id：" + Thread.currentThread().getId());
? ? }
? ? @Override
? ? public void onPMessage(String pattern, String channel, String message) {
? ? ? ? log.info("receive from redis channal: " + channel + ",pattern: " + pattern + ",message：" + message + ", 線程id：" + Thread.currentThread().getId());
? ? ? ? WebSocketServer.publish(message, pattern);
? ? ? ? WebSocketServer.publish(message, channel);
? ? }
? ? @Override
? ? public void onMessage(String channel, String message) {
? ? ? ? log.info("receive from redis channal: " + channel + ",message：" + message + ", 線程id：" + Thread.currentThread().getId());
? ? ? ? WebSocketServer.publish(message, channel);
? ? }
? ? @Override
? ? public void onUnsubscribe(String channel, int subscribedChannels) {
? ? ? ? log.info("unsubscribe redis channel：" + channel);
? ? }
? ? @Override
? ? public void onPUnsubscribe(String pattern, int subscribedChannels) {
? ? ? ? log.info("punsubscribe redis channel：" + pattern);
? ? }
}1.jedis監(jiān)聽(tīng)redis頻道的時(shí)候如果遇見(jiàn)異常會(huì)關(guān)閉連接導(dǎo)致后續(xù)沒(méi)有監(jiān)聽(tīng)該頻道，所以這里在subscribe捕獲到異常的時(shí)候會(huì)重新創(chuàng)建一個(gè)jedis連接訂閱該redis頻道。
webSocket訂閱推送類(lèi)
這個(gè)類(lèi)會(huì)有兩個(gè)ConcurrentHashMap<String, ConcurrentHashMap<String, WebSocketServer>>類(lèi)型類(lèi)變量，分別存儲(chǔ)訂閱和模糊訂閱的信息。
外面一層的String對(duì)應(yīng)的值是topic_name,里面一層的String對(duì)應(yīng)的值是sessionId。前端發(fā)送過(guò)來(lái)的消息里面對(duì)應(yīng)的這三類(lèi)操作其實(shí)就是對(duì)這兩個(gè)map里面的。
還有個(gè)ConcurrentHashMap<String, RedisPubSub>類(lèi)型的變量，存儲(chǔ)的是事件-RedisPubSub，便于取消訂閱的時(shí)候找到監(jiān)聽(tīng)該頻道（事件）的RedisPubSub對(duì)象。
信息進(jìn)行增加或者刪除;后端往前端推送數(shù)據(jù)也會(huì)根據(jù)不同的topic_name推送到不同的訂閱者這邊。
package com.curtain.core;
import com.alibaba.fastjson.JSON;
import com.curtain.config.WebsocketProperties;
import com.curtain.service.Cancelable;
import com.curtain.service.impl.TaskExecuteService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
/**
?* @Author Curtain
?* @Date 2021/5/14 16:49
?* @Description
?*/
@ServerEndpoint("/ws")
@Component
@Slf4j
public class WebSocketServer {
? ? /**
? ? ?* concurrent包的線程安全Set，用來(lái)存放每個(gè)客戶(hù)端對(duì)應(yīng)的MyWebSocket對(duì)象。
? ? ?*/
? ? private static volatile ConcurrentHashMap<String, ConcurrentHashMap<String, WebSocketServer>> webSocketMap = new ConcurrentHashMap<>();
? ? /**
? ? ?* 存放psub的事件
? ? ?**/
? ? private static volatile ConcurrentHashMap<String, ConcurrentHashMap<String, WebSocketServer>> pWebSocketMap = new ConcurrentHashMap<>();
? ? /**
? ? ?* 存放topic(pattern)-對(duì)應(yīng)的RedisPubsub
? ? ?*/
? ? private static volatile ConcurrentHashMap<String, RedisPubSub> redisPubSubMap = new ConcurrentHashMap<>();
? ? /**
? ? ?* 與某個(gè)客戶(hù)端的連接會(huì)話(huà)，需要通過(guò)它來(lái)給客戶(hù)端發(fā)送數(shù)據(jù)
? ? ?*/
? ? private Session session;
? ? private String sessionId = "";
? ? //要注入的對(duì)象
? ? private static TaskExecuteService executeService;
? ? private static WebsocketProperties properties;
? ? private Cancelable cancelable;
? ? @Autowired
? ? public void setTaskExecuteService(TaskExecuteService taskExecuteService) {
? ? ? ? WebSocketServer.executeService = taskExecuteService;
? ? }
? ? @Autowired
? ? public void setWebsocketProperties(WebsocketProperties properties) {
? ? ? ? WebSocketServer.properties = properties;
? ? }
? ? /**
? ? ?* 連接建立成功調(diào)用的方法
? ? ?*/
? ? @OnOpen
? ? public void onOpen(Session session) {
? ? ? ? this.session = session;
? ? ? ? this.sessionId = session.getId();
? ? ? ? //構(gòu)造推送數(shù)據(jù)
? ? ? ? Map pubHeader = new HashMap();
? ? ? ? pubHeader.put("name", "connect_status");
? ? ? ? pubHeader.put("type", "create");
? ? ? ? pubHeader.put("from", "pubsub");
? ? ? ? pubHeader.put("time", new Date().getTime() / 1000);
? ? ? ? Map pubPayload = new HashMap();
? ? ? ? pubPayload.put("status", "success");
? ? ? ? Map pubMap = new HashMap();
? ? ? ? pubMap.put("header", pubHeader);
? ? ? ? pubMap.put("payload", pubPayload);
? ? ? ? sendMessage(JSON.toJSONString(pubMap));
? ? ? ? cancelable = executeService.runPeriodly(() -> {
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? if (cancelable != null && !session.isOpen()) {
? ? ? ? ? ? ? ? ? ? log.info("斷開(kāi)連接，停止發(fā)送ping");
? ? ? ? ? ? ? ? ? ? cancelable.cancel();
? ? ? ? ? ? ? ? } else {
? ? ? ? ? ? ? ? ? ? String data = "ping";
? ? ? ? ? ? ? ? ? ? ByteBuffer payload = ByteBuffer.wrap(data.getBytes());
? ? ? ? ? ? ? ? ? ? session.getBasicRemote().sendPing(payload);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? } catch (IOException e) {
? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? }
? ? ? ? }, properties.getPeriod());
? ? }
? ? @OnMessage
? ? public void onMessage(String message) {
? ? ? ? synchronized (session) {
? ? ? ? ? ? Map msgMap = (Map) JSON.parse(message);
? ? ? ? ? ? String cmd = (String) msgMap.get("cmd");
? ? ? ? ? ? //訂閱消息
? ? ? ? ? ? if ("subscribe".equals(cmd)) {
? ? ? ? ? ? ? ? List<String> topics = (List<String>) msgMap.get("topic");
? ? ? ? ? ? ? ? //本地記錄訂閱信息
? ? ? ? ? ? ? ? for (int i = 0; i < topics.size(); i++) {
? ? ? ? ? ? ? ? ? ? String topic = topics.get(i);
? ? ? ? ? ? ? ? ? ? log.info("============================subscribe-start============================");
? ? ? ? ? ? ? ? ? ? log.info("sessionId：" + this.sessionId + "，開(kāi)始訂閱：" + topic);
? ? ? ? ? ? ? ? ? ? if (webSocketMap.containsKey(topic)) {//有人訂閱過(guò)了
? ? ? ? ? ? ? ? ? ? ? ? webSocketMap.get(topic).put(this.sessionId, this);
? ? ? ? ? ? ? ? ? ? } else {//之前還沒(méi)人訂閱過(guò)，所以需要訂閱redis頻道
? ? ? ? ? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = new ConcurrentHashMap<>();
? ? ? ? ? ? ? ? ? ? ? ? map.put(this.sessionId, this);
? ? ? ? ? ? ? ? ? ? ? ? webSocketMap.put(topic, map);
? ? ? ? ? ? ? ? ? ? ? ? new Thread(() -> {
? ? ? ? ? ? ? ? ? ? ? ? ? ? RedisPubSub redisPubSub = new RedisPubSub();
? ? ? ? ? ? ? ? ? ? ? ? ? ? //存入map
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSubMap.put(topic, redisPubSub);
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSub.subscribe(topic);
? ? ? ? ? ? ? ? ? ? ? ? }).start();
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? log.info("sessionId：" + this.sessionId + "，完成訂閱：" + topic);
? ? ? ? ? ? ? ? ? ? log();
? ? ? ? ? ? ? ? ? ? log.info("============================subscribe-end============================");
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? ? ? //psubscribe
? ? ? ? ? ? if ("psubscribe".equals(cmd)) {
? ? ? ? ? ? ? ? List<String> topics = (List<String>) msgMap.get("topic");
? ? ? ? ? ? ? ? //本地記錄訂閱信息
? ? ? ? ? ? ? ? for (int i = 0; i < topics.size(); i++) {
? ? ? ? ? ? ? ? ? ? String topic = topics.get(i);
? ? ? ? ? ? ? ? ? ? log.info("============================psubscribe-start============================");
? ? ? ? ? ? ? ? ? ? log.info("sessionId：" + this.sessionId + "，開(kāi)始模糊訂閱：" + topic);
? ? ? ? ? ? ? ? ? ? if (pWebSocketMap.containsKey(topic)) {//有人訂閱過(guò)了
? ? ? ? ? ? ? ? ? ? ? ? pWebSocketMap.get(topic).put(this.sessionId, this);
? ? ? ? ? ? ? ? ? ? } else {//之前還沒(méi)人訂閱過(guò)，所以需要訂閱redis頻道
? ? ? ? ? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = new ConcurrentHashMap<>();
? ? ? ? ? ? ? ? ? ? ? ? map.put(this.sessionId, this);
? ? ? ? ? ? ? ? ? ? ? ? pWebSocketMap.put(topic, map);
? ? ? ? ? ? ? ? ? ? ? ? new Thread(() -> {
? ? ? ? ? ? ? ? ? ? ? ? ? ? RedisPubSub redisPubSub = new RedisPubSub();
? ? ? ? ? ? ? ? ? ? ? ? ? ? //存入map
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSubMap.put(topic, redisPubSub);
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSub.psubscribe(topic);
? ? ? ? ? ? ? ? ? ? ? ? }).start();
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? log.info("sessionId：" + this.sessionId + "，完成模糊訂閱：" + topic);
? ? ? ? ? ? ? ? ? ? log();
? ? ? ? ? ? ? ? ? ? log.info("============================psubscribe-end============================");
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? ? ? //取消訂閱
? ? ? ? ? ? if ("unsubscribe".equals(cmd)) {
? ? ? ? ? ? ? ? List<String> topics = (List<String>) msgMap.get("topic");
? ? ? ? ? ? ? ? //刪除本地對(duì)應(yīng)的訂閱信息
? ? ? ? ? ? ? ? for (String topic : topics) {
? ? ? ? ? ? ? ? ? ? log.info("============================unsubscribe-start============================");
? ? ? ? ? ? ? ? ? ? log.info("sessionId：" + this.sessionId + "，開(kāi)始刪除訂閱：" + topic);
? ? ? ? ? ? ? ? ? ? if (webSocketMap.containsKey(topic)) {
? ? ? ? ? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = webSocketMap.get(topic);
? ? ? ? ? ? ? ? ? ? ? ? map.remove(this.sessionId);
? ? ? ? ? ? ? ? ? ? ? ? if (map.size() == 0) {//如果這個(gè)頻道沒(méi)有用戶(hù)訂閱了，則取消訂閱該redis頻道
? ? ? ? ? ? ? ? ? ? ? ? ? ? webSocketMap.remove(topic);
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSubMap.get(topic).unsubscribeAndClose(topic);
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSubMap.remove(topic);
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? if (pWebSocketMap.containsKey(topic)) {
? ? ? ? ? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = pWebSocketMap.get(topic);
? ? ? ? ? ? ? ? ? ? ? ? map.remove(this.sessionId);
? ? ? ? ? ? ? ? ? ? ? ? if (map.size() == 0) {//如果這個(gè)頻道沒(méi)有用戶(hù)訂閱了，則取消訂閱該redis頻道
? ? ? ? ? ? ? ? ? ? ? ? ? ? pWebSocketMap.remove(topic);
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSubMap.get(topic).punsubscribeAndClose(topic);
? ? ? ? ? ? ? ? ? ? ? ? ? ? redisPubSubMap.remove(topic);
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? log.info("sessionId：" + this.sessionId + "，完成刪除訂閱：" + topic);
? ? ? ? ? ? ? ? ? ? log();
? ? ? ? ? ? ? ? ? ? log.info("============================unsubscribe-end============================");
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? }
? ? @OnMessage
? ? public void onPong(PongMessage pongMessage) {
? ? ? ? try {
? ? ? ? ? ? log.debug(new String(pongMessage.getApplicationData().array(), "utf-8") + "接收到pong");
? ? ? ? } catch (UnsupportedEncodingException e) {
? ? ? ? ? ? e.printStackTrace();
? ? ? ? }
? ? }
? ? /**
? ? ?* 連接關(guān)閉調(diào)用的方法
? ? ?*/
? ? @OnClose
? ? public void onClose() {
? ? ? ? synchronized (session) {
? ? ? ? ? ? log.info("============================onclose-start============================");
? ? ? ? ? ? //刪除訂閱
? ? ? ? ? ? Iterator iterator = webSocketMap.keySet().iterator();
? ? ? ? ? ? while (iterator.hasNext()) {
? ? ? ? ? ? ? ? String topic = (String) iterator.next();
? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = webSocketMap.get(topic);
? ? ? ? ? ? ? ? map.remove(this.sessionId);
? ? ? ? ? ? ? ? if (map.size() == 0) {//如果這個(gè)頻道沒(méi)有用戶(hù)訂閱了，則取消訂閱該redis頻道
? ? ? ? ? ? ? ? ? ? webSocketMap.remove(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.get(topic).unsubscribeAndClose(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.remove(topic);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? ? ? //刪除模糊訂閱
? ? ? ? ? ? Iterator iteratorP = pWebSocketMap.keySet().iterator();
? ? ? ? ? ? while (iteratorP.hasNext()) {
? ? ? ? ? ? ? ? String topic = (String) iteratorP.next();
? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = pWebSocketMap.get(topic);
? ? ? ? ? ? ? ? map.remove(this.sessionId);
? ? ? ? ? ? ? ? if (map.size() == 0) {//如果這個(gè)頻道沒(méi)有用戶(hù)訂閱了，則取消訂閱該redis頻道
? ? ? ? ? ? ? ? ? ? pWebSocketMap.remove(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.get(topic).punsubscribeAndClose(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.remove(topic);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? ? ? log.info("sessionId：" + this.sessionId + "，斷開(kāi)連接：");
? ? ? ? ? ? //debug
? ? ? ? ? ? log();
? ? ? ? ? ? log.info("============================onclose-end============================");
? ? ? ? }
? ? }
? ? /**
? ? ?* @param session
? ? ?* @param error
? ? ?*/
? ? @OnError
? ? public void onError(Session session, Throwable error) {
? ? ? ? synchronized (session) {
? ? ? ? ? ? log.info("============================onError-start============================");
? ? ? ? ? ? log.error("用戶(hù)錯(cuò)誤,sessionId:" + session.getId() + ",原因:" + error.getMessage());
? ? ? ? ? ? error.printStackTrace();
? ? ? ? ? ? log.info("關(guān)閉錯(cuò)誤用戶(hù)對(duì)應(yīng)的連接");
? ? ? ? ? ? //刪除訂閱
? ? ? ? ? ? Iterator iterator = webSocketMap.keySet().iterator();
? ? ? ? ? ? while (iterator.hasNext()) {
? ? ? ? ? ? ? ? String topic = (String) iterator.next();
? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = webSocketMap.get(topic);
? ? ? ? ? ? ? ? map.remove(this.sessionId);
? ? ? ? ? ? ? ? if (map.size() == 0) {//如果這個(gè)頻道沒(méi)有用戶(hù)訂閱了，則取消訂閱該redis頻道
? ? ? ? ? ? ? ? ? ? webSocketMap.remove(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.get(topic).unsubscribeAndClose(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.remove(topic);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? ? ? //刪除模糊訂閱
? ? ? ? ? ? Iterator iteratorP = pWebSocketMap.keySet().iterator();
? ? ? ? ? ? while (iteratorP.hasNext()) {
? ? ? ? ? ? ? ? String topic = (String) iteratorP.next();
? ? ? ? ? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = pWebSocketMap.get(topic);
? ? ? ? ? ? ? ? map.remove(this.sessionId);
? ? ? ? ? ? ? ? if (map.size() == 0) {//如果這個(gè)頻道沒(méi)有用戶(hù)訂閱了，則取消訂閱該redis頻道
? ? ? ? ? ? ? ? ? ? pWebSocketMap.remove(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.get(topic).punsubscribeAndClose(topic);
? ? ? ? ? ? ? ? ? ? redisPubSubMap.remove(topic);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? ? ? log.info("完成錯(cuò)誤用戶(hù)對(duì)應(yīng)的連接關(guān)閉");
? ? ? ? ? ? //debug
? ? ? ? ? ? log();
? ? ? ? ? ? log.info("============================onError-end============================");
? ? ? ? }
? ? }
? ? /**
? ? ?* 實(shí)現(xiàn)服務(wù)器主動(dòng)推送
? ? ?*/
? ? public void sendMessage(String message) {
? ? ? ? synchronized (session) {
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? this.session.getBasicRemote().sendText(message);
? ? ? ? ? ? } catch (IOException e) {
? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? }
? ? ? ? }
? ? }
? ? public static void publish(String msg, String topic) {
? ? ? ? ConcurrentHashMap<String, WebSocketServer> map = webSocketMap.get(topic);
? ? ? ? if (map != null && map.values() != null) {
? ? ? ? ? ? for (WebSocketServer webSocketServer : map.values())
? ? ? ? ? ? ? ? webSocketServer.sendMessage(msg);
? ? ? ? }
? ? ? ? map = pWebSocketMap.get(topic);
? ? ? ? if (map != null && map.values() != null) {
? ? ? ? ? ? for (WebSocketServer webSocketServer : map.values())
? ? ? ? ? ? ? ? webSocketServer.sendMessage(msg);
? ? ? ? }
? ? }
? ? private void log() {
? ? ? ? log.info("<<<<<<<<<<<完成操作后，打印訂閱信息開(kāi)始>>>>>>>>>>");
? ? ? ? Iterator iterator1 = webSocketMap.keySet().iterator();
? ? ? ? while (iterator1.hasNext()) {
? ? ? ? ? ? String topic = (String) iterator1.next();
? ? ? ? ? ? log.info("topic：" + topic);
? ? ? ? ? ? Iterator iterator2 = webSocketMap.get(topic).keySet().iterator();
? ? ? ? ? ? while (iterator2.hasNext()) {
? ? ? ? ? ? ? ? String session = (String) iterator2.next();
? ? ? ? ? ? ? ? log.info("訂閱" + topic + "的sessionId：" + session);
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? log.info("<<<<<<<<<<<完成操作后，打印訂閱信息結(jié)束>>>>>>>>>>");
? ? }
}項(xiàng)目地址
上面介紹了核心代碼，下面是完整代碼地址
https://github.com/Curtain-Wang/websocket-redis-subscribe.git
Update20220415
參考評(píng)論區(qū)老哥的建議，將redis訂閱監(jiān)聽(tīng)類(lèi)里面的subscribe和psubscribe方法調(diào)整如下：
? ? //訂閱
? ? @Override
? ? public void subscribe(String... channels) {
? ? ? ? boolean done = true;
? ? ? ? while (done){
? ? ? ? ? ? Jedis jedis = jedisPool.getResource();
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? jedis.subscribe(this, channels);
? ? ? ? ? ? ? ? done = false;
? ? ? ? ? ? } catch (Exception e) {
? ? ? ? ? ? ? ? log.error(e.getMessage());
? ? ? ? ? ? ? ? if (jedis != null)
? ? ? ? ? ? ? ? ? ? jedis.close();
? ? ? ? ? ? ? ? //遇到異常后關(guān)閉連接重新訂閱
? ? ? ? ? ? ? ? log.info("監(jiān)聽(tīng)遇到異常，四秒后重新訂閱頻道：");
? ? ? ? ? ? ? ? Arrays.asList(channels).forEach(s -> {log.info(s);});
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? Thread.sleep(4000);
? ? ? ? ? ? ? ? } catch (InterruptedException interruptedException) {
? ? ? ? ? ? ? ? ? ? interruptedException.printStackTrace();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? }
? ? //模糊訂閱
? ? @Override
? ? public void psubscribe(String... channels) {
? ? ? ? boolean done = true;
? ? ? ? while (done){
? ? ? ? ? ? Jedis jedis = jedisPool.getResource();
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? jedis.psubscribe(this, channels);
? ? ? ? ? ? ? ? done = false;
? ? ? ? ? ? } catch (Exception e) {
? ? ? ? ? ? ? ? log.error(e.getMessage());
? ? ? ? ? ? ? ? if (jedis != null)
? ? ? ? ? ? ? ? ? ? jedis.close();
? ? ? ? ? ? ? ? //遇到異常后關(guān)閉連接重新訂閱
? ? ? ? ? ? ? ? log.info("監(jiān)聽(tīng)遇到異常，四秒后重新訂閱頻道：");
? ? ? ? ? ? ? ? Arrays.asList(channels).forEach(s -> {log.info(s);});
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? Thread.sleep(4000);
? ? ? ? ? ? ? ? } catch (InterruptedException interruptedException) {
? ? ? ? ? ? ? ? ? ? interruptedException.printStackTrace();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? }到此這篇關(guān)于websocket+redis動(dòng)態(tài)訂閱和動(dòng)態(tài)取消訂閱的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)websocket redis動(dòng)態(tài)訂閱 內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
在很多時(shí)候,我們遇到用戶(hù)簽到的場(chǎng)景,本文主要介紹了Redis應(yīng)用之簽到的使用,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2024-05-05
本文主要介紹了redis中bind配置的詳細(xì)步驟，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-07-07
這篇文章主要介紹了在redhat6.4安裝redis集群【教程】,需要的朋友可以參考下2016-05-05
這篇文章主要介紹了Redis Stat的安裝指南的相關(guān)資料,需要的朋友可以參考下2016-04-04
這篇文章主要為大家詳細(xì)介紹了Windows下Redis安裝配置教程,具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2016-11-11
Redis對(duì)批量數(shù)據(jù)實(shí)現(xiàn)分布式鎖的實(shí)現(xiàn)代碼
為了防止多人多電腦同時(shí)操作一條數(shù)據(jù)，我們自己開(kāi)發(fā)了一個(gè)簡(jiǎn)單的基于Redis實(shí)現(xiàn)的分布式鎖，Redis對(duì)批量數(shù)據(jù)實(shí)現(xiàn)分布式鎖相關(guān)知識(shí)感興趣的朋友一起看看吧2022-03-03
將MongoDB作為Redis式的內(nèi)存數(shù)據(jù)庫(kù)的使用方法
這篇文章主要介紹了將MongoDB作為Redis式的內(nèi)存數(shù)據(jù)庫(kù)的使用方法,原理其實(shí)只是將內(nèi)存虛擬作為磁盤(pán),需要的朋友可以參考下2015-06-06
這篇文章主要介紹了redis 交集、并集、差集的具體使用，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-02-02
本文主要介紹了Redis提供的三大持久化機(jī)制,即AOF日志、RDB快照以及混合持久化機(jī)制,結(jié)合圖文實(shí)例給大家講解的非常詳細(xì),感興趣的朋友一起看看吧2025-02-02
redis實(shí)現(xiàn)sentinel哨兵架構(gòu)的方法
哨兵是一個(gè)分布式系統(tǒng)，可以在一個(gè)架構(gòu)中運(yùn)行多個(gè)哨兵(sentinel) 進(jìn)程，這些進(jìn)程使用流言協(xié)議(gossip protocols)來(lái)接收關(guān)于Master主服務(wù)器是否下線的信息，這篇文章主要介紹了redis實(shí)現(xiàn)sentinel哨兵架構(gòu),需要的朋友可以參考下2022-11-11