websocket+redis动态订阅和动态取消订阅的实现示例_Redis_

{ "cmd":"subscribe", "topic":[ "topic_name" ] } 

{ "cmd":"psubscribe", "topic":[ "topic_name" ] } 

{ "cmd":"unsubscribe", "topic":[ "topic_name" ] } 

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();             //遇到异常后关闭连接重新订阅             log.info("监听遇到异常，四秒后重新订阅频道：");             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();             //遇到异常后关闭连接重新订阅             log.info("监听遇到异常，四秒后重新订阅频道：");             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);     } }

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，用来存放每个客户端对应的MyWebSocket对象。      */     private static volatile ConcurrentHashMap> webSocketMap = new ConcurrentHashMap<>();     /**      * 存放psub的事件      **/     private static volatile ConcurrentHashMap> pWebSocketMap = new ConcurrentHashMap<>();     /**      * 存放topic(pattern)-对应的RedisPubsub      */     private static volatile ConcurrentHashMap redisPubSubMap = new ConcurrentHashMap<>();     /**      * 与某个客户端的连接会话，需要通过它来给客户端发送数据      */     private Session session;     private String sessionId = "";     //要注入的对象     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;     }     /**      * 连接建立成功调用的方法      */     @OnOpen     public void onOpen(Session session) {         this.session = session;         this.sessionId = session.getId();         //构造推送数据         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("断开连接，停止发送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 topics = (List) 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 + "，开始订阅：" + topic);                     if (webSocketMap.containsKey(topic)) {//有人订阅过了                         webSocketMap.get(topic).put(this.sessionId, this);                     } else {//之前还没人订阅过，所以需要订阅redis频道                         ConcurrentHashMap 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 topics = (List) msgMap.get("topic"); 
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：手把手教你用Redis 实现点赞功能并且与数据库同步_Redis_
                                                                
下一篇：浅谈Redis中的自动过期机制_Redis_