{"id":44163,"date":"2024-12-01T14:15:21","date_gmt":"2024-12-01T06:15:21","guid":{"rendered":"https:\/\/fwq.ai\/blog\/44163\/"},"modified":"2024-12-01T14:15:21","modified_gmt":"2024-12-01T06:15:21","slug":"websocketredis%e5%8a%a8%e6%80%81%e8%ae%a2%e9%98%85%e5%92%8c%e5%8a%a8%e6%80%81%e5%8f%96%e6%b6%88%e8%ae%a2%e9%98%85%e7%9a%84%e5%ae%9e%e7%8e%b0%e7%a4%ba%e4%be%8b","status":"publish","type":"post","link":"https:\/\/fwq.ai\/blog\/44163\/","title":{"rendered":"websocket+redis\u52a8\u6001\u8ba2\u9605\u548c\u52a8\u6001\u53d6\u6d88\u8ba2\u9605\u7684\u5b9e\u73b0\u793a\u4f8b"},"content":{"rendered":"

<\/b> <\/p>\n

websocket+redis\u52a8\u6001\u8ba2\u9605\u548c\u52a8\u6001\u53d6\u6d88\u8ba2\u9605\u7684\u5b9e\u73b0\u793a\u4f8b<\/h1>\n

<\/i>0\u6d4f\u89c8<\/span>
\n<\/i>\u6536\u85cf<\/span> <\/p>\n

\u600e\u4e48\u5165\u95e8\u6570\u636e\u5e93\u7f16\u7a0b\uff1f\u9700\u8981\u5b66\u4e60\u54ea\u4e9b\u77e5\u8bc6\u70b9\uff1f\u8fd9\u662f\u65b0\u624b\u4eec\u521a\u63a5\u89e6\u7f16\u7a0b\u65f6\u5e38\u89c1\u7684\u95ee\u9898\uff1b\u4e0b\u9762golang\u5b66\u4e60\u7f51\u5c31\u6765\u7ed9\u5927\u5bb6\u6574\u7406\u5206\u4eab\u4e00\u4e9b\u77e5\u8bc6\u70b9\uff0c\u5e0c\u671b\u80fd\u591f\u7ed9\u521d\u5b66\u8005\u4e00\u4e9b\u5e2e\u52a9\u3002\u672c\u7bc7\u6587\u7ae0\u5c31\u6765\u4ecb\u7ecd\u300awebsocket+redis\u52a8\u6001\u8ba2\u9605\u548c\u52a8\u6001\u53d6\u6d88\u8ba2\u9605\u7684\u5b9e\u73b0\u793a\u4f8b\u300b\uff0c\u6d89\u53ca\u5230websocketredis\u3001\u52a8\u6001\u8ba2\u9605\uff0c\u6709\u9700\u8981\u7684\u53ef\u4ee5\u6536\u85cf\u4e00\u4e0b<\/p>\n

\u539f\u7406<\/h2>\n

websocket\u7684\u8ba2\u9605\u5c31\u662f\u5728\u524d\u540e\u7aef\u5efa\u7acbws\u8fde\u63a5\u4e4b\u540e\uff0c\u524d\u7aef\u901a\u8fc7\u53d1\u9001\u4e00\u5b9a\u683c\u5f0f\u7684\u6d88\u606f\uff0c\u540e\u7aef\u89e3\u6790\u51fa\u6765\u53bb\u8ba2\u9605\u6216\u8005\u53d6\u6d88\u8ba2\u9605redis\u9891\u9053\u3002<\/p>\n

\u8ba2\u9605\u9891\u9053\u6d88\u606f\u683c\u5f0f:<\/p>\n

{\n    \"cmd\":\"subscribe\",\n    \"topic\":[\n        \"topic_name\"\n    ]\n}\n<\/pre>\n
\u6a21\u7cca\u8ba2\u9605\u683c\u5f0f<\/p>\n
{\n    \"cmd\":\"psubscribe\",\n    \"topic\":[\n        \"topic_name\"\n    ]\n}\n<\/pre>\n
\u53d6\u6d88\u8ba2\u9605\u683c\u5f0f<\/p>\n
{\n    \"cmd\":\"unsubscribe\",\n    \"topic\":[\n        \"topic_name\"\n    ]\n}\n<\/pre>\n
\u4e24\u4e2a\u6838\u5fc3\u7c7b\uff0c\u4e00\u4e2a\u662fredis\u7684\u8ba2\u9605\u76d1\u542c\u7c7b\uff0c\u4e00\u4e2a\u662fwebsocket\u7684\u53d1\u5e03\u8ba2\u9605\u7c7b\u3002<\/p>\n
redis\u8ba2\u9605\u76d1\u542c\u7c7b<\/h2>\n
package com.curtain.core;\n\nimport com.curtain.config.GetBeanUtil;\nimport lombok.extern.slf4j.Slf4j;\nimport org.springframework.stereotype.Component;\nimport redis.clients.jedis.Jedis;\nimport redis.clients.jedis.JedisPool;\nimport redis.clients.jedis.JedisPubSub;\n\nimport java.util.Arrays;\n\n\/**\n * @Author Curtain\n * @Date 2021\/6\/7 14:27\n * @Description\n *\/\n@Component\n@Slf4j\npublic class RedisPubSub extends JedisPubSub {\n    private JedisPool jedisPool = GetBeanUtil.getBean(JedisPool.class);\n    private Jedis jedis;\n\n    \/\/\u8ba2\u9605\n    public void subscribe(String... channels) {\n        jedis = jedisPool.getResource();\n        try {\n            jedis.subscribe(this, channels);\n        } catch (Exception e) {\n            log.error(e.getMessage());\n            if (jedis != null)\n                jedis.close();\n            \/\/\u9047\u5230\u5f02\u5e38\u540e\u5173\u95ed\u8fde\u63a5\u91cd\u65b0\u8ba2\u9605\n            log.info(\"\u76d1\u542c\u9047\u5230\u5f02\u5e38\uff0c\u56db\u79d2\u540e\u91cd\u65b0\u8ba2\u9605\u9891\u9053\uff1a\");\n            Arrays.asList(channels).forEach(s -> {log.info(s);});\n            try {\n                Thread.sleep(4000);\n            } catch (InterruptedException interruptedException) {\n                interruptedException.printStackTrace();\n            }\n            subscribe(channels);\n        }\n    }\n\n    \/\/\u6a21\u7cca\u8ba2\u9605\n    public void psubscribe(String... channels) {\n        Jedis jedis = jedisPool.getResource();\n        try {\n            jedis.psubscribe(this, channels);\n        } catch (ArithmeticException e) {\/\/\u53d6\u6d88\u8ba2\u9605\u6545\u610f\u9020\u6210\u7684\u5f02\u5e38\n            if (jedis != null)\n                jedis.close();\n        } catch (Exception e) {\n            log.error(e.getMessage());\n            if (jedis != null)\n                jedis.close();\n            \/\/\u9047\u5230\u5f02\u5e38\u540e\u5173\u95ed\u8fde\u63a5\u91cd\u65b0\u8ba2\u9605\n            log.info(\"\u76d1\u542c\u9047\u5230\u5f02\u5e38\uff0c\u56db\u79d2\u540e\u91cd\u65b0\u8ba2\u9605\u9891\u9053\uff1a\");\n            Arrays.asList(channels).forEach(s -> {log.info(s);});\n            try {\n                Thread.sleep(4000);\n            } catch (InterruptedException interruptedException) {\n                interruptedException.printStackTrace();\n            }\n            psubscribe(channels);\n        }\n    }\n\n    public void unsubscribeAndClose(String... channels){\n        unsubscribe(channels);\n        if (jedis != null && !isSubscribed())\n            jedis.close();\n    }\n\n    public void punsubscribeAndClose(String... channels){\n        punsubscribe(channels);\n        if (jedis != null && !isSubscribed())\n            jedis.close();\n    }\n\n    @Override\n    public void onSubscribe(String channel, int subscribedChannels) {\n        log.info(\"subscribe redis channel\uff1a\" + channel + \", \u7ebf\u7a0bid\uff1a\" + Thread.currentThread().getId());\n    }\n\n    @Override\n    public void onPSubscribe(String pattern, int subscribedChannels) {\n        log.info(\"psubscribe redis channel\uff1a\" + pattern + \", \u7ebf\u7a0bid\uff1a\" + Thread.currentThread().getId());\n    }\n\n    @Override\n    public void onPMessage(String pattern, String channel, String message) {\n        log.info(\"receive from redis channal: \" + channel + \",pattern: \" + pattern + \",message\uff1a\" + message + \", \u7ebf\u7a0bid\uff1a\" + Thread.currentThread().getId());\n        WebSocketServer.publish(message, pattern);\n        WebSocketServer.publish(message, channel);\n\n    }\n\n    @Override\n    public void onMessage(String channel, String message) {\n        log.info(\"receive from redis channal: \" + channel + \",message\uff1a\" + message + \", \u7ebf\u7a0bid\uff1a\" + Thread.currentThread().getId());\n        WebSocketServer.publish(message, channel);\n    }\n\n    @Override\n    public void onUnsubscribe(String channel, int subscribedChannels) {\n        log.info(\"unsubscribe redis channel\uff1a\" + channel);\n    }\n\n    @Override\n    public void onPUnsubscribe(String pattern, int subscribedChannels) {\n        log.info(\"punsubscribe redis channel\uff1a\" + pattern);\n    }\n}<\/pre>\n
1.jedis\u76d1\u542credis\u9891\u9053\u7684\u65f6\u5019\u5982\u679c\u9047\u89c1\u5f02\u5e38\u4f1a\u5173\u95ed\u8fde\u63a5\u5bfc\u81f4\u540e\u7eed\u6ca1\u6709\u76d1\u542c\u8be5\u9891\u9053\uff0c\u6240\u4ee5\u8fd9\u91cc\u5728subscribe\u6355\u83b7\u5230\u5f02\u5e38\u7684\u65f6\u5019\u4f1a\u91cd\u65b0\u521b\u5efa\u4e00\u4e2ajedis\u8fde\u63a5\u8ba2\u9605\u8be5redis\u9891\u9053\u3002<\/p>\n
webSocket\u8ba2\u9605\u63a8\u9001\u7c7b<\/h2>\n
\u8fd9\u4e2a\u7c7b\u4f1a\u6709\u4e24\u4e2aConcurrentHashMap >\u7c7b\u578b\u7c7b\u53d8\u91cf\uff0c\u5206\u522b\u5b58\u50a8\u8ba2\u9605\u548c\u6a21\u7cca\u8ba2\u9605\u7684\u4fe1\u606f\u3002 <\/p>\n
\u5916\u9762\u4e00\u5c42\u7684String\u5bf9\u5e94\u7684\u503c\u662ftopic_name,\u91cc\u9762\u4e00\u5c42\u7684String\u5bf9\u5e94\u7684\u503c\u662fsessionId\u3002\u524d\u7aef\u53d1\u9001\u8fc7\u6765\u7684\u6d88\u606f\u91cc\u9762\u5bf9\u5e94\u7684\u8fd9\u4e09\u7c7b\u64cd\u4f5c\u5176\u5b9e\u5c31\u662f\u5bf9\u8fd9\u4e24\u4e2amap\u91cc\u9762\u7684\u3002<\/p>\n
\u8fd8\u6709\u4e2aConcurrentHashMap \u7c7b\u578b\u7684\u53d8\u91cf\uff0c\u5b58\u50a8\u7684\u662f\u4e8b\u4ef6-RedisPubSub\uff0c\u4fbf\u4e8e\u53d6\u6d88\u8ba2\u9605\u7684\u65f6\u5019\u627e\u5230\u76d1\u542c\u8be5\u9891\u9053\uff08\u4e8b\u4ef6\uff09\u7684RedisPubSub\u5bf9\u8c61\u3002 <\/p>\n
\u4fe1\u606f\u8fdb\u884c\u589e\u52a0\u6216\u8005\u5220\u9664;\u540e\u7aef\u5f80\u524d\u7aef\u63a8\u9001\u6570\u636e\u4e5f\u4f1a\u6839\u636e\u4e0d\u540c\u7684topic_name\u63a8\u9001\u5230\u4e0d\u540c\u7684\u8ba2\u9605\u8005\u8fd9\u8fb9\u3002<\/p>\n
package com.curtain.core;\n\nimport com.alibaba.fastjson.JSON;\nimport com.curtain.config.WebsocketProperties;\nimport com.curtain.service.Cancelable;\nimport com.curtain.service.impl.TaskExecuteService;\nimport lombok.extern.slf4j.Slf4j;\nimport org.springframework.beans.factory.annotation.Autowired;\nimport org.springframework.stereotype.Component;\n\nimport javax.websocket.*;\nimport javax.websocket.server.ServerEndpoint;\nimport java.io.IOException;\nimport java.io.UnsupportedEncodingException;\nimport java.nio.ByteBuffer;\nimport java.util.*;\nimport java.util.concurrent.ConcurrentHashMap;\n\n\n\/**\n * @Author Curtain\n * @Date 2021\/5\/14 16:49\n * @Description\n *\/\n@ServerEndpoint(\"\/ws\")\n@Component\n@Slf4j\npublic class WebSocketServer {\n    \/**\n     * concurrent\u5305\u7684\u7ebf\u7a0b\u5b89\u5168Set\uff0c\u7528\u6765\u5b58\u653e\u6bcf\u4e2a\u5ba2\u6237\u7aef\u5bf9\u5e94\u7684MyWebSocket\u5bf9\u8c61\u3002\n     *\/\n    private static volatile ConcurrentHashMap\n   \n    > webSocketMap = new ConcurrentHashMap();\n    \/**\n     * \u5b58\u653epsub\u7684\u4e8b\u4ef6\n     **\/\n    private static volatile ConcurrentHashMap\n    \n     > pWebSocketMap = new ConcurrentHashMap();     \/**      * \u5b58\u653etopic(pattern)-\u5bf9\u5e94\u7684RedisPubsub      *\/     private static volatile ConcurrentHashMap\n     \n       redisPubSubMap = new ConcurrentHashMap();     \/**      * \u4e0e\u67d0\u4e2a\u5ba2\u6237\u7aef\u7684\u8fde\u63a5\u4f1a\u8bdd\uff0c\u9700\u8981\u901a\u8fc7\u5b83\u6765\u7ed9\u5ba2\u6237\u7aef\u53d1\u9001\u6570\u636e      *\/     private Session session;     private String sessionId = \"\";     \/\/\u8981\u6ce8\u5165\u7684\u5bf9\u8c61     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;     }     \/**      * \u8fde\u63a5\u5efa\u7acb\u6210\u529f\u8c03\u7528\u7684\u65b9\u6cd5      *\/     @OnOpen     public void onOpen(Session session) {         this.session = session;         this.sessionId = session.getId();         \/\/\u6784\u9020\u63a8\u9001\u6570\u636e         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(\"\u65ad\u5f00\u8fde\u63a5\uff0c\u505c\u6b62\u53d1\u9001ping\");                     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\");             \/\/\u8ba2\u9605\u6d88\u606f             if (\"subscribe\".equals(cmd)) {                 List\n      \n        topics = (List\n       \n        ) msgMap.get(\"topic\");                 \/\/\u672c\u5730\u8bb0\u5f55\u8ba2\u9605\u4fe1\u606f                 for (int i = 0; i map = new ConcurrentHashMap();                         map.put(this.sessionId, this);                         webSocketMap.put(topic, map);                         new Thread(() -> {                             RedisPubSub redisPubSub = new RedisPubSub();                             \/\/\u5b58\u5165map                             redisPubSubMap.put(topic, redisPubSub);                             redisPubSub.subscribe(topic);                         }).start();                     }                     log.info(\"sessionId\uff1a\" + this.sessionId + \"\uff0c\u5b8c\u6210\u8ba2\u9605\uff1a\" + topic);                     log();                     log.info(\"============================subscribe-end============================\");                 }             }             \/\/psubscribe             if (\"psubscribe\".equals(cmd)) {                 List\n        \n          topics = (List\n         \n          ) msgMap.get(\"topic\");                 \/\/\u672c\u5730\u8bb0\u5f55\u8ba2\u9605\u4fe1\u606f                 for (int i = 0; i map = new ConcurrentHashMap();                         map.put(this.sessionId, this);                         pWebSocketMap.put(topic, map);                         new Thread(() -> {                             RedisPubSub redisPubSub = new RedisPubSub();                             \/\/\u5b58\u5165map                             redisPubSubMap.put(topic, redisPubSub);                             redisPubSub.psubscribe(topic);                         }).start();                     }                     log.info(\"sessionId\uff1a\" + this.sessionId + \"\uff0c\u5b8c\u6210\u6a21\u7cca\u8ba2\u9605\uff1a\" + topic);                     log();                     log.info(\"============================psubscribe-end============================\");                 }             }             \/\/\u53d6\u6d88\u8ba2\u9605             if (\"unsubscribe\".equals(cmd)) {                 List\n          \n            topics = (List\n           \n            ) msgMap.get(\"topic\");                 \/\/\u5220\u9664\u672c\u5730\u5bf9\u5e94\u7684\u8ba2\u9605\u4fe1\u606f                 for (String topic : topics) {                     log.info(\"============================unsubscribe-start============================\");                     log.info(\"sessionId\uff1a\" + this.sessionId + \"\uff0c\u5f00\u59cb\u5220\u9664\u8ba2\u9605\uff1a\" + topic);                     if (webSocketMap.containsKey(topic)) {                         ConcurrentHashMap\n            \n              map = webSocketMap.get(topic);                         map.remove(this.sessionId);                         if (map.size() == 0) {\/\/\u5982\u679c\u8fd9\u4e2a\u9891\u9053\u6ca1\u6709\u7528\u6237\u8ba2\u9605\u4e86\uff0c\u5219\u53d6\u6d88\u8ba2\u9605\u8be5redis\u9891\u9053                             webSocketMap.remove(topic);                             redisPubSubMap.get(topic).unsubscribeAndClose(topic);                             redisPubSubMap.remove(topic);                         }                     }                     if (pWebSocketMap.containsKey(topic)) {                         ConcurrentHashMap\n             \n               map = pWebSocketMap.get(topic);                         map.remove(this.sessionId);                         if (map.size() == 0) {\/\/\u5982\u679c\u8fd9\u4e2a\u9891\u9053\u6ca1\u6709\u7528\u6237\u8ba2\u9605\u4e86\uff0c\u5219\u53d6\u6d88\u8ba2\u9605\u8be5redis\u9891\u9053                             pWebSocketMap.remove(topic);                             redisPubSubMap.get(topic).punsubscribeAndClose(topic);                             redisPubSubMap.remove(topic);                         }                     }                     log.info(\"sessionId\uff1a\" + this.sessionId + \"\uff0c\u5b8c\u6210\u5220\u9664\u8ba2\u9605\uff1a\" + topic);                     log();                     log.info(\"============================unsubscribe-end============================\");                 }             }         }     }     @OnMessage     public void onPong(PongMessage pongMessage) {         try {             log.debug(new String(pongMessage.getApplicationData().array(), \"utf-8\") + \"\u63a5\u6536\u5230pong\");         } catch (UnsupportedEncodingException e) {             e.printStackTrace();         }     }     \/**      * \u8fde\u63a5\u5173\u95ed\u8c03\u7528\u7684\u65b9\u6cd5      *\/     @OnClose     public void onClose() {         synchronized (session) {             log.info(\"============================onclose-start============================\");             \/\/\u5220\u9664\u8ba2\u9605             Iterator iterator = webSocketMap.keySet().iterator();             while (iterator.hasNext()) {                 String topic = (String) iterator.next();                 ConcurrentHashMap\n              \n                map = webSocketMap.get(topic);                 map.remove(this.sessionId);                 if (map.size() == 0) {\/\/\u5982\u679c\u8fd9\u4e2a\u9891\u9053\u6ca1\u6709\u7528\u6237\u8ba2\u9605\u4e86\uff0c\u5219\u53d6\u6d88\u8ba2\u9605\u8be5redis\u9891\u9053                     webSocketMap.remove(topic);                     redisPubSubMap.get(topic).unsubscribeAndClose(topic);                     redisPubSubMap.remove(topic);                 }             }             \/\/\u5220\u9664\u6a21\u7cca\u8ba2\u9605             Iterator iteratorP = pWebSocketMap.keySet().iterator();             while (iteratorP.hasNext()) {                 String topic = (String) iteratorP.next();                 ConcurrentHashMap\n               \n                 map = pWebSocketMap.get(topic);                 map.remove(this.sessionId);                 if (map.size() == 0) {\/\/\u5982\u679c\u8fd9\u4e2a\u9891\u9053\u6ca1\u6709\u7528\u6237\u8ba2\u9605\u4e86\uff0c\u5219\u53d6\u6d88\u8ba2\u9605\u8be5redis\u9891\u9053                     pWebSocketMap.remove(topic);                     redisPubSubMap.get(topic).punsubscribeAndClose(topic);                     redisPubSubMap.remove(topic);                 }             }             log.info(\"sessionId\uff1a\" + this.sessionId + \"\uff0c\u65ad\u5f00\u8fde\u63a5\uff1a\");             \/\/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(\"\u7528\u6237\u9519\u8bef,sessionId:\" + session.getId() + \",\u539f\u56e0:\" + error.getMessage());             error.printStackTrace();             log.info(\"\u5173\u95ed\u9519\u8bef\u7528\u6237\u5bf9\u5e94\u7684\u8fde\u63a5\");             \/\/\u5220\u9664\u8ba2\u9605             Iterator iterator = webSocketMap.keySet().iterator();             while (iterator.hasNext()) {                 String topic = (String) iterator.next();                 ConcurrentHashMap\n                \n                  map = webSocketMap.get(topic);                 map.remove(this.sessionId);                 if (map.size() == 0) {\/\/\u5982\u679c\u8fd9\u4e2a\u9891\u9053\u6ca1\u6709\u7528\u6237\u8ba2\u9605\u4e86\uff0c\u5219\u53d6\u6d88\u8ba2\u9605\u8be5redis\u9891\u9053                     webSocketMap.remove(topic);                     redisPubSubMap.get(topic).unsubscribeAndClose(topic);                     redisPubSubMap.remove(topic);                 }             }             \/\/\u5220\u9664\u6a21\u7cca\u8ba2\u9605             Iterator iteratorP = pWebSocketMap.keySet().iterator();             while (iteratorP.hasNext()) {                 String topic = (String) iteratorP.next();                 ConcurrentHashMap\n                 \n                   map = pWebSocketMap.get(topic);                 map.remove(this.sessionId);                 if (map.size() == 0) {\/\/\u5982\u679c\u8fd9\u4e2a\u9891\u9053\u6ca1\u6709\u7528\u6237\u8ba2\u9605\u4e86\uff0c\u5219\u53d6\u6d88\u8ba2\u9605\u8be5redis\u9891\u9053                     pWebSocketMap.remove(topic);                     redisPubSubMap.get(topic).punsubscribeAndClose(topic);                     redisPubSubMap.remove(topic);                 }             }             log.info(\"\u5b8c\u6210\u9519\u8bef\u7528\u6237\u5bf9\u5e94\u7684\u8fde\u63a5\u5173\u95ed\");             \/\/debug             log();             log.info(\"============================onError-end============================\");         }     }     \/**      * \u5b9e\u73b0\u670d\u52a1\u5668\u4e3b\u52a8\u63a8\u9001      *\/     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\n                  \n                    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(\">>>>>>>>>\");         Iterator iterator1 = webSocketMap.keySet().iterator();         while (iterator1.hasNext()) {             String topic = (String) iterator1.next();             log.info(\"topic\uff1a\" + topic);             Iterator iterator2 = webSocketMap.get(topic).keySet().iterator();             while (iterator2.hasNext()) {                 String session = (String) iterator2.next();                 log.info(\"\u8ba2\u9605\" + topic + \"\u7684sessionId\uff1a\" + session);             }         }         log.info(\">>>>>>>>>\");     } }\n                  \n                 \n                \n               \n              \n             \n            \n           \n          \n         \n        \n       \n      \n     \n    \n   <\/pre>\n
\u9879\u76ee\u5730\u5740<\/h2>\n
\u4e0a\u9762\u4ecb\u7ecd\u4e86\u6838\u5fc3\u4ee3\u7801\uff0c\u4e0b\u9762\u662f\u5b8c\u6574\u4ee3\u7801\u5730\u5740<\/p>\n<\/p>\n
Update20220415<\/h2>\n
\u53c2\u8003\u8bc4\u8bba\u533a\u8001\u54e5\u7684\u5efa\u8bae\uff0c\u5c06redis\u8ba2\u9605\u76d1\u542c\u7c7b\u91cc\u9762\u7684subscribe\u548cpsubscribe\u65b9\u6cd5\u8c03\u6574\u5982\u4e0b\uff1a<\/p>\n
    \/\/\u8ba2\u9605\n    @Override\n    public void subscribe(String... channels) {\n        boolean done = true;\n        while (done){\n            Jedis jedis = jedisPool.getResource();\n            try {\n                jedis.subscribe(this, channels);\n                done = false;\n            } catch (Exception e) {\n                log.error(e.getMessage());\n                if (jedis != null)\n                    jedis.close();\n                \/\/\u9047\u5230\u5f02\u5e38\u540e\u5173\u95ed\u8fde\u63a5\u91cd\u65b0\u8ba2\u9605\n                log.info(\"\u76d1\u542c\u9047\u5230\u5f02\u5e38\uff0c\u56db\u79d2\u540e\u91cd\u65b0\u8ba2\u9605\u9891\u9053\uff1a\");\n                Arrays.asList(channels).forEach(s -> {log.info(s);});\n                try {\n                    Thread.sleep(4000);\n                } catch (InterruptedException interruptedException) {\n                    interruptedException.printStackTrace();\n                }\n            }\n        }\n    }\n    \/\/\u6a21\u7cca\u8ba2\u9605\n    @Override\n    public void psubscribe(String... channels) {\n        boolean done = true;\n        while (done){\n            Jedis jedis = jedisPool.getResource();\n            try {\n                jedis.psubscribe(this, channels);\n                done = false;\n            } catch (Exception e) {\n                log.error(e.getMessage());\n                if (jedis != null)\n                    jedis.close();\n                \/\/\u9047\u5230\u5f02\u5e38\u540e\u5173\u95ed\u8fde\u63a5\u91cd\u65b0\u8ba2\u9605\n                log.info(\"\u76d1\u542c\u9047\u5230\u5f02\u5e38\uff0c\u56db\u79d2\u540e\u91cd\u65b0\u8ba2\u9605\u9891\u9053\uff1a\");\n                Arrays.asList(channels).forEach(s -> {log.info(s);});\n                try {\n                    Thread.sleep(4000);\n                } catch (InterruptedException interruptedException) {\n                    interruptedException.printStackTrace();\n                }\n            }\n        }\n    }<\/pre>\n
\u5230\u8fd9\u91cc\uff0c\u6211\u4eec\u4e5f\u5c31\u8bb2\u5b8c\u4e86\u300awebsocket+redis\u52a8\u6001\u8ba2\u9605\u548c\u52a8\u6001\u53d6\u6d88\u8ba2\u9605\u7684\u5b9e\u73b0\u793a\u4f8b\u300b\u7684\u5185\u5bb9\u4e86\u3002\u4e2a\u4eba\u8ba4\u4e3a\uff0c\u57fa\u7840\u77e5\u8bc6\u7684\u5b66\u4e60\u548c\u5de9\u56fa\uff0c\u662f\u4e3a\u4e86\u66f4\u597d\u7684\u5c06\u5176\u8fd0\u7528\u5230\u9879\u76ee\u4e2d\uff0c\u6b22\u8fce\u5173\u6ce8golang\u5b66\u4e60\u7f51\u516c\u4f17\u53f7\uff0c\u5e26\u4f60\u4e86\u89e3\u66f4\u591a\u5173\u4e8eredis\u7684\u77e5\u8bc6\u70b9\uff01<\/p>\n
    \u7248\u672c\u58f0\u660e \u672c\u6587\u8f6c\u8f7d\u4e8e\uff1a\u811a\u672c\u4e4b\u5bb6 \u5982\u6709\u4fb5\u72af\uff0c\u8bf7\u8054\u7cfb \u5220\u9664  <\/p>\n
\n
\n <\/dt>\n
\n   \u6d45\u8c08Redis\u7f13\u5b58\u96ea\u5d29\u89e3\u51b3\u65b9\u6848\n <\/dd>\n<\/dl>\n
\n
\n <\/dt>\n
\n   Redis\u4e2dRedisson\u5e03\u9686\u8fc7\u6ee4\u5668\u7684\u5b66\u4e60\n <\/dd>\n<\/dl>\n","protected":false},"excerpt":{"rendered":"
websocket+redis\u52a8\u6001\u8ba2\u9605\u548c\u52a8\u6001\u53d6\u6d88\u8ba2\u9605\u7684\u5b9e\u73b0\u793a\u4f8b 0\u6d4f\u89c8 \u6536\u85cf \u600e\u4e48\u5165\u95e8\u6570\u636e\u5e93\u7f16\u7a0b\uff1f\u9700\u8981\u5b66\u4e60\u54ea\u4e9b\u77e5\u8bc6\u70b9\uff1f\u8fd9\u662f\u65b0\u624b\u4eec\u521a\u63a5\u89e6\u7f16\u7a0b\u65f6\u5e38\u89c1\u7684\u95ee\u9898\uff1b\u4e0b\u9762golang\u5b66\u4e60\u7f51\u5c31\u6765\u7ed9\u5927\u5bb6\u6574\u7406\u5206\u4eab\u4e00\u4e9b\u77e5\u8bc6\u70b9\uff0c\u5e0c\u671b\u80fd\u591f\u7ed9\u521d\u5b66\u8005\u4e00\u4e9b\u5e2e\u52a9\u3002\u672c\u7bc7\u6587\u7ae0\u5c31\u6765\u4ecb\u7ecd\u300awebsocket+redis\u52a8\u6001\u8ba2\u9605\u548c\u52a8\u6001\u53d6\u6d88\u8ba2\u9605\u7684\u5b9e\u73b0\u793a\u4f8b\u300b\uff0c\u6d89\u53ca\u5230websocketredis\u3001\u52a8\u6001\u8ba2\u9605\uff0c\u6709\u9700\u8981\u7684\u53ef\u4ee5\u6536\u85cf\u4e00\u4e0b \u539f\u7406 websocket\u7684\u8ba2\u9605\u5c31\u662f\u5728\u524d\u540e\u7aef\u5efa\u7acbws\u8fde\u63a5\u4e4b\u540e\uff0c\u524d\u7aef\u901a\u8fc7\u53d1\u9001\u4e00\u5b9a\u683c\u5f0f\u7684\u6d88\u606f\uff0c\u540e\u7aef\u89e3\u6790\u51fa\u6765\u53bb\u8ba2\u9605\u6216\u8005\u53d6\u6d88\u8ba2\u9605redis\u9891\u9053\u3002 \u8ba2\u9605\u9891\u9053\u6d88\u606f\u683c\u5f0f: { “cmd”:”subscribe”, “topic”:[ “topic_name” ] } \u6a21\u7cca\u8ba2\u9605\u683c\u5f0f { “cmd”:”psubscribe”, “topic”:[ “topic_name” ] } \u53d6\u6d88\u8ba2\u9605\u683c\u5f0f { “cmd”:”unsubscribe”, “topic”:[ “topic_name” ] } \u4e24\u4e2a\u6838\u5fc3\u7c7b\uff0c\u4e00\u4e2a\u662fredis\u7684\u8ba2\u9605\u76d1\u542c\u7c7b\uff0c\u4e00\u4e2a\u662fwebsocket\u7684\u53d1\u5e03\u8ba2\u9605\u7c7b\u3002 redis\u8ba2\u9605\u76d1\u542c\u7c7b 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 […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[8],"tags":[],"class_list":["post-44163","post","type-post","status-publish","format-standard","hentry","category-os"],"_links":{"self":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/posts\/44163","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/comments?post=44163"}],"version-history":[{"count":0,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/posts\/44163\/revisions"}],"wp:attachment":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/media?parent=44163"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/categories?post=44163"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/tags?post=44163"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}