From fe63fd580c5b65fa398dfe1fdf44466b26c6aadd Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Sat, 22 Mar 2025 13:38:30 +0800
Subject: [PATCH] 优化
---
ard-work/src/main/java/com/ard/tianyu/component/netty/TcpClientSession.java | 4
ard-work/src/main/java/com/ard/tianyu/component/websocket/handler/TyMessageHandler.java | 61 +++++++++++--------
ard-work/src/main/java/com/ard/tianyu/domain/enums/MessageType.java | 29 +++++++++
ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java | 4
ard-work/src/main/java/com/ard/tianyu/component/netty/handler/ClientHandler.java | 6 +-
ard-admin/src/main/resources/application-dev.yml | 1
ard-work/src/main/java/com/ard/tianyu/component/netty/NettyTcpClient.java | 17 +++--
7 files changed, 81 insertions(+), 41 deletions(-)
diff --git a/ard-admin/src/main/resources/application-dev.yml b/ard-admin/src/main/resources/application-dev.yml
index da01e00..4bff9d9 100644
--- a/ard-admin/src/main/resources/application-dev.yml
+++ b/ard-admin/src/main/resources/application-dev.yml
@@ -43,6 +43,7 @@
# 天宇tcp服务器地址
tcpserver: 59.110.237.99
tcpport: 8125
+
#流媒体配置
mediamtx:
# 流媒体服务器地址
diff --git a/ard-work/src/main/java/com/ard/tianyu/component/netty/TyNettyTcpClient.java b/ard-work/src/main/java/com/ard/tianyu/component/netty/NettyTcpClient.java
similarity index 84%
rename from ard-work/src/main/java/com/ard/tianyu/component/netty/TyNettyTcpClient.java
rename to ard-work/src/main/java/com/ard/tianyu/component/netty/NettyTcpClient.java
index cf88289..326c5b5 100644
--- a/ard-work/src/main/java/com/ard/tianyu/component/netty/TyNettyTcpClient.java
+++ b/ard-work/src/main/java/com/ard/tianyu/component/netty/NettyTcpClient.java
@@ -21,9 +21,9 @@
@Component
@Slf4j
-public class TyNettyTcpClient {
+public class NettyTcpClient {
- public static Map<String, TyBoxSession> channels = new HashMap<>();
+ public static Map<String, TcpClientSession> channels = new HashMap<>();
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
@@ -59,11 +59,11 @@
*/
public void initConnection(String host, Integer port) {
- log.info("初始化连接天宇tcp服务器【{}:{}】", host, port);
- String currentBoxSn = "";
+ log.info("初始化连接TCP服务器【{}:{}】", host, port);
+ String clientId = "clientId_1";
//判断是否已经存在连接，如果存在则关闭之前的连接
if (channels.containsKey(host + ":" + port)) {
- currentBoxSn = channels.get(host + ":" + port).getSubBoxSn();
+ clientId = channels.get(host + ":" + port).getClientId();
channels.get(host + ":" + port).getChannelFuture().channel().close();
}
@@ -71,12 +71,12 @@
Bootstrap bootstrap = getBootstrap();
ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture futureListener) -> {
if (futureListener.isSuccess()) {
- log.info("连接天宇TCP服务器成功【{}:{}】", host, port);
+ log.info("连接TCP服务器成功【{}:{}】", host, port);
TyTcpServerCache.putTcpServerCache(host + ":" + port, true);
} else {
final EventLoop eventLoop = futureListener.channel().eventLoop();
eventLoop.schedule(() -> {
- log.info("连接天宇TCP服务器失败【{}:{}】,正尝试重连...", host, port);
+ log.info("连接TCP服务器失败【{}:{}】,正尝试重连...", host, port);
TyTcpServerCache.putTcpServerCache(host + ":" + port, false);
try {
initConnection(host, port);
@@ -87,7 +87,8 @@
}
});
//将连接存入集合
- channels.put(host + ":" + port, TyBoxSession.builder().subBoxSn(currentBoxSn).channelFuture(future).isReConnect(!currentBoxSn.equals("")).build());
+ channels.put(host + ":" + port,
+ TcpClientSession.builder().ClientId(clientId).channelFuture(future).isReConnect(!clientId.isEmpty()).build());
}
//创建天宇连接
diff --git a/ard-work/src/main/java/com/ard/tianyu/component/netty/TyBoxSession.java b/ard-work/src/main/java/com/ard/tianyu/component/netty/TcpClientSession.java
similarity index 84%
rename from ard-work/src/main/java/com/ard/tianyu/component/netty/TyBoxSession.java
rename to ard-work/src/main/java/com/ard/tianyu/component/netty/TcpClientSession.java
index 907a3bd..0aff5a7 100644
--- a/ard-work/src/main/java/com/ard/tianyu/component/netty/TyBoxSession.java
+++ b/ard-work/src/main/java/com/ard/tianyu/component/netty/TcpClientSession.java
@@ -10,9 +10,9 @@
@Builder
@AllArgsConstructor
@NoArgsConstructor
-public class TyBoxSession {
+public class TcpClientSession {
- public String subBoxSn;
+ public String ClientId;
public ChannelFuture channelFuture;
diff --git a/ard-work/src/main/java/com/ard/tianyu/component/netty/handler/ClientHandler.java b/ard-work/src/main/java/com/ard/tianyu/component/netty/handler/ClientHandler.java
index a284d48..4ea5b91 100644
--- a/ard-work/src/main/java/com/ard/tianyu/component/netty/handler/ClientHandler.java
+++ b/ard-work/src/main/java/com/ard/tianyu/component/netty/handler/ClientHandler.java
@@ -1,7 +1,7 @@
package com.ard.tianyu.component.netty.handler;
import com.ard.tianyu.domain.cache.TyTcpServerCache;
-import com.ard.tianyu.component.netty.TyNettyTcpClient;
+import com.ard.tianyu.component.netty.NettyTcpClient;
import com.ard.tianyu.component.netty.command.BaseCmd;
import com.ard.tianyu.service.ITyService;
import io.netty.channel.ChannelHandler;
@@ -25,7 +25,7 @@
private ITyService tyService;
@Resource
- private TyNettyTcpClient tyNettyTcpClient;
+ private NettyTcpClient nettyTcpClient;
@Override
public void channelRead0(ChannelHandlerContext ctx, BaseCmd command) {
@@ -75,7 +75,7 @@
ctx.channel().eventLoop().schedule(() -> {
log.error("天宇服务端链接不上，开始重连操作...");
try {
- tyNettyTcpClient.initConnection(host, port);
+ nettyTcpClient.initConnection(host, port);
} catch (Exception e) {
log.error("天宇服务端连接异常: {}", e.getMessage());
}
diff --git a/ard-work/src/main/java/com/ard/tianyu/component/websocket/handler/TyMessageHandler.java b/ard-work/src/main/java/com/ard/tianyu/component/websocket/handler/TyMessageHandler.java
index f5094b0..bfb4cfa 100644
--- a/ard-work/src/main/java/com/ard/tianyu/component/websocket/handler/TyMessageHandler.java
+++ b/ard-work/src/main/java/com/ard/tianyu/component/websocket/handler/TyMessageHandler.java
@@ -3,6 +3,7 @@
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.ard.tianyu.component.websocket.WebSocketSessionManager;
+import com.ard.tianyu.domain.enums.MessageType;
import com.ard.tianyu.domain.message.*;
import com.ard.tianyu.service.control.UavControlService;
import com.ard.tianyu.component.websocket.utils.WebSocketUtil;
@@ -12,6 +13,7 @@
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
+
import javax.annotation.Resource;
@@ -33,34 +35,39 @@
private UavControlManager uavControlManager;
@Resource
private UavAllManager uavAllManager;
+
@Override
public void handleTextMessage(WebSocketSession session, TextMessage message) {
- String userIdStr = (String) session.getAttributes().get("userId");
- Long userId = Long.parseLong(userIdStr);
- String payload = message.getPayload();
- log.info("【cmd】Received message: {}", payload);
- log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount());
-
- BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class);
- switch (baseMessage.getType()) {
- case "stream":
- handleStreamMessage(payload, userId);
- break;
- case "state":
- handleStateMessage(payload, userId);
- break;
- case "telemetry":
- handleTelemetryMessage(payload, userId);
- break;
- case "signal":
- handleSignalMessage(payload, userId);
- break;
- case "controller":
- handleControlMessage(payload, userId);
- break;
- case "all":
- handleAllMessage(payload, userId);
- break;
+ try {
+ String userIdStr = (String) session.getAttributes().get("userId");
+ Long userId = Long.parseLong(userIdStr);
+ String payload = message.getPayload();
+ log.info("【cmd】Received message: {}", payload);
+ log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount());
+ BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class);
+ MessageType messageType = MessageType.fromString(baseMessage.getType());
+ switch (messageType) {
+ case STREAM:
+ handleStreamMessage(payload, userId);
+ break;
+ case STATE:
+ handleStateMessage(payload, userId);
+ break;
+ case TELEMETRY:
+ handleTelemetryMessage(payload, userId);
+ break;
+ case SIGNAL:
+ handleSignalMessage(payload, userId);
+ break;
+ case CONTROLLER:
+ handleControlMessage(payload, userId);
+ break;
+ case ALL:
+ handleAllMessage(payload, userId);
+ break;
+ }
+ } catch (Exception e) {
+ log.error("Error handling message: {}", message, e);
}
}
@@ -215,6 +222,7 @@
}
}
}
+
private void handleAllMessage(String payload, Long userId) {
// 解析接收到的JSON消息，将其转换为BaseMessage对象
BaseMessage<UavControlMessage> controlMessage = JSON.parseObject(payload,
@@ -238,6 +246,7 @@
}
}
}
+
@Override
public void afterConnectionEstablished(WebSocketSession session) {
log.info("【cmd】Connected: {}", session.getId());
diff --git a/ard-work/src/main/java/com/ard/tianyu/domain/enums/MessageType.java b/ard-work/src/main/java/com/ard/tianyu/domain/enums/MessageType.java
new file mode 100644
index 0000000..de16d3b
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/tianyu/domain/enums/MessageType.java
@@ -0,0 +1,29 @@
+package com.ard.tianyu.domain.enums;
+
+public enum MessageType {
+ STREAM("stream"),
+ STATE("state"),
+ TELEMETRY("telemetry"),
+ SIGNAL("signal"),
+ CONTROLLER("controller"),
+ 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);
+ }
+}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java b/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
index 60be244..cef7380 100644
--- a/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
+++ b/ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
@@ -34,7 +34,7 @@
import com.ard.tianyu.domain.dto.signal.SignalInfo;
import com.ard.tianyu.domain.dto.state.UavState;
import com.ard.tianyu.domain.dto.telemetry.TelemetryData;
-import com.ard.tianyu.component.netty.TyNettyTcpClient;
+import com.ard.tianyu.component.netty.NettyTcpClient;
import com.ard.tianyu.component.netty.command.control.AbsAngleControlCmd;
import com.ard.tianyu.component.netty.command.control.AngleControlCmd;
import com.ard.tianyu.component.netty.command.control.SpeedControlCmd;
@@ -827,7 +827,7 @@
*/
private void sendMessage(String key, BaseCmd msg) {
try {
- ChannelFuture future = TyNettyTcpClient.channels.get(key).getChannelFuture();
+ ChannelFuture future = NettyTcpClient.channels.get(key).getChannelFuture();
Channel channel = future.channel();
if (channel.isActive()) {
InetSocketAddress ipSocket = (InetSocketAddress) channel.remoteAddress();
--
Gitblit v1.9.3