From c38b35bc968b876b478f149f69291b6386cbe63b Mon Sep 17 00:00:00 2001
From: zhaowenyu <35589726@qq.com>
Date: Tue, 28 Oct 2025 15:04:51 +0800
Subject: [PATCH] 2025-10-28 zwy 更新WebSocket代码。
---
ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsOutputReceiver.java | 3
ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsReceiver.java | 2
ard-sample/src/main/java/com/ard/sample/component/websocket/model/Payload.java | 24 ++++
ard-sample/src/main/java/com/ard/sample/component/websocket/config/WebSocketDockConfig.java | 33 +++-
ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java | 189 +++++++++++++++++++++++++++++++
ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java | 27 ++++
ard-sample/src/main/java/com/ard/sample/component/websocket/model/WebSocketMessage.java | 25 ++++
7 files changed, 292 insertions(+), 11 deletions(-)
diff --git a/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsOutputReceiver.java b/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsOutputReceiver.java
index 46447fd..0d90eb0 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsOutputReceiver.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsOutputReceiver.java
@@ -1,5 +1,6 @@
package com.ard.sample.component.mqtt.model;
+import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
/**
@@ -10,7 +11,9 @@
@Data
public class EventsOutputReceiver {
+ @JsonProperty("status")
private String status;
+ @JsonProperty("progress")
private OutputProgressReceiver progress;
}
diff --git a/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsReceiver.java b/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsReceiver.java
index e1f6f86..c1ddad7 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsReceiver.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/mqtt/model/EventsReceiver.java
@@ -26,4 +26,6 @@
private String sn;
+ private String type;
+
}
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/config/WebSocketDockConfig.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/config/WebSocketDockConfig.java
index da048be..f6df97e 100644
--- a/ard-sample/src/main/java/com/ard/sample/component/websocket/config/WebSocketDockConfig.java
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/config/WebSocketDockConfig.java
@@ -1,18 +1,29 @@
package com.ard.sample.component.websocket.config;
-import org.springframework.context.annotation.Bean;
+import com.ard.sample.component.websocket.handler.*;
+import com.ard.sample.component.websocket.interceptor.DockHandshakeInterceptor;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
-import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+import org.springframework.web.socket.config.annotation.EnableWebSocket;
+import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
-/**
- * 启用WebSocket支持，注册端点处理器
- */
@Configuration
-public class WebSocketDockConfig {
+@EnableWebSocket // 启用原生WebSocket支持
+public class WebSocketDockConfig implements WebSocketConfigurer {
- // 自动注册@ServerEndpoint注解的类
- @Bean
- public ServerEndpointExporter serverEndpointExporter() {
- return new ServerEndpointExporter();
+ @Autowired
+ private DockWebSocketHandler dockWebSocketHandler;
+
+ // 注册WebSocket处理器和路径
+ @Override
+ public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
+ // 1. 注册处理器：路径 "/ws/dock/**" 由 DockWebSocketHandler 处理
+ // 2. 添加握手拦截器：验证用户身份DockHandshakeInterceptor
+ // 3. 配置跨域：允许所有域名（生产环境需限制）
+ registry.addHandler(dockWebSocketHandler, "/ws/dock/**")
+ .addInterceptors(new DockHandshakeInterceptor())
+ .setAllowedOrigins("*");
+
}
-}
\ No newline at end of file
+}
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java
new file mode 100644
index 0000000..8d488f3
--- /dev/null
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/enums/MessageType.java
@@ -0,0 +1,27 @@
+package com.ard.sample.component.websocket.enums;
+
+public enum MessageType {
+ STATUS("status"),
+ CONTROL("control"),
+ PROGRESS("progress"),
+ ALL("all");
+
+ private final String type;
+
+ MessageType(String type) {
+ this.type = type;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public static MessageType fromString(String type) {
+ for (MessageType messageType : MessageType.values()) {
+ if (messageType.type.equalsIgnoreCase(type)) {
+ return messageType;
+ }
+ }
+ throw new IllegalArgumentException("Unknown message type: " + type);
+ }
+}
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
new file mode 100644
index 0000000..98c1618
--- /dev/null
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java
@@ -0,0 +1,189 @@
+package com.ard.sample.component.websocket.handler;
+
+import com.alibaba.fastjson2.JSON;
+import com.ard.sample.component.websocket.model.Payload;
+import com.ard.sample.component.websocket.model.WebSocketMessage;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+public class DockWebSocketHandler extends TextWebSocketHandler {
+
+ // 存储用户会话：userId -> WebSocketSession
+ private final Map<String, WebSocketSession> userSessions = new ConcurrentHashMap<>();
+
+ // 订阅关系：key=订阅标识(sn+type)，value=订阅用户ID集合
+ private final Map<String, Set<String>> subscriptions = new ConcurrentHashMap<>();
+
+ // 关键：为每个WebSocketSession分配独立的锁对象，确保写操作同步
+ private final Map<WebSocketSession, Object> sessionLocks = new ConcurrentHashMap<>();
+
+ /**
+ * 连接建立后回调
+ */
+ @Override
+ public void afterConnectionEstablished(WebSocketSession session) throws Exception {
+ // 从会话属性中获取认证后的用户ID
+ String userId = (String) session.getAttributes().get("userId");
+ if (userId == null) {
+ session.close(CloseStatus.POLICY_VIOLATION.withReason("未认证"));
+ return;
+ }
+
+ userSessions.put(userId, session);
+ // 为新会话初始化锁对象
+ sessionLocks.put(session, new Object());
+ System.out.println("用户[" + userId + "]已连接，当前在线人数：" + userSessions.size());
+ }
+
+ /**
+ * 处理客户端发送的消息
+ */
+ @Override
+ protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
+ String userId = (String) session.getAttributes().get("userId");
+ String jsonMessage = message.getPayload();
+
+ try {
+ // 解析客户端消息
+ WebSocketMessage webSocketMessage = JSON.parseObject(jsonMessage, WebSocketMessage.class);
+ String type = webSocketMessage.getType();
+ Payload payload = webSocketMessage.getPayload();
+
+ // 验证消息类型是否合法
+ if (!Arrays.asList("status", "control", "progress").contains(type)) {
+ session.sendMessage(new TextMessage("不支持的消息类型：" + type));
+ return;
+ }
+
+ // 处理订阅/取消订阅
+ handleSubscription(userId, type, payload);
+
+ } catch (Exception e) {
+ System.err.println("消息处理错误：" + e.getMessage());
+ session.sendMessage(new TextMessage("消息格式错误"));
+ }
+ }
+
+ /**
+ * 处理订阅/取消订阅逻辑
+ */
+ private void handleSubscription(String userId, String type, Payload payload) {
+ String sn = payload.getSn();
+ boolean subscribe = payload.isSubscribe();
+ String subscriptionKey = getSubscriptionKey(sn, type); // 生成唯一订阅标识
+
+ if (subscribe) {
+ // 订阅操作
+ subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId);
+ System.out.println("用户[" + userId + "]订阅了 设备[" + sn + "]的[" + type + "]类型消息");
+ } else {
+ // 取消订阅操作
+ if (subscriptions.containsKey(subscriptionKey)) {
+ subscriptions.get(subscriptionKey).remove(userId);
+ if (subscriptions.get(subscriptionKey).isEmpty()) {
+ subscriptions.remove(subscriptionKey);
+ }
+ System.out.println("用户[" + userId + "]取消订阅了 设备[" + sn + "]的[" + type + "]类型消息");
+ }
+ }
+ }
+
+ /**
+ * 连接关闭后回调
+ */
+ @Override
+ public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
+ String userId = (String) session.getAttributes().get("userId");
+ if (userId != null) {
+ userSessions.remove(userId);
+ sessionLocks.remove(session); // 移除会话锁
+ // 清除该用户的所有订阅
+ removeUserAllSubscriptions(userId);
+ System.out.println("用户[" + userId + "]已断开连接，当前在线人数：" + userSessions.size());
+ }
+ }
+
+ /**
+ * 清除用户的所有订阅
+ */
+ private void removeUserAllSubscriptions(String userId) {
+ for (Map.Entry<String, Set<String>> entry : subscriptions.entrySet()) {
+ entry.getValue().remove(userId);
+ if (entry.getValue().isEmpty()) {
+ subscriptions.remove(entry.getKey());
+ }
+ }
+ }
+
+ /**
+ * 向订阅了特定设备特定类型的用户发送消息
+ */
+ public void sendMessageToSubscribers(String sn, String type, String content) {
+ String subscriptionKey = getSubscriptionKey(sn, type);
+ Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, new HashSet<>());
+
+ if (subscriberIds.isEmpty()) {
+ //System.out.println("没有用户订阅 设备[" + sn + "]的[" + type + "]类型消息");
+ return;
+ }
+
+ try {
+ // 构建发送的消息
+// WebSocketMessage message = new WebSocketMessage();
+// message.setType(type);
+//
+// Payload payload = new Payload();
+// payload.setSn(sn);
+// payload.setSubscribe(true); // 仅作为标识，无实际意义
+// message.setPayload(payload);
+//
+// String jsonMessage = JSON.toJSONString(message);
+
+ // 向所有订阅者发送消息
+ for (String userId : subscriberIds) {
+ WebSocketSession session = userSessions.get(userId);
+ // 1. 检查会话是否有效
+ if (session == null || !session.isOpen()) {
+ // 移除无效会话
+ userSessions.remove(userId);
+ sessionLocks.remove(session);
+ continue;
+ }
+
+ // 2. 获取当前会话的锁对象，确保同步写操作
+ Object lock = sessionLocks.get(session);
+ if (lock == null) {
+ continue; // 理论上不会出现，防止NPE
+ }
+
+ // 3. 同步发送消息，避免并发冲突
+ synchronized (lock) {
+ session.sendMessage(new TextMessage(content));
+ System.out.println("向用户[" + userId + "]发送 设备[" + sn + "]的[" + type + "]类型消息:" + content);
+ }
+// if (session != null && session.isOpen()) {
+// session.sendMessage(new TextMessage(content));
+// System.out.println("向用户[" + userId + "]发送 设备[" + sn + "]的[" + type + "]类型消息");
+// System.out.println(content);
+// }
+ }
+ } catch (IOException e) {
+ System.err.println("发送消息错误：" + e.getMessage());
+ }
+ }
+
+ /**
+ * 生成订阅标识（sn+type的组合）
+ */
+ private String getSubscriptionKey(String sn, String type) {
+ return sn + ":" + type;
+ }
+}
\ No newline at end of file
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/model/Payload.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/model/Payload.java
new file mode 100644
index 0000000..cbe293f
--- /dev/null
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/model/Payload.java
@@ -0,0 +1,24 @@
+package com.ard.sample.component.websocket.model;
+
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * 消息内容实体类
+ * */
+@Data
+//@Builder
+public class Payload {
+
+// public Payload() {}
+
+ /**
+ * 设备序列号
+ * */
+ private String sn;
+
+ /**
+ * 是否订阅
+ * */
+ private boolean subscribe;
+}
diff --git a/ard-sample/src/main/java/com/ard/sample/component/websocket/model/WebSocketMessage.java b/ard-sample/src/main/java/com/ard/sample/component/websocket/model/WebSocketMessage.java
new file mode 100644
index 0000000..2b90c9a
--- /dev/null
+++ b/ard-sample/src/main/java/com/ard/sample/component/websocket/model/WebSocketMessage.java
@@ -0,0 +1,25 @@
+package com.ard.sample.component.websocket.model;
+
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * 消息实体类
+ * */
+@Data
+//@Builder
+public class WebSocketMessage {
+
+// public WebSocketMessage() {}
+
+ /**
+ * 访问类型，例如：status、control、progress等
+ * */
+ private String type;
+
+ /**
+ * 访问的消息实体类
+ * */
+ private Payload payload;
+ //private String userId; // 服务器端添加的用户ID
+}
--
Gitblit v1.9.3