接口 V2 美满：分布式环境下的 WebSocket 实现与 Token 校验

🎯 本文档具体先容了怎样利用WebSocket协议优化客户端与服务端之间的通讯，特殊是在处理惩罚异步订单创建关照的场景中。通过引入WebSocket取代传统的HTTP哀求-相应模式，实现了服务器主动向客户端推送数据的功能，极大地进步了及时性和服从。文中起首概述了WebSocket的上风，随后深入探究了其在分布式体系中的具体实现，包罗依靠管理、网关设置、WebSocket服务类的操持以及消息队列的利用等关键环节。特殊地，针对分布式架构下WebSocket毗连状态同步标题，提出了一种基于消息队列广播机制的办理方案，确保了体系的可扩展性和稳固性。同时，还夸大了心跳检测机制的告急性，以维护毗连的有效性。
🏠️ HelloDam/场快订（场馆预定 SaaS 平台）
  

前言

在时间段预定接口 V2 中，用户预定之后，会发送一个消息，让消息队列异步创建订单。此时客户端是无法知道服务端什么时间完成订单创建的，因此须要服务端告知客户端。但是以往都是客户端给服务端发 http 哀求，但是服务端怎样主动告知客户端呢？
这个时间就须要请出我们本日的主角 WebSocket 了
WebSocket 先容

WebSocket是一种在单个TCP毗连上举行全双工通讯的协议。它使得客户端和服务器之间的数据互换变得更加简单，允许服务器直接向客户端推送数据而不必由客户端发起哀求。这种特性让及时性要求较高的应用，如即时通讯工具、在线游戏以及及时买卖业务体系等，可以或许更加高效地举行数据交互。通过WebSocket，开发者可以构建相应更快、性能更高的网络应用，同时镌汰不须要的网络开销和延长。相比传统的HTTP哀求-相应模式，WebSocket提供了更低的延长和更高的服从，特殊是在须要频仍更新数据的应用场景中表现出色。
因此利用了 WebSocket ，一旦客户端和服务端创建了毗连，当订单创建乐成之后，服务端直接别订单数据推送给客户端即可。
流程图

user1、user2 和 user3 分别发起 WebSocket 毗连，起首颠末网关，毗连哀求被分发到差异的服务中。WebSocket 服务吸收到毗连哀求之后，对其举行登录校验，假如校验乐成，将其 Session 信息存储在服务器的内存中，假如校验失败，直接关闭 Session 。此中 user1、user2 的Session信息被存储在 WebSocket 服务1 中，user3 的Session信息被存储在 WebSocket 服务2 中。
当用户预定时间段，天生订单之后，场馆服务向消息队列中发生订单数据。接着消息队列将订单数据广播到 WebSocket 服务1 和 WebSocket 服务2中。WebSocket 服务2 发现本身的内存中存有 user3 的Session，因此将订单数据通过该 Session 发送给 user3 。
临时无法在飞书文档外展示此内容
具体实现

为相识耦 WebSocket 和其他服务，单独创建一个 WebSocket 服务。

依靠

1. <dependencies>
2.     <dependency>
3.         <groupId>com.vrs</groupId>
4.         <artifactId>vrs-web</artifactId>
5.         <version>1.0-SNAPSHOT</version>
6.     </dependency>
7.     <dependency>
8.         <groupId>org.dam</groupId>
9.         <artifactId>vrs-rocketmq</artifactId>
10.         <version>1.0-SNAPSHOT</version>
11.     </dependency>
12.     <dependency>
13.         <groupId>com.vrs</groupId>
14.         <artifactId>vrs-common</artifactId>
15.     </dependency>
16.     <dependency>
17.         <groupId>com.vrs</groupId>
18.         <artifactId>vrs-idempotent</artifactId>
19.         <version>1.0-SNAPSHOT</version>
20.     </dependency>
21.     <dependency>
22.         <groupId>com.alibaba.cloud</groupId>
23.         <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
24.     </dependency>
25.     <!-- websocket -->
26.     <dependency>
27.         <groupId>org.springframework.boot</groupId>
28.         <artifactId>spring-boot-starter-websocket</artifactId>
29.     </dependency>
30. </dependencies>

复制代码

网关设置

