websocket优化代码

jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/message/websocket/SocketHandler.java

@@ -9,11 +9,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 /**
- * 监听消息(采用redis发布订阅方式发送消息)
+ * 监听消息(通过redis发布订阅，推送消息)
+ * 此方案：解决集群部署的问题，多实例节点（也就是发送消息端先发送消息到redis中，每个服务节点收到redis消息，再触发具体的ws推送）
  * @author: jeecg-boot
  */
 @Slf4j
-@Component
+@Component(WebSocket.REDIS_TOPIC_NAME)
 public class SocketHandler implements JeecgRedisListener {
 
     @Autowired
@@ -21,15 +22,17 @@ public class SocketHandler implements JeecgRedisListener {
 
     @Override
     public void onMessage(BaseMap map) {
-        log.info("【SocketHandler消息】Redis Listerer:" + map.toString());
+        log.info("【Redis发布订阅模式】redis Listener: {}，参数：{}",WebSocket.REDIS_TOPIC_NAME, map.toString());
 
         String userId = map.get("userId");
         String message = map.get("message");
         if (ObjectUtil.isNotEmpty(userId)) {
+            //pc端消息推送具体人
             webSocket.pushMessage(userId, message);
-            //app端消息推送
+            //app端消息推送具体人
             webSocket.pushMessage(userId+CommonSendStatus.APP_SESSION_SUFFIX, message);
         } else {
+            //推送全部
             webSocket.pushMessage(message);
         }
 

jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/message/websocket/WebSocket.java

@@ -1,25 +1,17 @@
 package org.jeecg.modules.message.websocket;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArraySet;
-
 import javax.annotation.Resource;
-import javax.websocket.OnClose;
+import javax.websocket.*;
-import javax.websocket.OnMessage;
-import javax.websocket.OnOpen;
-import javax.websocket.Session;
 import javax.websocket.server.PathParam;
 import javax.websocket.server.ServerEndpoint;
 
+import com.alibaba.fastjson.JSONObject;
 import org.jeecg.common.base.BaseMap;
 import org.jeecg.common.constant.WebsocketConst;
 import org.jeecg.common.modules.redis.client.JeecgRedisClient;
 import org.springframework.stereotype.Component;
-
-import com.alibaba.fastjson.JSONObject;
-
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -31,109 +23,126 @@ import lombok.extern.slf4j.Slf4j;
 @Slf4j
 @ServerEndpoint("/websocket/{userId}")
 public class WebSocket {
-
+    
-    private Session session;
+    /**线程安全Map*/
+    private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
 
     /**
-     * 用户ID
+     * Redis触发监听名字
      */
-    private String userId;
+    public static final String REDIS_TOPIC_NAME = "socketHandler";
-
-    private static final String REDIS_TOPIC_NAME = "socketHandler";
-
     @Resource
     private JeecgRedisClient jeecgRedisClient;
 
-    /**
-     * 缓存 webSocket连接到单机服务class中（整体方案支持集群）
-     */
-    private static CopyOnWriteArraySet<WebSocket> webSockets = new CopyOnWriteArraySet<>();
-    /**
-     * 线程安全Map
-     */
-    private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
 
+    //==========【websocket接受、推送消息等方法 —— 具体服务节点推送ws消息】========================================================================================
     @OnOpen
     public void onOpen(Session session, @PathParam(value = "userId") String userId) {
         try {
-            //TODO 通过header中获取token，进行check
-            this.session = session;
-            this.userId = userId;
-            webSockets.add(this);
             sessionPool.put(userId, session);
-            log.info("【websocket消息】有新的连接，总数为:" + webSockets.size());
+            log.info("【系统 WebSocket】有新的连接，总数为:" + sessionPool.size());
         } catch (Exception e) {
         }
     }
 
     @OnClose
-    public void onClose() {
+    public void onClose(@PathParam("userId") String userId) {
         try {
-            webSockets.remove(this);
+            sessionPool.remove(userId);
-            sessionPool.remove(this.userId);
+            log.info("【系统 WebSocket】连接断开，总数为:" + sessionPool.size());
-            log.info("【websocket消息】连接断开，总数为:" + webSockets.size());
-        } catch (Exception e) {
-        }
-    }
-
-
-    /**
-     * 服务端推送消息
-     *
-     * @param userId
-     * @param message
-     */
-    public void pushMessage(String userId, String message) {
-        Session session = sessionPool.get(userId);
-        if (session != null && session.isOpen()) {
-            try {
-                //update-begin-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
-                synchronized (session){
-                    log.info("【websocket消息】 单点消息:" + message);
-                    session.getBasicRemote().sendText(message);
-                }
-                //update-end-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-        }
-    }
-
-    /**
-     * 服务器端推送消息
-     */
-    public void pushMessage(String message) {
-        try {
-            webSockets.forEach(ws -> ws.session.getAsyncRemote().sendText(message));
         } catch (Exception e) {
             e.printStackTrace();
         }
     }
 
+    /**
+     * ws推送消息
+     *
+     * @param userId
+     * @param message
+     */
+    public void pushMessage(String userId, String message) {
+        for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
+            //userId key值= {用户id + "_"+ 登录token的md5串}
+            //TODO vue2未改key新规则，暂时不影响逻辑
+            if (item.getKey().contains(userId)) {
+                Session session = item.getValue();
+                try {
+                    //update-begin-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
+                    synchronized (session){
+                        log.info("【系统 WebSocket】推送单人消息:" + message);
+                        session.getBasicRemote().sendText(message);
+                    }
+                    //update-end-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
+                } catch (Exception e) {
+                    log.error(e.getMessage(),e);
+                }
+            }
+        }
+    }
 
+    /**
+     * ws遍历群发消息
+     */
+    public void pushMessage(String message) {
+        try {
+            for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
+                try {
+                    item.getValue().getAsyncRemote().sendText(message);
+                } catch (Exception e) {
+                    log.error(e.getMessage(), e);
+                }
+            }
+            log.info("【系统 WebSocket】群发消息:" + message);
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
+    }
+
+
+    /**
+     * ws接受客户端消息
+     */
     @OnMessage
-    public void onMessage(String message) {
+    public void onMessage(String message, @PathParam(value = "userId") String userId) {
-        //todo 现在有个定时任务刷，应该去掉
+        if(!"ping".equals(message) && !WebsocketConst.CMD_CHECK.equals(message)){
-        log.debug("【websocket消息】收到客户端消息:" + message);
+            log.info("【系统 WebSocket】收到客户端消息:" + message);
+        }else{
+            log.debug("【系统 WebSocket】收到客户端消息:" + message);
+        }
+        
+        //------------------------------------------------------------------------------
         JSONObject obj = new JSONObject();
         //业务类型
         obj.put(WebsocketConst.MSG_CMD, WebsocketConst.CMD_CHECK);
         //消息内容
         obj.put(WebsocketConst.MSG_TXT, "心跳响应");
-        //update-begin-author:taoyan date:20220308 for: 消息通知长连接启动心跳机制，后端代码小bug #3473
+        this.pushMessage(userId, obj.toJSONString());
-        for (WebSocket webSocket : webSockets) {
+        //------------------------------------------------------------------------------
-            webSocket.pushMessage(obj.toJSONString());
-        }
-        //update-end-author:taoyan date:20220308 for: 消息通知长连接启动心跳机制，后端代码小bug #3473
     }
 
+    /**
+     * 配置错误信息处理
+     *
+     * @param session
+     * @param t
+     */
+    @OnError
+    public void onError(Session session, Throwable t) {
+        log.warn("【系统 WebSocket】消息出现错误");
+        //t.printStackTrace();
+    }
+    //==========【系统 WebSocket接受、推送消息等方法 —— 具体服务节点推送ws消息】========================================================================================
+    
+
+    //==========【采用redis发布订阅模式——推送消息】========================================================================================
     /**
      * 后台发送消息到redis
      *
      * @param message
      */
     public void sendMessage(String message) {
-        log.info("【websocket消息】广播消息:" + message);
+        //log.info("【系统 WebSocket】广播消息:" + message);
         BaseMap baseMap = new BaseMap();
         baseMap.put("userId", "");
         baseMap.put("message", message);
@@ -141,7 +150,7 @@ public class WebSocket {
     }
 
     /**
-     * 此为单点消息
+     * 此为单点消息 redis
      *
      * @param userId
      * @param message
@@ -154,7 +163,7 @@ public class WebSocket {
     }
 
     /**
-     * 此为单点消息(多人)
+     * 此为单点消息(多人) redis
      *
      * @param userIds
      * @param message
@@ -164,5 +173,6 @@ public class WebSocket {
             sendMessage(userId, message);
         }
     }
-
+    //=======【采用redis发布订阅模式——推送消息】==========================================================================================
+    
 }