Merge remote-tracking branch 'origin/dev_2.0.0' into dev_2.0.0

virtual-patient-common/src/main/java/com/supervision/constant/UserTokenConstant.java

@@ -7,4 +7,6 @@ public interface UserTokenConstant {
 
     Integer KICK_CODE = 10000;
 
+    Integer KEEPALIVE_CODE = 10001;
+
 }

virtual-patient-web/src/main/java/com/supervision/VirtualPatientApplication.java

@@ -5,11 +5,13 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.ComponentScans;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.web.socket.config.annotation.EnableWebSocket;
 
 @SpringBootApplication
 @MapperScan(basePackages = {"com.supervision.**.mapper"})
 @EnableWebSocket
+@EnableScheduling
 public class VirtualPatientApplication {
 
     public static void main(String[] args) {

virtual-patient-web/src/main/java/com/supervision/config/WebSocketConfig.java

@@ -2,6 +2,8 @@ package com.supervision.config;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.web.socket.server.standard.ServerEndpointExporter;
 
 @Configuration
@@ -14,4 +16,13 @@ public class WebSocketConfig {
     public ServerEndpointExporter serverEndpointExporter(){
         return new ServerEndpointExporter();
     }
+
+    @Bean
+    public TaskScheduler taskScheduler(){
+        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
+        threadPoolTaskScheduler.setThreadNamePrefix("SockJS-");
+        threadPoolTaskScheduler.setPoolSize(Runtime.getRuntime().availableProcessors());
+        threadPoolTaskScheduler.setRemoveOnCancelPolicy(true);
+        return threadPoolTaskScheduler;
+    }
 }

virtual-patient-web/src/main/java/com/supervision/controller/UserController.java

@@ -27,6 +27,8 @@ import org.springframework.web.bind.annotation.*;
 
 import java.time.LocalDateTime;
 import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 @Api(tags = "用户管理")
@@ -43,6 +45,7 @@ public class UserController {
 
     private final UserResourceCheck userResourceCheck;
 
+
     @Value("${ws.nginx-ip:}")
     private String wsIp;
     @Value("${ws.nginx-port:}")
@@ -117,5 +120,11 @@ public class UserController {
         throw new BusinessException("未获取到ws的nginx地址,请确认配置文件是否配置");
     }
 
+    @ApiOperation("获取当前在线的用户")
+    @GetMapping("queryCurrentOnlineUser")
+    public Map<Object, Object> queryCurrentOnlineUser(){
+        return redisTemplate.opsForHash().entries(UserTokenConstant.USER_WEBSOCKET_CACHE);
+    }
+
 
 }

virtual-patient-web/src/main/java/com/supervision/task/WebsocketKeepaliveTask.java

@@ -0,0 +1,48 @@
+package com.supervision.task;
+
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.json.JSONUtil;
+import com.supervision.constant.UserTokenConstant;
+import com.supervision.controller.WebSocketServer;
+import com.supervision.usermanage.UserResourceCheck;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.Session;
+import java.io.IOException;
+import java.util.Map;
+
+@Component
+@Slf4j
+@RequiredArgsConstructor
+public class WebsocketKeepaliveTask {
+
+    private final RedisTemplate<String, String> redisTemplate;
+
+    @Scheduled(fixedDelay = 5 * 1000L)
+    public void keepalive() {
+        log.info("websocket保活接口开始,每5秒钟发送一次消息");
+        for (Map.Entry<String, Session> entries : WebSocketServer.SESSION_POOL.entrySet()) {
+            String userId = entries.getKey();
+            Session session = entries.getValue();
+            if (ObjectUtil.isNotEmpty(session)) {
+                try {
+
+                    session.getBasicRemote().sendText(JSONUtil.toJsonStr(new UserResourceCheck.Keepalive()));
+                    log.info("用户:{}的websocket保活成功", userId);
+                } catch (IOException e) {
+                    log.error("用户:{}的websocket连接异常", userId, e);
+                    // 连接异常的用户,移除
+                    WebSocketServer.SESSION_POOL.remove(userId);
+                    // 移除redis中该用户的缓存
+                    redisTemplate.opsForHash().delete(UserTokenConstant.USER_WEBSOCKET_CACHE, userId);
+                }
+            }
+        }
+        log.info("websocket保活接口结束");
+    }
+
+}

virtual-patient-web/src/main/java/com/supervision/usermanage/KickUserListener.java

@@ -24,10 +24,6 @@ public class KickUserListener implements MessageListener {
         String messageString = message.toString();
         UserWebSocketDTO user = JSONUtil.toBean(messageString, UserWebSocketDTO.class);
         log.info("Redis的Channel:{}收到踢用户{}下线消息", UserTokenConstant.KICK_CHANNEL, user.getUserId());
-        try {
-            userResourceCheck.kickUser(user.getUserId(), user.getIgnoreSessionId());
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
+        userResourceCheck.kickUser(user.getUserId(), user.getIgnoreSessionId());
     }
 }

virtual-patient-web/src/main/java/com/supervision/usermanage/UserResourceCheck.java

@@ -13,10 +13,14 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 
 import javax.websocket.*;
+
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
 
 @Slf4j
 @Component
@@ -37,7 +41,7 @@ public class UserResourceCheck {
     }
 
     @Lock4j(name = "achieveDiagnoseResourceAndOpenConnection")
-    public void achieveDiagnoseResourceAndOpenConnection(String uid, Session session){
+    public void achieveDiagnoseResourceAndOpenConnection(String uid, Session session) {
         // 如果小于数字人最大连接数,则可以连接
         if (!achieveDiagnoseResource()) {
             throw new BusinessException("暂时没有资源,建立连接失败");
@@ -46,21 +50,33 @@ public class UserResourceCheck {
         // 链接之前先把之前的用户踢下线(ignoreSessionId防止把当前用户踢下线)
         // 注意,这里如果用户没有进到问诊页面,只是在问诊大厅时,是不会被踢掉的.(因为这时没有建立websocket连接,还没与放到SESSION_POOL里面去)
         redisTemplate.convertAndSend(UserTokenConstant.KICK_CHANNEL, JSONUtil.toJsonStr(new UserWebSocketDTO(uid, session.getId())));
-        log.info("尝试踢该用户{}的其他session下线,忽略sessionId:{}",uid,session.getId());
+        log.info("尝试踢该用户{}的其他session下线,忽略sessionId:{}", uid, session.getId());
         redisTemplate.opsForHash().put(UserTokenConstant.USER_WEBSOCKET_CACHE, uid, session.getId());
     }
 
+    /**
+     * websocket保活接口,每5秒钟发送一次消息
+     */
+
     // 实现一个方法用于踢下线用户,走的是Redis的消息队列
-    public void kickUser(String userId, String ignoreSessionId) throws IOException {
+    public void kickUser(String userId, String ignoreSessionId) {
         log.info("尝试踢用户:{}下线", userId);
         Session session = WebSocketServer.SESSION_POOL.get(userId);
         // 只有不是忽略剔除的sessionId才可以踢下线
         if (ObjectUtil.isNotEmpty(session) && !StrUtil.equals(ignoreSessionId, session.getId())) {
-            session.getBasicRemote().sendText(JSONUtil.toJsonStr(new Kick()));
-            session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "用户被踢下线"));
-            WebSocketServer.SESSION_POOL.remove(userId);
-            log.info("踢用户:{},sessionId:{} 下线成功", userId, session.getId());
-            return;
+            try {
+                session.getBasicRemote().sendText(JSONUtil.toJsonStr(new Kick()));
+                session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "用户被踢下线"));
+                WebSocketServer.SESSION_POOL.remove(userId);
+                log.info("踢用户:{},sessionId:{} 下线成功", userId, session.getId());
+                return;
+            } catch (IOException e) {
+                log.error("用户:{}的websocket连接异常", e.getMessage());
+                // 连接异常的用户,移除
+                WebSocketServer.SESSION_POOL.remove(userId);
+                // 移除redis中该用户的缓存
+                redisTemplate.opsForHash().delete(UserTokenConstant.USER_WEBSOCKET_CACHE, userId);
+            }
         }
         log.info("踢用户:{}下线,未找到用户,踢下线失败", userId);
     }
@@ -73,4 +89,12 @@ public class UserResourceCheck {
 
     }
 
+    @Data
+    public static class Keepalive {
+        private final Integer code = UserTokenConstant.KEEPALIVE_CODE;
+
+        private final String message = "keepalive";
+
+    }
+
 }