当访问 /websocket/** 路径时，将哀求转化到 WebSocket 服务，留意，转发的时间添加了前缀ws:

1. - id: vrs-websocket
2.   uri: lb:ws://vrs-websocket
3.   predicates:
4.     - Path=/websocket/**
5.   filters:
6.     - name: TokenValidate
7.       args:
8.         whitePathList:
9.           - /websocket/**

复制代码

【去除默认过滤器】
假如像如许全局设置了默认过滤器，DedupeResponseHeader过滤器的作用是对指定的相应头（在这个例子中为Vary、Access-Control-Allow-Origin和Access-Control-Allow-Credentials）举行去重。当有多个雷同名称的相应头时，它会按照给定的计谋保存此中的一个。这里的计谋是RETAIN_FIRST，意味着它将保存这些头部中第一次出现的谁人，而删除后续出现的重复头部。

1. spring:
2.   cloud:
3.     gateway:
4.       default-filters:
5.         - DedupeResponseHeader=Vary Access-Control-Allow-Origin Access-Control-Allow-Credentials, RETAIN_FIRST

复制代码

发起 WebSocket 毗连的时间，会报如下错误，这是由于修改了只读的哀求头

1. java.lang.UnsupportedOperationException: null
2.         at org.springframework.http.ReadOnlyHttpHeaders.set(ReadOnlyHttpHeaders.java:108) ~[spring-web-6.0.9.jar:6.0.9]
3.         Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
4. Error has been observed at the following site(s):
5.         *__checkpoint ⇢ org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter [DefaultWebFilterChain]
6.         *__checkpoint ⇢ HTTP GET "/websocket/admin?token=eyJhbGciOiJIUzUxMiIsInppcCI6IkdaSVAifQ.H4sIAAAAAAAA_6tWKi5NUrJScgwN8dANDXYNUtJRSq0oULIyNDe2NDMyNrYw0lEqLU4t8kwBilmYmZgZm5sbG5mbGViYGpgYQyX9EnNTgYYkpuRm5ilBhEIqC4BCRrUAvgeVqmEAAAA.e7wanr0gKu4FD-Y_afO2MEIECxZ6oMKGlf8zarZp-GOmzqL5n354gasKr7GKKs4H3Pq0CYJQECO_Rv9ixGsvZQ" [ExceptionHandlingWebHandler]
7. Original Stack Trace:
8.                 at org.springframework.http.ReadOnlyHttpHeaders.set(ReadOnlyHttpHeaders.java:108) ~[spring-web-6.0.9.jar:6.0.9]

复制代码

因此须要将上述设置删除，假如还须要这些默认设置，可以到具体的路由下面设置，就像下面一样

1. spring:
2.   cloud:
3.     gateway:
4.       routes:
5.         - id: vrs-admin
6.           uri: lb://vrs-admin
7.           predicates:
8.             - Path=/admin/**
9.           filters:
10.             - DedupeResponseHeader=Vary Access-Control-Allow-Origin Access-Control-Allow-Credentials, RETAIN_FIRST
11.             - name: TokenValidate
12.               args:
13.                 whitePathList:
14.                   - /admin/user/v1/login
15.                   - /admin/user/v1/wechatLogin
16.                   - ...

复制代码

WebSocket设置类

设置类 WebSocketConfig 重要用于设置和初始化 WebSocket 服务器端点，并处理惩罚与 WebSocket 毗连相干的操纵，具体功能如下：

• Spring Bean 注册：通过 @Configuration 注解标明这是一个 Spring 设置类。在该类中界说了一个 @Bean 方法 serverEndpointExporter()，它返回一个 ServerEndpointExporter 实例。这个实例的作用是主动注册利用了 @ServerEndpoint 注解声明的 WebSocket 端点对象到 Spring 容器中。
  
• 握手哀求修改：modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) 方法重写了父类中的同名方法，用于在创建 WebSocket 毗连前对握手哀求举行自界说修改。在这个例子中，方法实验从握手哀求参数中获取名为 “token” 的参数，并将其存储在 ServerEndpointConfig 对象的用户属性中（即 sec.getUserProperties().put("token", token);）。这使得后续逻辑可以通过访问端点设置对象来获取令牌信息。
  
• 端点实例化：getEndpointInstance(Class<T> clazz) 方法重写了父类的方法，用于提供自界说逻辑来实例化被 @ServerEndpoint 标注的 WebSocket 端点类。在这个实现中，它直接调用了父类的实现 super.getEndpointInstance(clazz) 来创建端点实例。通常环境下，除非须要特殊的实例化逻辑，否则可以直接利用父类的默认实现。
  

1. package com.vrs.config;
3. import jakarta.websocket.HandshakeResponse;
4. import jakarta.websocket.server.HandshakeRequest;
5. import jakarta.websocket.server.ServerEndpointConfig;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
8. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
10. import java.util.List;
11. import java.util.Map;
13. /**
14. * @Author dam
15. * @create 2025/1/24 15:25
16. */
17. @Configuration
18. public class WebSocketConfig extends ServerEndpointConfig.Configurator {
20.     /**
21.      * 这个bean会自动注册使用了@ServerEndpoint注解声明的对象
22.      *
23.      * @return
24.      */
25.     @Bean
26.     public ServerEndpointExporter serverEndpointExporter() {
27.         return new ServerEndpointExporter();
28.     }
30.     /**
31.      * 建立握手时，连接前的操作
32.      */
33.     @Override
34.     public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) {
35.         // 获取请求参数
36.         Map<String, List<String>> parameterMap = request.getParameterMap();
37.         List<String> tokenList = parameterMap.get("token");
38.         if (tokenList != null && !tokenList.isEmpty()) {
39.             String token = tokenList.get(0);
40.             sec.getUserProperties().put("token", token);
41.         }
42.     }
44.     /**
45.      * 初始化端点对象,也就是被@ServerEndpoint所标注的对象
46.      */
47.     @Override
48.     public <T> T getEndpointInstance(Class<T> clazz) throws InstantiationException {
49.         return super.getEndpointInstance(clazz);
50.     }
51. }

