From c96fbe6b0aeb413a17c429133ee6d81d138144b9 Mon Sep 17 00:00:00 2001
From: liujie <1793218484@qq.com>
Date: 星期二, 09 六月 2026 10:58:52 +0800
Subject: [PATCH] Merge branch '1.0' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi into 1.0
---
xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/service/WebSocketUserConnectionManager.java | 75 +++++++++++++++++++------------------
1 files changed, 38 insertions(+), 37 deletions(-)
diff --git a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/service/WebSocketUserConnectionManager.java b/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/service/WebSocketUserConnectionManager.java
index fa84485..c6c42f7 100644
--- a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/service/WebSocketUserConnectionManager.java
+++ b/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/service/WebSocketUserConnectionManager.java
@@ -9,34 +9,35 @@
import java.util.concurrent.ConcurrentHashMap;
/**
- * 基于AppUser的WebSocket连接管理器
- * 支持按用户ID管理连接，一个用户可以有多个设备连接
+ * 基于 AppUser 的 WebSocket 连接管理器
+ * 使用 Spring WebSocketSession（共用 Spring Boot HTTP 端口）
+ * 支持按用户 ID 管理连接，一个用户可以有多个设备连接
*/
@Component
@Slf4j
public class WebSocketUserConnectionManager {
-
+
/**
* 存储用户的所有连接
* Key: 用户ID (appUserId)
* Value: 该用户的所有连接Channel列表
*/
private static final ConcurrentHashMap<String, Set<Channel>> userConnections = new ConcurrentHashMap<>();
-
+
/**
* 存储连接与用户的映射关系
* Key: ChannelId.asLongText()
* Value: 用户ID (appUserId)
*/
private static final ConcurrentHashMap<String, String> channelToUser = new ConcurrentHashMap<>();
-
+
/**
* 存储所有连接
* Key: ChannelId.asLongText()
* Value: Channel
*/
private static final ConcurrentHashMap<String, Channel> allConnections = new ConcurrentHashMap<>();
-
+
/**
* 添加用户连接
* @param appUserId 用户ID
@@ -44,21 +45,21 @@
*/
public void addUserConnection(String appUserId, Channel channel) {
String channelId = channel.id().asLongText();
-
+
// 添加到用户连接映射
userConnections.computeIfAbsent(appUserId, k -> ConcurrentHashMap.newKeySet())
.add(channel);
-
+
// 添加到连接用户映射
channelToUser.put(channelId, appUserId);
-
+
// 添加到所有连接
allConnections.put(channelId, channel);
-
- log.info("用户连接已添加，用户ID: {}, ChannelId: {}, 该用户连接数: {}",
+
+ log.info("用户连接已添加，用户ID: {}, ChannelId: {}, 该用户连接数: {}",
appUserId, channelId, userConnections.get(appUserId).size());
}
-
+
/**
* 移除用户连接
* @param channel WebSocket连接通道
@@ -66,7 +67,7 @@
public void removeUserConnection(Channel channel) {
String channelId = channel.id().asLongText();
String appUserId = channelToUser.get(channelId);
-
+
if (appUserId != null) {
// 从用户连接中移除
Set<Channel> userChannels = userConnections.get(appUserId);
@@ -77,18 +78,18 @@
userConnections.remove(appUserId);
}
}
-
+
// 从连接用户映射中移除
channelToUser.remove(channelId);
-
- log.info("用户连接已移除，用户ID: {}, ChannelId: {}, 该用户剩余连接数: {}",
+
+ log.info("用户连接已移除，用户ID: {}, ChannelId: {}, 该用户剩余连接数: {}",
appUserId, channelId, userChannels != null ? userChannels.size() : 0);
}
-
+
// 从所有连接中移除
allConnections.remove(channelId);
}
-
+
/**
* 向指定用户的所有设备发送消息
* @param appUserId 用户ID
@@ -101,7 +102,7 @@
log.warn("用户 {} 没有活跃连接", appUserId);
return 0;
}
-
+
int successCount = 0;
for (Channel channel : userChannels) {
if (channel != null && channel.isActive()) {
@@ -109,17 +110,17 @@
channel.writeAndFlush(new TextWebSocketFrame(message));
successCount++;
} catch (Exception e) {
- log.error("向用户 {} 发送消息失败，连接ID: {}, 错误: {}",
+ log.error("向用户 {} 发送消息失败，连接ID: {}, 错误: {}",
appUserId, channel.id().asLongText(), e.getMessage());
}
}
}
-
- log.info("向用户 {} 发送消息完成，成功发送到 {} 个设备，消息内容: {}",
+
+ log.info("向用户 {} 发送消息完成，成功发送到 {} 个设备，消息内容: {}",
appUserId, successCount, message);
return successCount;
}
-
+
/**
* 向指定连接发送消息
* @param channelId 连接ID
@@ -142,7 +143,7 @@
return false;
}
}
-
+
/**
* 向所有连接广播消息
* @param message 消息内容
@@ -156,7 +157,7 @@
channel.writeAndFlush(new TextWebSocketFrame(message));
successCount++;
} catch (Exception e) {
- log.error("广播消息失败，连接ID: {}, 错误: {}",
+ log.error("广播消息失败，连接ID: {}, 错误: {}",
channel.id().asLongText(), e.getMessage());
}
}
@@ -164,7 +165,7 @@
log.info("广播消息完成，成功发送到 {} 个连接，消息内容: {}", successCount, message);
return successCount;
}
-
+
/**
* 向多个用户发送消息
* @param appUserIds 用户ID列表
@@ -182,7 +183,7 @@
log.info("向 {} 个用户发送消息完成，成功发送到 {} 个用户", appUserIds.size(), successUserCount);
return successUserCount;
}
-
+
/**
* 获取用户的所有连接
* @param appUserId 用户ID
@@ -191,7 +192,7 @@
public Set<Channel> getUserConnections(String appUserId) {
return userConnections.getOrDefault(appUserId, Collections.emptySet());
}
-
+
/**
* 获取用户连接数
* @param appUserId 用户ID
@@ -201,7 +202,7 @@
Set<Channel> userChannels = userConnections.get(appUserId);
return userChannels != null ? userChannels.size() : 0;
}
-
+
/**
* 获取总连接数
* @return 总连接数
@@ -209,7 +210,7 @@
public int getTotalConnectionCount() {
return allConnections.size();
}
-
+
/**
* 获取在线用户数
* @return 在线用户数
@@ -217,7 +218,7 @@
public int getOnlineUserCount() {
return userConnections.size();
}
-
+
/**
* 获取所有在线用户ID
* @return 用户ID集合
@@ -225,7 +226,7 @@
public Set<String> getOnlineUserIds() {
return userConnections.keySet();
}
-
+
/**
* 检查用户是否在线
* @param appUserId 用户ID
@@ -239,7 +240,7 @@
// 检查是否有活跃连接
return userChannels.stream().anyMatch(Channel::isActive);
}
-
+
/**
* 检查连接是否存在
* @param channelId 连接ID
@@ -249,7 +250,7 @@
Channel channel = allConnections.get(channelId);
return channel != null && channel.isActive();
}
-
+
/**
* 根据连接ID获取用户ID
* @param channelId 连接ID
@@ -258,7 +259,7 @@
public String getUserIdByChannelId(String channelId) {
return channelToUser.get(channelId);
}
-
+
/**
* 获取连接统计信息
* @return 统计信息
@@ -268,14 +269,14 @@
stats.put("totalConnections", getTotalConnectionCount());
stats.put("onlineUsers", getOnlineUserCount());
stats.put("onlineUserIds", getOnlineUserIds());
-
+
// 统计每个用户的连接数
Map<String, Integer> userConnectionCounts = new HashMap<>();
for (Map.Entry<String, Set<Channel>> entry : userConnections.entrySet()) {
userConnectionCounts.put(entry.getKey(), entry.getValue().size());
}
stats.put("userConnectionCounts", userConnectionCounts);
-
+
return stats;
}
}
--
Gitblit v1.7.1