SpringCloud整合WebSocket实现用户监控

@
目录

• 前言
  
  
  
  • 一、建项目
    
    
    
    • 
      
      
      
      • 1. 在父项目ams-cloud下建立maven子项目ams-websocket
        
      • 2.pom文件添加常用依赖，另外添加redis依赖等，我这里直接引用common模块
        
      • 3.添加bootstrap.yml文件
        
      • 4.创建application
        
      
      
    
    
  • 二、添加config类、拦截器类、处理器类等
    
    
    
    • 
      
      
      
      • 1.添加config类
        
      • 2.添加拦截器类
        
      • 3.添加处理器类
        
      
      
    
    
  • 三、添加controller
    
    
    
    • 
      
      
      
      • 1.controller如下：
        
        
        
        • 2.运行结果
          
        
        
      
      
    
    
  
  

---

1. [/code][size=5]一、建项目[/size]
3. [size=3]1. 在父项目ams-cloud下建立maven子项目ams-websocket[/size]
5. [img]https://img-blog.csdnimg.cn/e57204e3c5f64ce386ce31e0ffc60544.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAZm9seWg=,size_20,color_FFFFFF,t_70,g_se,x_16[/img]
6. [size=3]2.pom文件添加常用依赖，另外添加redis依赖等，我这里直接引用common模块[/size]
8. [code]    <dependencies>
9.         <dependency>
10.             <groupId>com.alibaba.cloud</groupId>
11.             <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
12.         </dependency>
13.         <dependency>
14.             <groupId>com.alibaba.cloud</groupId>
15.             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
16.         </dependency>
17.         <dependency>
18.             <groupId>com.alibaba.cloud</groupId>
19.             <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
20.         </dependency>
22.         <dependency>
23.             <groupId>org.springframework.boot</groupId>
24.             <artifactId>spring-boot-starter-web</artifactId>
25.         </dependency>
26.         <dependency>
27.             <groupId>org.springframework.boot</groupId>
28.             <artifactId>spring-boot-starter-websocket</artifactId>
29.         </dependency>
31.         <dependency>
32.             <groupId>com.ams</groupId>
33.             <artifactId>ams-common</artifactId>
34.             <version>${ams.version}</version>
35.         </dependency>
37.     </dependencies>

复制代码

3.添加bootstrap.yml文件

1. server:
2.   port: 21000
4. spring:
5.   application:
6.     name: ams-websocket
7.   cloud:
8.     nacos:
9.       # 注册中心
10.       discovery:
11.         server-addr: http://192.168.132.129:8848
12.       # 配置中心
13.       config:
14.         server-addr: ${spring.cloud.nacos.discovery.server-addr}
15.         file-extension: yaml
16.         shared-configs[0]:
17.           data-id: ams-common.yaml
18.           refresh: true
19.          
20. logging:
21.   level:
22.     spring.: DEBUG

复制代码

4.创建application

1. @EnableDiscoveryClient
2. @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
3. @ComponentScan("com.ams")
4. public class WebsocketApp {
5.     public static void main(String[] args) {
6.         SpringApplication.run(WebsocketApp.class,args);
7.     }
8. }

复制代码

二、添加config类、拦截器类、处理器类等

1.添加config类

1. @Configuration
2. @EnableWebSocket
3. public class WebsocketConfig implements WebSocketConfigurer {
5.     @Autowired
6.     private RedisUtils redisUtils;
8.     @Override
9.     public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
10.         // 注册websocket组件 添加处理器和拦截器
11.         // websocket是websocket服务器的请求路径可以自己定义
12.         registry.addHandler(new WebsocketHandler(redisUtils),"/websocket")
13.                 //指定自定义拦截器
14.                 .addInterceptors(new WebsocketInterceptor(redisUtils))
15.                 //允许跨域
16.                 .setAllowedOrigins("*");
18.         //在某些低版本的浏览器中不支持websocket可以用sock-js替代
19.         registry.addHandler(new WebsocketHandler(redisUtils),"/sock-js")
20.                 // 指定自定义拦截器
21.                 .addInterceptors(new WebsocketInterceptor(redisUtils))
22.                 // 允许跨域
23.                 .setAllowedOrigins("*")
24.                 // 开启sockJs支持
25.                 .withSockJS();
27.     }
29. }

复制代码

2.添加拦截器类