复制代码

WebSocket服务类

WebSocketServer 类是为实现及时通讯而操持的，可以或许有效地管理多个客户端之间的双向通讯以及保持这些通讯的稳固性和可靠性。它通过 Spring 的 @Component 和 Jakarta WebSocket 的 @ServerEndpoint 注解被注册为一个 Spring Bean，并监听路径为 /websocket/{username} 的 WebSocket 哀求。该类利用一个静态的 ConcurrentHashMap 来存储每个用户的会话 (Session) 和末了一次活动时间，以跟踪在线用户和他们的生动状态。它实现了以下关键功能：

• 毗连管理：处理惩罚用户的毗连创建 (onOpen) 和关闭 (onClose) 事故，包罗校验用户提供的 token 是否有效。
  
• 消息处理惩罚：吸收来自客户端的消息 (onMessage) 并据此更新用户的末了活动时间，支持发送 PING/PONG 心跳消息来维持毗连。
  
• 心跳检测：通过定时任务每30秒查抄一次用户的心跳，若某用户凌驾60秒未活动，则主动断开其毗连，确保资源的有效利用。
  
• 消息发送：提供了一个方法用于向特定用户发送消息。
  

1. package com.vrs.controller;
3. import com.vrs.config.WebSocketConfig;
4. import com.vrs.constant.RedisCacheConstant;
5. import com.vrs.utils.JwtUtil;
6. import jakarta.websocket.*;
7. import jakarta.websocket.server.PathParam;
8. import jakarta.websocket.server.ServerEndpoint;
9. import lombok.extern.slf4j.Slf4j;
10. import org.springframework.beans.factory.annotation.Autowired;
11. import org.springframework.data.redis.core.StringRedisTemplate;
12. import org.springframework.stereotype.Component;
13. import org.springframework.util.StringUtils;
15. import java.io.IOException;
16. import java.util.Map;
17. import java.util.concurrent.ConcurrentHashMap;
18. import java.util.concurrent.Executors;
19. import java.util.concurrent.ScheduledExecutorService;
20. import java.util.concurrent.TimeUnit;
22. /**
23. * @Author dam
24. * @create 2024/1/24 14:32
25. */
26. // 将WebSocketServer注册为spring的一个bean
27. @ServerEndpoint(value = "/websocket/{username}", configurator = WebSocketConfig.class)
28. @Component
29. @Slf4j(topic = "WebSocketServer")
30. public class WebSocketServer {
32.     /**
33.      * 心跳检查间隔时间（单位：秒）
34.      */
35.     private static final int HEARTBEAT_INTERVAL = 30;
37.     /**
38.      * 心跳超时时间（单位：秒）
39.      */
40.     private static final int HEARTBEAT_TIMEOUT = 60;
42.     /**
43.      * 记录当前在线连接的客户端的session
44.      */
45.     private static final Map<String, Session> usernameAndSessionMap = new ConcurrentHashMap<>();
47.     /**
48.      * 记录用户最后一次活动时间
49.      */
50.     private static final Map<String, Long> lastActivityTimeMap = new ConcurrentHashMap<>();
52.     /**
53.      * 直接通过 Autowired 注入的话，redisTemplate为null，因此使用这种引入方式
54.      */
55.     private static StringRedisTemplate redisTemplate;
56.     @Autowired
57.     public void setRabbitTemplate(StringRedisTemplate redisTemplate) {
58.         WebSocketServer.redisTemplate = redisTemplate;
59.     }
61.     /**
62.      * 定时任务线程池，用于心跳检查
63.      */
64.     private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
66.     // 初始化心跳检查任务
67.     static {
68.         scheduler.scheduleAtFixedRate(WebSocketServer::checkHeartbeat, HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL, TimeUnit.SECONDS);
69.     }
71.     /**
72.      * 浏览器和服务端连接建立成功之后会调用这个方法
73.      */
74.     @OnOpen
75.     public void onOpen(Session session, @PathParam("username") String username, EndpointConfig config) {
76.         // 校验 token 是否有效
77.         String token = (String) config.getUserProperties().get("token");
78.         boolean validToken = validToken(token);
79.         if (!validToken) {
80.             try {
81.                 session.close(new CloseReason(CloseReason.CloseCodes.VIOLATED_POLICY, "无效的token，请先登录"));
82.             } catch (IOException e) {
83.                 e.printStackTrace();
84.             }
85.         }
87.         // 如果用户已存在，关闭旧连接
88.         if (usernameAndSessionMap.containsKey(username)) {
89.             Session oldSession = usernameAndSessionMap.get(username);
90.             if (oldSession != null && oldSession.isOpen()) {
91.                 try {
92.                     oldSession.close();
93.                 } catch (IOException e) {
94.                     log.error("关闭旧连接时发生错误", e);
95.                 }
96.             }
97.         }
99.         // 记录新连接
100.         usernameAndSessionMap.put(username, session);
101.         // 记录用户活动时间
102.         lastActivityTimeMap.put(username, System.currentTimeMillis());
103.         log.info("有新用户加入，username={}, 当前在线人数为：{}", username, usernameAndSessionMap.size());
104.     }
106.     /**
107.      * 连接关闭调用的方法
108.      */
109.     @OnClose
110.     public void onClose(Session session, @PathParam("username") String username) throws IOException {
111.         try {
112.             if (session != null && session.isOpen()) {
113.                 session.close();
114.             }
115.         } catch (IOException e) {
116.             log.error("关闭连接时发生错误", e);
117.         } finally {
118.             usernameAndSessionMap.remove(username);
119.             lastActivityTimeMap.remove(username);
120.             log.info("有一连接关闭，移除username={}的用户session, 当前在线人数为：{}", username, usernameAndSessionMap.size());
121.         }
122.     }
124.     /**
125.      * 发生错误的时候会调用这个方法
126.      */
127.     @OnError
128.     public void onError(Session session, Throwable error) {
129.         log.error("发生错误，原因：" + error.getMessage());
130.         error.printStackTrace();
131.     }
133.     /**
134.      * 收到客户端消息时调用
135.      */
136.     @OnMessage
137.     public void onMessage(String message, Session session, @PathParam("username") String username) {
138.         // 更新用户最后一次活动时间
139.         lastActivityTimeMap.put(username, System.currentTimeMillis());
141.         if ("PING".equals(message)) {
142.             log.debug("收到来自 {} 的心跳检测请求", username);
143.         } else {
144.             log.info("收到来自 {} 的消息: {}", username, message);
145.         }
146.     }
148.     /**
149.      * 服务端发送消息给客户端
150.      */
151.     public void sendMessage(String toUsername, String message) {
152.         try {
153.             Session toSession = usernameAndSessionMap.get(toUsername);
154.             if (toSession != null && toSession.isOpen()) {
155.                 toSession.getBasicRemote().sendText(message);
156.             } else {
157.                 log.warn("用户 {} 的会话已关闭或不存在", toUsername);
158.             }
159.         } catch (Exception e) {
160.             log.error("服务端发送消息给客户端失败", e);
161.         }
162.     }
165.     /**
166.      * 关闭心跳检测超时的 session
167.      */
168.     private static void checkHeartbeat() {
169.         long currentTime = System.currentTimeMillis();
170.         for (Map.Entry<String, Long> entry : lastActivityTimeMap.entrySet()) {
171.             String username = entry.getKey();
172.             long lastActivityTime = entry.getValue();
173.             if (currentTime - lastActivityTime > HEARTBEAT_TIMEOUT * 1000) {
174.                 log.info("用户 {} 心跳超时，关闭连接", username);
175.                 Session session = usernameAndSessionMap.get(username);
176.                 if (session != null) {
177.                     try {
178.                         session.close();
179.                     } catch (IOException e) {
180.                         log.error("关闭连接时发生错误", e);
181.                     }
182.                 }
183.                 usernameAndSessionMap.remove(username);
184.                 lastActivityTimeMap.remove(username);
185.             }
186.         }
187.     }
189.     /**
190.      * 校验 token 有效
191.      *
192.      * @param token
193.      * @return
194.      */
195.     private boolean validToken(String token) {
196.         String userName = "";
197.         try {
198.             // 如果从 token 中解析用户名错误，说明 token 是捏造的，或者已经失效
199.             userName = JwtUtil.getUsername(token);
200.         } catch (Exception e) {
201.             return false;
202.         }
203.         if (StringUtils.hasText(userName) && StringUtils.hasText(token) &&
204.                 (redisTemplate.opsForHash().get(RedisCacheConstant.USER_LOGIN_KEY + userName, token)) != null) {
205.             // --if-- 如果可以通过 token 从 Redis 中获取到用户的登录信息，说明通过校验
206.             return true;
207.         }
208.         return false;
209.     }
211. }

