tcjlj/flvx
1
0
Fork 0
mirror of https://github.com/Sagit-chu/flvx.git synced 2026-09-29 07:56:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

gost通讯改为ws

Browse Source
This commit is contained in:
qaq
2025-06-25 10:49:06 +08:00
parent f3e9b31ed3
commit f7e82cdc8b
47 changed files with 3528 additions and 1513 deletions
Download Patch File Download Diff File Expand all files Collapse all files
springboot-backend/src/main/java/com/admin/common/utils/WebSocketServer.java
+165 -19
View File
@@ -1,6 +1,9 @@
package com.admin.common.utils;
import com.admin.common.dto.GostConfigDto;
import com.admin.common.dto.GostDto;
import com.admin.common.task.CheckGostConfigAsync;
import com.admin.entity.Node;
import com.admin.service.NodeService;
import com.alibaba.fastjson.JSONObject;
@@ -14,8 +17,11 @@ import org.springframework.web.socket.handler.TextWebSocketHandler;
import javax.annotation.Resource;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.UUID;
@Slf4j
@@ -24,25 +30,62 @@ public class WebSocketServer extends TextWebSocketHandler {
@Resource
NodeService nodeService;
// 存储所有活跃的 WebSocket 连接
@Resource
CheckGostConfigAsync checkGostConfigAsync;
// 存储所有活跃的 WebSocket 连接（
private static final CopyOnWriteArraySet<WebSocketSession> activeSessions = new CopyOnWriteArraySet<>();
// 存储节点ID和对应的WebSocket session映射
private static final ConcurrentHashMap<Long, WebSocketSession> nodeSessions = new ConcurrentHashMap<>();
// 为每个session提供锁对象，防止并发发送消息
private static final ConcurrentHashMap<String, Object> sessionLocks = new ConcurrentHashMap<>();
// 存储等待响应的请求，key为requestId，value为CompletableFuture
private static final ConcurrentHashMap<String, CompletableFuture<GostDto>> pendingRequests = new ConcurrentHashMap<>();
//接受客户端消息
@Override
public void handleTextMessage(WebSocketSession session, TextMessage message) {
try {
if (StringUtils.isNoneBlank(message.getPayload())) {
//log.info("收到消息: {}", message.getPayload());
String id = session.getAttributes().get("id").toString();
String type = session.getAttributes().get("type").toString();
// 先发送确认消息
sendToUser(session, "ok");
if (message.getPayload().contains("memory_usage")){
// 先发送确认消息
sendToUser(session, "{\"type\":\"call\"}");
} else if (message.getPayload().contains("config_report")) {
log.info("收到消息: {}", message.getPayload());
JSONObject jsonObject = JSONObject.parseObject(message.getPayload());
String string = jsonObject.getString("data");
GostConfigDto gostConfigDto = JSONObject.parseObject(string, GostConfigDto.class);
checkGostConfigAsync.cleanNodeConfigs(id, gostConfigDto);
} else if (message.getPayload().contains("requestId")) {
log.info("收到消息: {}", message.getPayload());
// 处理命令响应消息
try {
JSONObject responseJson = JSONObject.parseObject(message.getPayload());
String requestId = responseJson.getString("requestId");
String responseMessage = responseJson.getString("message");
if (requestId != null) {
CompletableFuture<GostDto> future = pendingRequests.remove(requestId);
if (future != null) {
GostDto result = new GostDto();
result.setMsg(responseMessage != null ? responseMessage : "无响应消息");
future.complete(result);
}
}
} catch (Exception e) {
log.error("处理响应消息失败: {}", e.getMessage(), e);
}
} else {
log.info("收到消息: {}", message.getPayload());
}
// 如果是节点类型，转发消息给其他会话
if (Objects.equals(type, "1")) {
JSONObject jsonObject = new JSONObject();
@@ -70,21 +113,30 @@ public class WebSocketServer extends TextWebSocketHandler {
try {
String id = session.getAttributes().get("id").toString();
String type = session.getAttributes().get("type").toString();
if (!Objects.equals(type, "1")) {
// 网页管理员连接
activeSessions.add(session);
}else {
Node byId = nodeService.getById(id);
} else {
// 客户端节点连接
Long nodeId = Long.valueOf(id);
nodeSessions.put(nodeId, session);
// 更新节点状态为在线
Node byId = nodeService.getById(nodeId);
if (byId != null) {
byId.setStatus(1);
nodeService.updateById(byId);
// 广播节点上线状态给所有管理员
JSONObject res = new JSONObject();
res.put("id", id);
res.put("type", "status");
res.put("data", 1);
broadcastMessage(res.toJSONString());
}
}
log.info("WebSocket 连接建立成功 - id: {}, type: {}, 当前连接数: {}", id, type, activeSessions.size());
} catch (Exception e) {
log.error("建立连接时发生异常: {}", e.getMessage(), e);
@@ -100,25 +152,35 @@ public class WebSocketServer extends TextWebSocketHandler {
String sessionId = session.getId();
if (!Objects.equals(type, "1")) {
// 连接关闭
activeSessions.remove(session);
}else {
Node byId = nodeService.getById(id);
} else {
// 客户端节点连接关闭
Long nodeId = Long.valueOf(id);
nodeSessions.remove(nodeId);
// 更新节点状态为离线
Node byId = nodeService.getById(nodeId);
if (byId != null) {
byId.setStatus(0);
nodeService.updateById(byId);
JSONObject res = new JSONObject();
res.put("id", id);
res.put("type", "status");
res.put("data", 0);
broadcastMessage(res.toJSONString());
}
}
// 清理session锁对象
sessionLocks.remove(sessionId);
log.info("WebSocket 连接关闭 - id: {}, sessionId: {}, 关闭状态: {}, 当前连接数: {}",
id, sessionId, status, activeSessions.size());
// 清理该节点的待处理请求
if (Objects.equals(type, "1")) {
clearPendingRequestsForNode(Long.valueOf(id));
}
} catch (Exception e) {
log.error("关闭连接时发生异常: {}", e.getMessage(), e);
@@ -139,15 +201,34 @@ public class WebSocketServer extends TextWebSocketHandler {
}
} catch (Exception e) {
log.error("发送WebSocket消息失败 [sessionId={}]: {}", sessionId, e.getMessage());
activeSessions.remove(socketSession);
sessionLocks.remove(sessionId);
cleanupSession(socketSession);
}
}
} else {
activeSessions.remove(socketSession);
if (socketSession != null) {
sessionLocks.remove(socketSession.getId());
}
cleanupSession(socketSession);
}
}
/**
* 清理失效的session，自动识别是节点session还是管理员session
*/
private static void cleanupSession(WebSocketSession session) {
if (session == null) return;
String sessionId = session.getId();
// 清理session锁
sessionLocks.remove(sessionId);
boolean removedFromAdmin = activeSessions.remove(session);
if (!removedFromAdmin) {
nodeSessions.entrySet().removeIf(entry -> {
if (entry.getValue() == session) {
return true;
}
return false;
});
}
}
@@ -157,4 +238,69 @@ public class WebSocketServer extends TextWebSocketHandler {
sendToUser(session, message);
}
}
/**
* 清理指定节点的待处理请求
*/
private static void clearPendingRequestsForNode(Long nodeId) {
// 完成所有待处理的请求，设置为连接断开错误
pendingRequests.entrySet().removeIf(entry -> {
CompletableFuture<GostDto> future = entry.getValue();
if (!future.isDone()) {
GostDto errorResult = new GostDto();
errorResult.setMsg("节点连接已断开");
future.complete(errorResult);
}
return true; // 移除所有请求
});
}
public static GostDto send_msg(Long node_id, Object msg, String type) {
WebSocketSession nodeSession = nodeSessions.get(node_id);
if (nodeSession == null) {
GostDto result = new GostDto();
result.setMsg("节点不在线");
return result;
}
if (!nodeSession.isOpen()) {
nodeSessions.remove(node_id);
sessionLocks.remove(nodeSession.getId());
GostDto result = new GostDto();
result.setMsg("节点连接已断开");
return result;
}
// 生成唯一的请求ID
String requestId = UUID.randomUUID().toString();
// 创建CompletableFuture用于等待响应
CompletableFuture<GostDto> future = new CompletableFuture<>();
pendingRequests.put(requestId, future);
try {
JSONObject data = new JSONObject();
data.put("type", type);
data.put("data", msg);
data.put("requestId", requestId);
sendToUser(nodeSession, data.toJSONString());
GostDto result = future.get(10, TimeUnit.SECONDS);
return result;
} catch (Exception e) {
pendingRequests.remove(requestId);
GostDto result = new GostDto();
if (e instanceof java.util.concurrent.TimeoutException) {
result.setMsg("等待响应超时");
} else {
result.setMsg("发送消息失败: " + e.getMessage());
}
log.error("发送消息到节点{}失败: {}", node_id, e.getMessage(), e);
return result;
}
}
}