1. @Slf4j
2. public class WebsocketInterceptor extends HttpSessionHandshakeInterceptor  {
4.     private final RedisUtils redisUtils;
5.     public WebsocketInterceptor(RedisUtils redisUtils){
6.         this.redisUtils=redisUtils;
7.     }
8.    
9.         /** 管理握手过程，存入用户信息 */
10.     @Override
11.     public boolean beforeHandshake(ServerHttpRequest request,
12.                                    ServerHttpResponse response,
13.                                    WebSocketHandler handler,
14.                                    Map<String, Object> map) throws Exception {
15.         String header_key = "Sec-WebSocket-Protocol";
16.         HttpHeaders headers = request.getHeaders();
17.         String token = headers.getFirst(header_key);
18.         log.info("token = [{}]", token);
19.         HttpHeaders responseHeaders = response.getHeaders();
20.         UserAuthDTO userAuthDTO = getUserInfo(token);
21.         if(userAuthDTO == null){
22.             log.error("socket连接失败 ---> token过期 ---> [{}]", token);
23.             response.setStatusCode(HttpStatus.NETWORK_AUTHENTICATION_REQUIRED);
24.             return false;
25.         }
26.         map.put("userAuthDTO",userAuthDTO);
27.         responseHeaders.add(header_key,token);
28.         return super.beforeHandshake(request, response, handler, map);
30.     }
32.     /** 通过redis读取用户信息 **/
33.     public UserAuthDTO getUserInfo(String token) {
34.         String key = token;
35.         if(StrUtil.isEmpty(key)){
36.             return null;
37.         }
38.         UserAuthDTO userAuthDTO = (UserAuthDTO) redisUtils.get(key);
39.         if(userAuthDTO==null){
40.             log.error("redis用户信息空 ---> 登录过期/token不正确");
41.             return null;
42.         }
43.         return userAuthDTO;
44.     }
45. }

复制代码

3.添加处理器类

1. @Slf4j
2. public class WebsocketHandler extends AbstractWebSocketHandler  {
4.     //定义全局变量用于保存所有用户的会话
5.     /** 系统管理员 **/
6.     public static final Map<String, WebSocketSession> SYSUSER_SOCKETS = new HashMap<>();
8.     private static RedisUtils redisUtils;
9.     public WebsocketHandler(RedisUtils redisUtils){
10.         WebsocketHandler.redisUtils = redisUtils;
11.     }
13.     /**
14.      * webSocket连接创建后调用
15.      */
16.     @Override
17.     public void afterConnectionEstablished(WebSocketSession session) throws Exception {
18.         Map<String, Object> attrMap = session.getAttributes();
19.         UserAuthDTO userAuthDTO = (UserAuthDTO) attrMap.get("userAuthDTO");
20.         SYSUSER_SOCKETS.put(userAuthDTO.getUserId().toString(),session);
21.         log.info("管理员[{}]连接成功，当前在线人数[{}]", userAuthDTO.getUsername(), SYSUSER_SOCKETS.size());
22.     }
24.     /**
25.      * 接收到消息会调用
26.      */
27.     @Override
28.     protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
29.         try {
30.                        log.info("收到客户端消息[{}]", message);
31.             if(session.isOpen()){
32.                 session.sendMessage(message);
33.             }
34.         } catch (Exception e) {
35.             e.printStackTrace();
36.         }
37.     }
39.     /**
40.      * 连接关闭会调用
41.      */
42.     @Override
43.     public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
44.         log.info("客户端关闭连接....");
45.         Map<String, Object> attrMap = session.getAttributes();
46.         // 删除缓存
47.         deleteSocket(attrMap);
48.         // 关闭连接
49.         session.close();
50.         log.info("已关闭socket连接");
51.     }
53.     /**
54.      * 删除socket缓存
55.      */
56.     public static void deleteSocket(Map<String, Object> attr) {
57.         UserAuthDTO userAuthDTO = (UserAuthDTO) attr.get("userAuthDTO");
58.         if (!SYSUSER_SOCKETS.isEmpty()) {
59.             SYSUSER_SOCKETS.remove(userAuthDTO.getUserId().toString());
60.             log.info("管理员[{}]关闭连接了，当前在线人数[{}]", userAuthDTO.getUsername(), SYSUSER_SOCKETS.size());
61.         }
62.     }
64.     /**
65.      * 连接出错会调用
66.      */
67.     @Override
68.     public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
69.         log.error("socket连接出错...");
70.         exception.printStackTrace();
71.         Map<String, Object> attrMap = session.getAttributes();
72.         //删除缓存
73.         deleteSocket(attrMap);
74.         // 关闭连接
75.         session.close();
76.         log.error("已关闭socket连接");
77.     }
78. }

复制代码

三、添加controller

1.controller如下：

1. @RestController
2. @RequestMapping("/websocket")
3. @Slf4j
4. @RequiredArgsConstructor
5. public class WebsocketController {
7.     /** 获取在线管理用户 */
8.     @GetMapping(value = "getLoginSysUser")
9.     public R<List<UserAuthDTO>> getLoginSysUser(){
10.         Map<String, WebSocketSession> userMap = WebsocketHandler.SYSUSER_SOCKETS;
11.         List<UserAuthDTO> list = new ArrayList<>();
12.         userMap.forEach((k,v)->{
13.             Map<String, Object> attrMap = v.getAttributes();
14.             UserAuthDTO userAuthDTO = (UserAuthDTO) attrMap.get("userAuthDTO");
15.             list.add(userAuthDTO);
16.         });
17.         return R.ok(list);
18.     }
19.    
20. }

复制代码

2.运行结果

1. 2022-03-04 16:58:56.822  INFO 15272 --- [nio-21000-exec-2] c.c.f.m.w.config.WebsocketInterceptor    : token = [83868c09-a744-4a7a-a16e-24dc3d7e8cdc]
2. 2022-03-04 16:58:59.335  INFO 15272 --- [nio-21000-exec-2] c.c.f.m.w.handler.WebsocketHandler       : 管理员[admin]连接成功，当前在线人数[1]

复制代码