复制代码

MQ斲丧者

1. package com.vrs.rocketMq.listener;
3. import com.vrs.constant.RocketMqConstant;
4. import com.vrs.controller.WebSocketServer;
5. import com.vrs.domain.dto.mq.WebsocketMqDTO;
6. import com.vrs.templateMethod.MessageWrapper;
7. import lombok.RequiredArgsConstructor;
8. import lombok.SneakyThrows;
9. import lombok.extern.slf4j.Slf4j;
10. import org.apache.rocketmq.spring.annotation.MessageModel;
11. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
12. import org.apache.rocketmq.spring.annotation.SelectorType;
13. import org.apache.rocketmq.spring.core.RocketMQListener;
14. import org.springframework.stereotype.Component;
16. /**
17. * 执行预订流程 消费者
18. *
19. * @Author dam
20. * @create 2024/9/20 21:30
21. */
22. @Slf4j(topic = RocketMqConstant.VENUE_TOPIC)
23. @Component
24. @RocketMQMessageListener(topic = RocketMqConstant.VENUE_TOPIC,
25.         consumerGroup = RocketMqConstant.VENUE_CONSUMER_GROUP + "-" + RocketMqConstant.WEBSOCKET_SEND_MESSAGE_TAG,
26.         // 需要使用广播模式
27.         messageModel = MessageModel.BROADCASTING,
28.         // 监听tag
29.         selectorType = SelectorType.TAG,
30.         selectorExpression = RocketMqConstant.WEBSOCKET_SEND_MESSAGE_TAG
31. )
32. @RequiredArgsConstructor
33. public class WebSocketSendMessageListener implements RocketMQListener<MessageWrapper<WebsocketMqDTO>> {
35.     private final WebSocketServer webSocketServer;
37.     /**
38.      * 消费消息的方法
39.      * 方法报错就会拒收消息
40.      *
41.      * @param messageWrapper 消息内容，类型和上面的泛型一致。如果泛型指定了固定的类型，消息体就是我们的参数
42.      */
43.     @SneakyThrows
44.     @Override
45.     public void onMessage(MessageWrapper<WebsocketMqDTO> messageWrapper) {
46.         // 开头打印日志，平常可 Debug 看任务参数，线上可报平安（比如消息是否消费，重新投递时获取参数等）
47.         log.info("[消费者] websocket发生消息给{}", messageWrapper.getMessage().getToUsername());
48.         webSocketServer.sendMessage(messageWrapper.getMessage().getToUsername(), messageWrapper.getMessage().getMessage());
49.     }
50. }

复制代码

启动类

1. package com.vrs;
3. import org.springframework.boot.SpringApplication;
4. import org.springframework.boot.autoconfigure.SpringBootApplication;
5. import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
7. /**
8. * @Author dam
9. * @create 2025/01/24 16:34
10. */
11. @SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
12. public class VrsWebSocketApplication {
13.     public static void main(String[] args) {
14.         SpringApplication.run(VrsWebSocketApplication.class, args);
15.     }
16. }

复制代码

设置文件

1. server:
2.   port: 7054
3. spring:
4.   profiles:
5.     active: dam
6.   application:
7.     name: vrs-websocket
8.   cloud:
9.     nacos:
10.       discovery:
11.         server-addr: 127.0.0.1:8848
12.   data:
13.     redis:
14.       host: 127.0.0.1
15.       port: 6379
16.       password: 12345678
17.       database: 0
18.       timeout: 1800000
19.       jedis:
20.         pool:
21.           max-active: 20 #最大连接数
22.           max-wait: -1    #最大阻塞等待时间(负数表示没限制)
23.           max-idle: 5    #最大空闲
24.           min-idle: 0     #最小空闲
25. rocketmq:
26.   # rocketMq的nameServer地址
27.   name-server: 127.0.0.1:9876
28.   producer:
29.     # 生产者组别
30.     group: vrs-websocket-group
31.     # 消息发送的超时时间
32.     send-message-timeout: 10000
33.     # 异步消息发送失败重试次数
34.     retry-times-when-send-async-failed: 1
35.     # 发送消息的最大大小，单位字节，这里等于4M
36.     max-message-size: 999999999

复制代码

留意事项

登录验证

为了防止被人恶意发生大量 WebSocket 毗连，占用服务器资源，因此在创建毗连的时间，须要举行登录验证，用户登录了才可以创建 WebSocket 毗连。
由于创建 WebSocket 毗连时，无法像之前的 http 哀求一样在哀求头携带 token 信息，因此之前网关实现的登录校验机制不收效，须要我们针对 WebSocket 毗连额外实现一套登录验证方式。
假设前端发起 WebSocket 毗连的代码如下：

1. new WebSocket("ws://localhost:7049/websocket/admin?token=dahidaho");

复制代码

WebSocket 设置类

在modifyHandshake中，将客户端发起毗连哀求时的 token 设置到属性中，如许背面就可以将 token 获取出来举行校验，假如说校验不通过，就关闭 WebSokcet 毗连
token校验

代码位于WebSocketServer类中，当调用validToken校验失败之后，通过session.close来关闭毗连

1. /**
2. * 校验 token 有效
3. *
4. * @param token
5. * @return
6. */
7. private boolean validToken(String token) {
8.     String userName = "";
9.     try {
10.         // 如果从 token 中解析用户名错误，说明 token 是捏造的，或者已经失效
11.         userName = JwtUtil.getUsername(token);
12.     } catch (Exception e) {
13.         return false;
14.     }
15.     if (StringUtils.hasText(userName) && StringUtils.hasText(token) &&
16.             (redisTemplate.opsForHash().get(RedisCacheConstant.USER_LOGIN_KEY + userName, token)) != null) {
17.         // --if-- 如果可以通过 token 从 Redis 中获取到用户的登录信息，说明通过校验
18.         return true;
19.     }
20.     return false;
21. }
23. /**
24. * 浏览器和服务端连接建立成功之后会调用这个方法
25. */
26. @OnOpen
27. public void onOpen(Session session, @PathParam("username") String username, EndpointConfig config) {
28.     // 校验 token 是否有效
29.     String token = (String) config.getUserProperties().get("token");
30.     boolean validToken = validToken(token);
31.     if (!validToken) {
32.         try {
33.             session.close(new CloseReason(CloseReason.CloseCodes.VIOLATED_POLICY, "无效的token，请先登录"));
34.         } catch (IOException e) {
35.             e.printStackTrace();
36.         }
37.     }
39.     // 如果用户已存在，关闭旧连接
40.     if (usernameAndSessionMap.containsKey(username)) {
41.         Session oldSession = usernameAndSessionMap.get(username);
42.         if (oldSession != null && oldSession.isOpen()) {
43.             try {
44.                 oldSession.close();
45.             } catch (IOException e) {
46.                 log.error("关闭旧连接时发生错误", e);
47.             }
48.         }
49.     }
51.     // 记录新连接
52.     usernameAndSessionMap.put(username, session);
53.     // 记录用户活动时间
54.     lastActivityTimeMap.put(username, System.currentTimeMillis());
55.     log.info("有新用户加入，username={}, 当前在线人数为：{}", username, usernameAndSessionMap.size());
56. }

复制代码

分布式 WebSocket

由于我们的项目是分布式架构的，假如vrs-websocket启动多个服务的话，须要处理惩罚如下标题：
WebSocketServer中的用户名及其对应的session信息usernameAndSessionMap是存储在当地的，假设发起毗连的时间，session被存储在呆板 1 上面。后续服务端要关照客户端时，怎么知道当前用户的信息是存储在呆板1、呆板 2 照旧呆板 3 呢？
由于 Session 无法直接序列化存储到 Redis 中，为相识决这个标题，本文通过借助消息队列来办理。
服务端要发送消息给客户端时，先将消息发送至消息队列中，消息设置为广播模式。后续多台摆设了vrs-websocket的呆板去消息队列中获取消息来斲丧，假如呆板查抄到了这条消息的吸收者 session 就在呆板上，则实验发送，否则直接 return 即可。
【消息生产者】

1. package com.vrs.rocketMq.producer;
3. import cn.hutool.core.util.StrUtil;
4. import com.vrs.constant.RocketMqConstant;
5. import com.vrs.domain.dto.mq.WebsocketMqDTO;
6. import com.vrs.templateMethod.AbstractCommonSendProduceTemplate;
7. import com.vrs.templateMethod.BaseSendExtendDTO;
8. import com.vrs.templateMethod.MessageWrapper;
9. import lombok.extern.slf4j.Slf4j;
10. import org.apache.rocketmq.common.message.MessageConst;
11. import org.springframework.messaging.Message;
12. import org.springframework.messaging.support.MessageBuilder;
13. import org.springframework.stereotype.Component;
15. import java.util.UUID;
17. /**
18. * websocket发送消息 生产者
19. *
20. * @Author dam
21. * @create 2024/9/20 16:00
22. */
23. @Slf4j
24. @Component
25. public class WebsocketSendMessageProducer extends AbstractCommonSendProduceTemplate<WebsocketMqDTO> {
27.     @Override
28.     protected BaseSendExtendDTO buildBaseSendExtendParam(WebsocketMqDTO messageSendEvent) {
29.         return BaseSendExtendDTO.builder()
30.                 .eventName("执行时间段预定")
31.                 .topic(RocketMqConstant.VENUE_TOPIC)
32.                 .tag(RocketMqConstant.WEBSOCKET_SEND_MESSAGE_TAG)
33.                 .sentTimeout(2000L)
34.                 .build();
35.     }
37.     @Override
38.     protected Message<?> buildMessage(WebsocketMqDTO messageSendEvent, BaseSendExtendDTO requestParam) {
39.         String keys = StrUtil.isEmpty(requestParam.getKeys()) ? UUID.randomUUID().toString() : requestParam.getKeys();
40.         return MessageBuilder
41.                 .withPayload(new MessageWrapper(keys, messageSendEvent))
42.                 .setHeader(MessageConst.PROPERTY_KEYS, keys)
43.                 .setHeader(MessageConst.PROPERTY_TAGS, requestParam.getTag())
44.                 .build();
45.     }
46. }

复制代码

【消息斲丧者】
斲丧者的代码就在具体实现中，这里不重复放
【利用】

1. // 通过 websocket 发送消息，通知前端
2. websocketSendMessageProducer.sendMessage(WebsocketMqDTO.builder()
3.         .toUsername(orderDO.getUserName())
4.         .message(JSON.toJSONString(orderDO))
5.         .build());

复制代码

心跳检测

用户创建 WebSocket 毗连之后的 session 数据是存储在服务器当地的，随着毗连数目标增长，session会占用大量的内存，心跳检测是为了定期整理那些无效的毗连。
在WebSocketServer中，通过定时任务每30秒查抄一次客户端的心跳状态，记录每个用户的末了活动时间。假如当前时间与某用户末了活动时间之差凌驾60秒，则以为该用户心跳超时，服务端将关闭其WebSocket毗连并整理相干记录。客户端需定期向服务端发送"ING"消息以维持毗连生动，确保不会因超时而被服务端断开。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。