From f9ee3fa619d478e5e592dfaa99a5b993feb6c67a Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Fri, 17 Oct 2025 15:06:19 +0800
Subject: [PATCH] 优化tcp连接
---
/dev/null | 38 ------
ard-work/src/main/java/com/ard/work/component/netty/TcpClientUtil.java | 42 +++++++
ard-work/src/main/java/com/ard/work/component/netty/cache/TyTcpServerCache.java | 68 +++++++++++
ard-work/src/main/java/com/ard/work/service/impl/TyServiceImpl.java | 86 +++++--------
ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java | 82 +++----------
ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java | 10 -
6 files changed, 167 insertions(+), 159 deletions(-)
diff --git a/ard-work/src/main/java/com/ard/work/component/netty/TcpClientSession.java b/ard-work/src/main/java/com/ard/work/component/netty/TcpClientSession.java
deleted file mode 100644
index 3ae32c8..0000000
--- a/ard-work/src/main/java/com/ard/work/component/netty/TcpClientSession.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package com.ard.work.component.netty;
-
-import io.netty.channel.ChannelFuture;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@Builder
-@AllArgsConstructor
-@NoArgsConstructor
-public class TcpClientSession {
-
- private String serverId;
-
- private String host;
-
- private Integer port;
-
- public String ClientId;
-
- public ChannelFuture channelFuture;
-
- public Boolean isReConnect;
-
-}
diff --git a/ard-work/src/main/java/com/ard/work/component/netty/TcpClientUtil.java b/ard-work/src/main/java/com/ard/work/component/netty/TcpClientUtil.java
new file mode 100644
index 0000000..d886bba
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/work/component/netty/TcpClientUtil.java
@@ -0,0 +1,42 @@
+package com.ard.work.component.netty;
+
+import com.ard.work.component.netty.command.BaseCmd;
+import com.ard.work.component.netty.cache.TyTcpServerCache;
+import io.netty.channel.Channel;
+import lombok.extern.slf4j.Slf4j;
+import java.net.InetSocketAddress;
+
+/**
+ * Description:
+ * Author: lsy
+ * Date: 2025年10月17日14:27
+ **/
+@Slf4j
+public class TcpClientUtil {
+
+ /**
+ * 向指定边缘服务器发送消息
+ */
+ public static void sendMessage(String serverId, BaseCmd msg) {
+ try {
+ Channel channel = TyTcpServerCache.getChannel(serverId);
+ if (channel == null) {
+ log.error("未找到边缘服务器 [{}] 的连接", serverId);
+ return;
+ }
+
+ if (!channel.isActive()) {
+ log.warn("边缘服务器 [{}] 连接未激活，无法发送消息", serverId);
+ return;
+ }
+
+ InetSocketAddress remote = (InetSocketAddress) channel.remoteAddress();
+ channel.writeAndFlush(msg);
+ log.warn("→ 向边缘服务器 [{}] ({}:{}) 发送消息，命令码：0x{}",
+ serverId, remote.getHostString(), remote.getPort(),
+ String.format("%02X", msg.getCmdCode() & 0xFF));
+ } catch (Exception e) {
+ log.error("发送消息到边缘服务器 [{}] 异常：{}", serverId, e.getMessage());
+ }
+ }
+}
diff --git a/ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java b/ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java
index a52c955..2060ae7 100644
--- a/ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java
+++ b/ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java
@@ -1,11 +1,10 @@
package com.ard.work.component.netty;
-import com.ard.work.component.netty.command.BaseCmd;
import com.ard.work.component.netty.command.register.HeartbeatCmd;
import com.ard.work.component.netty.handler.ClientHandler;
import com.ard.work.component.netty.handler.MessageProtocolDecoder;
import com.ard.work.component.netty.handler.MessageProtocolEncoder;
-import com.ard.work.domain.cache.TyTcpServerCache;
+import com.ard.work.component.netty.cache.TyTcpServerCache;
import com.ard.work.domain.entity.ArdServerInfo;
import com.ard.work.service.IArdServerInfoService;
import io.netty.bootstrap.Bootstrap;
@@ -18,20 +17,16 @@
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
+
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
-import java.net.InetSocketAddress;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+
@Component
@Slf4j
public class TyTcpClientManager {
- /**
- * 以 serverId 作为唯一 key
- */
- public static final Map<String, TcpClientSession> channels = new ConcurrentHashMap<>();
+
private final EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
@@ -75,7 +70,7 @@
disconnectServer(serverId);
Bootstrap bootstrap = getBootstrap();
- ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture f) -> {
+ bootstrap.connect(host, port).addListener((ChannelFuture f) -> {
if (f.isSuccess()) {
log.info("✅ 服务器 [{}] 连接成功", serverId);
Channel channel = f.channel();
@@ -83,6 +78,10 @@
TyTcpServerCache.putTcpServerCache(serverId, true);
// 断开服务器，修改状态为： 0 - 在线
ardServerService.updateServerStatus(serverId, "1");
+
+ // 成功后再加入 channels
+ TyTcpServerCache.putChannel(serverId, channel);
+
} else {
log.warn("❌ 服务器 [{}] 连接失败", serverId);
TyTcpServerCache.putTcpServerCache(serverId, false);
@@ -90,59 +89,23 @@
ardServerService.updateServerStatus(serverId, "0");
}
});
-
- TcpClientSession session = TcpClientSession.builder()
- .ClientId("TY_" + serverId)
- .channelFuture(future)
- .isReConnect(true)
- .serverId(serverId)
- .host(host)
- .port(port)
- .build();
-
- channels.put(serverId, session);
}
/**
* 断开服务器连接
*/
public void disconnectServer(String serverId) {
- TcpClientSession session = channels.remove(serverId);
- if (session != null && session.getChannelFuture() != null) {
- session.getChannelFuture().channel().close();
+ Channel channel = TyTcpServerCache.getChannel(serverId);
+ if (channel != null) {
+ channel.close();
log.info("断开服务器 [{}] 连接", serverId);
// 断开服务器，修改状态为： 0 - 离线
ardServerService.updateServerStatus(serverId, "0");
+ TyTcpServerCache.removeChannel(serverId);
}
TyTcpServerCache.removeTcpServerCache(serverId);
}
- /**
- * 向指定边缘服务器发送消息
- */
- public void sendMessage(Long serverId, BaseCmd msg) {
- try {
- TcpClientSession session = channels.get(serverId);
- if (session == null) {
- log.error("未找到六合服务器 [{}] 的连接", serverId);
- return;
- }
-
- ChannelFuture future = session.getChannelFuture();
- if (future == null || !future.channel().isActive()) {
- log.warn("六合服务器 [{}] 连接未激活，无法发送消息", serverId);
- return;
- }
-
- InetSocketAddress remote = (InetSocketAddress) future.channel().remoteAddress();
- future.channel().writeAndFlush(msg);
- log.debug("→ 向六合服务器 [{}] ({}:{}) 发送消息，命令码：0x{}",
- serverId, remote.getHostString(), remote.getPort(),
- String.format("%02X", msg.getCmdCode() & 0xFF));
- } catch (Exception e) {
- log.error("发送消息到六合服务器 [{}] 异常：{}", serverId, e.getMessage());
- }
- }
/**
* 启动时从数据库加载所有边缘服务器并建立 TCP 连接
@@ -172,6 +135,7 @@
if (Boolean.FALSE.equals(state)) { // 连接失败的服务器
ArdServerInfo serverInfo = ardServerService.selectArdServerInfoById(key);
if (serverInfo != null) {
+ log.warn("服务器 [{}] 连接失败，尝试重连", key);
connectServer(serverInfo);
}
}
@@ -184,18 +148,14 @@
@Scheduled(fixedRate = 5000) // 每5秒执行一次
public void sendHeartBeat() {
long timestamp = System.currentTimeMillis();
- channels.values().stream()
- .filter(session -> {
- ChannelFuture future = session.getChannelFuture();
- // 只保留活跃连接
- return future != null && future.channel().isActive();
- })
- .forEach(session -> {
+ TyTcpServerCache.getChannels().stream()
+ // 过滤出仍然活跃的连接
+ .filter(channel -> channel != null && channel.isActive())
+ .forEach(channel -> {
+ String serverId = (String) channel.attr(AttributeKey.valueOf("serverId")).get();
HeartbeatCmd heartbeat = new HeartbeatCmd(timestamp);
- ChannelFuture future = session.getChannelFuture();
- future.channel().writeAndFlush(heartbeat);
- log.info("→ 向边缘服务器 [{}] ({}:{}) 发送心跳，时间戳={}",
- session.getServerId(), session.getHost(), session.getPort(), timestamp);
+ channel.writeAndFlush(heartbeat);
+ log.info("→ 向边缘服务器 [{}]发送心跳，时间戳={}", serverId, timestamp);
});
}
}
diff --git a/ard-work/src/main/java/com/ard/work/component/netty/cache/TyTcpServerCache.java b/ard-work/src/main/java/com/ard/work/component/netty/cache/TyTcpServerCache.java
new file mode 100644
index 0000000..aec481d
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/work/component/netty/cache/TyTcpServerCache.java
@@ -0,0 +1,68 @@
+package com.ard.work.component.netty.cache;
+
+import io.netty.channel.Channel;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 六合tcp服务器缓存
+ *
+ * @author liusuyi
+ * @date 2024年11月01日13:53
+ **/
+public class TyTcpServerCache {
+ /**
+ * 以 serverId 作为唯一 key 存储 Channel
+ */
+ private static Map<String, Channel> channels = new ConcurrentHashMap<>();
+
+ //存储通道
+ public static void putChannel(String serverId, Channel channel) {
+ channels.put(serverId, channel);
+ }
+
+ //获取通道
+ public static Channel getChannel(String serverId) {
+ return channels.get(serverId);
+ }
+
+ //删除通道
+ public static void removeChannel(String serverId) {
+ channels.remove(serverId);
+ }
+
+ //获取所有
+ public static Collection<Channel> getChannels() {
+ return Collections.unmodifiableCollection(channels.values());
+ }
+
+
+ /**
+ * 以 serverId 作为唯一 key 存储 连接状态
+ */
+ private static final ConcurrentHashMap<String, Boolean> tcpServerCache = new ConcurrentHashMap<>();
+
+ // 存储TCP服务连接状态到缓存
+ public static void putTcpServerCache(String serverId, Boolean state) {
+ tcpServerCache.put(serverId, state);
+ }
+
+ // 从缓存中获取TCP服务连接状态
+ public static Boolean getTcpServerCache(String serverId) {
+ return tcpServerCache.getOrDefault(serverId, false);
+ }
+
+ // 删除缓存中的TCP服务连接状态
+ public static void removeTcpServerCache(String serverId) {
+ tcpServerCache.remove(serverId);
+ }
+
+ public static Map<String, Boolean> getTcpServerCacheMap() {
+ return new HashMap<>(tcpServerCache); // 返回一个副本，避免外部修改原始缓存
+ }
+
+}
diff --git a/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java b/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java
index 86f6795..afc3752 100644
--- a/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java
+++ b/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java
@@ -1,22 +1,19 @@
package com.ard.work.component.netty.handler;
-import com.ard.work.component.netty.TyTcpClientManager;
-import com.ard.work.component.netty.command.register.HeartbeatCmd;
-import com.ard.work.domain.cache.TyTcpServerCache;
import com.ard.work.component.netty.command.BaseCmd;
+import com.ard.work.component.netty.command.register.HeartbeatCmd;
+import com.ard.work.component.netty.cache.TyTcpServerCache;
import com.ard.work.service.ITyService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
-import java.util.concurrent.TimeUnit;
@Component
@@ -26,9 +23,6 @@
@Resource
private ITyService tyService;
-
- @Resource
- private TyTcpClientManager tyTcpClientManager;
@Override
public void channelRead0(ChannelHandlerContext ctx, BaseCmd command) {
diff --git a/ard-work/src/main/java/com/ard/work/domain/cache/TyTcpServerCache.java b/ard-work/src/main/java/com/ard/work/domain/cache/TyTcpServerCache.java
deleted file mode 100644
index e03a482..0000000
--- a/ard-work/src/main/java/com/ard/work/domain/cache/TyTcpServerCache.java
+++ /dev/null
@@ -1,38 +0,0 @@
-package com.ard.work.domain.cache;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * 六合tcp服务器缓存
- *
- * @author liusuyi
- * @date 2024年11月01日13:53
- **/
-public class TyTcpServerCache {
- // 使用 ConcurrentHashMap 来缓存TCP服务连接状态，key 是 ip:port（服务器地址） value 是连接状态
- private static final ConcurrentHashMap<String, Boolean> tcpServerCache = new ConcurrentHashMap<>();
-
- // 存储TCP服务连接状态到缓存
- public static void putTcpServerCache(String serverId,Boolean state) {
- tcpServerCache.put(serverId, state);
- }
-
- // 从缓存中获取TCP服务连接状态
- public static Boolean getTcpServerCache(String serverId) {
- return tcpServerCache.getOrDefault(serverId,false);
- }
-
- // 删除缓存中的TCP服务连接状态
- public static void removeTcpServerCache(String serverId) {
- tcpServerCache.remove(serverId);
- }
-
- public static Map<String, Boolean> getTcpServerCacheMap() {
- return new HashMap<>(tcpServerCache); // 返回一个副本，避免外部修改原始缓存
- }
-
-}
diff --git a/ard-work/src/main/java/com/ard/work/service/impl/TyServiceImpl.java b/ard-work/src/main/java/com/ard/work/service/impl/TyServiceImpl.java
index 810d866..39eb921 100644
--- a/ard-work/src/main/java/com/ard/work/service/impl/TyServiceImpl.java
+++ b/ard-work/src/main/java/com/ard/work/service/impl/TyServiceImpl.java
@@ -1,66 +1,61 @@
package com.ard.work.service.impl;
import com.ard.common.utils.ByteUtils;
-import com.ard.work.aspect.annotation.CheckTakeoffAltitude;
import com.ard.work.aspect.annotation.CheckControlRight;
+import com.ard.work.aspect.annotation.CheckTakeoffAltitude;
+import com.ard.work.component.forest.TyClient;
+import com.ard.work.component.netty.TcpClientUtil;
import com.ard.work.component.netty.command.BaseCmd;
import com.ard.work.component.netty.command.camera.*;
-import com.ard.work.component.netty.command.manual.*;
-import com.ard.work.component.netty.command.register.*;
-import com.ard.work.component.netty.command.settings.*;
-import com.ard.work.component.netty.command.speaker.*;
-import com.ard.work.component.netty.command.wayline.*;
-import com.ard.work.domain.dto.camera.*;
-import com.ard.work.domain.dto.manual.*;
-import com.ard.work.domain.dto.speaker.*;
-import com.ard.work.domain.dto.wayline.*;
-import com.ard.work.domain.entity.ArdDroneInfo;
-import com.ard.work.domain.entity.ArdServerInfo;
-import com.ard.work.domain.receiver.*;
-import com.ard.work.component.forest.TyClient;
-import com.ard.work.domain.cache.UavSignalInfoCache;
-import com.ard.work.domain.cache.UavStateCache;
-import com.ard.work.domain.cache.UavTelemetryDataCache;
-import com.ard.work.domain.dto.BoxHistoryDTO;
-import com.ard.work.domain.dto.BoxPhotosDTO;
-import com.ard.work.domain.dto.BoxSettingsDTO;
-import com.ard.work.domain.dto.control.AbsAngleControlDTO;
-import com.ard.work.domain.dto.control.AngleControlDTO;
-import com.ard.work.domain.dto.control.SpeedControlDTO;
-import com.ard.work.domain.dto.converter.wayline.PlanLineDataConverter;
-import com.ard.work.domain.dto.settings.ReturnHeightSettingDTO;
-import com.ard.work.domain.dto.settings.ReturnPointSettingDTO;
-import com.ard.work.domain.dto.settings.VideoStreamSettingDTO;
-import com.ard.work.domain.dto.signal.SignalInfo;
-import com.ard.work.domain.dto.state.UavState;
-import com.ard.work.domain.dto.telemetry.TelemetryData;
import com.ard.work.component.netty.command.control.AbsAngleControlCmd;
import com.ard.work.component.netty.command.control.AngleControlCmd;
import com.ard.work.component.netty.command.control.SpeedControlCmd;
+import com.ard.work.component.netty.command.manual.*;
import com.ard.work.component.netty.command.notice.BoxPowerStatusReplyCmd;
import com.ard.work.component.netty.command.notice.FlyWayEndReplyCmd;
import com.ard.work.component.netty.command.notice.UavReadyReplyCmd;
import com.ard.work.component.netty.command.real.SignalInfoCmd;
import com.ard.work.component.netty.command.real.UavStateCmd;
import com.ard.work.component.netty.command.real.UavTelemetryDataCmd;
+import com.ard.work.component.netty.command.register.*;
+import com.ard.work.component.netty.command.settings.*;
+import com.ard.work.component.netty.command.speaker.*;
+import com.ard.work.component.netty.command.wayline.*;
import com.ard.work.component.netty.handler.ErrorCodeEnum;
import com.ard.work.component.netty.protobuf.TelemetryDataPB;
+import com.ard.work.domain.cache.UavSignalInfoCache;
+import com.ard.work.domain.cache.UavStateCache;
+import com.ard.work.domain.cache.UavTelemetryDataCache;
+import com.ard.work.domain.dto.BoxHistoryDTO;
+import com.ard.work.domain.dto.BoxPhotosDTO;
+import com.ard.work.domain.dto.BoxSettingsDTO;
+import com.ard.work.domain.dto.camera.*;
+import com.ard.work.domain.dto.control.AbsAngleControlDTO;
+import com.ard.work.domain.dto.control.AngleControlDTO;
+import com.ard.work.domain.dto.control.SpeedControlDTO;
+import com.ard.work.domain.dto.converter.wayline.PlanLineDataConverter;
+import com.ard.work.domain.dto.manual.*;
+import com.ard.work.domain.dto.settings.ReturnHeightSettingDTO;
+import com.ard.work.domain.dto.settings.ReturnPointSettingDTO;
+import com.ard.work.domain.dto.settings.VideoStreamSettingDTO;
+import com.ard.work.domain.dto.signal.SignalInfo;
+import com.ard.work.domain.dto.speaker.*;
+import com.ard.work.domain.dto.state.UavState;
+import com.ard.work.domain.dto.telemetry.TelemetryData;
+import com.ard.work.domain.dto.wayline.*;
+import com.ard.work.domain.entity.ArdDroneInfo;
+import com.ard.work.domain.entity.ArdServerInfo;
+import com.ard.work.domain.message.UavControlType;
+import com.ard.work.domain.receiver.*;
import com.ard.work.service.IArdDroneInfoService;
import com.ard.work.service.IArdServerInfoService;
import com.ard.work.service.ITyService;
-import com.ard.work.domain.message.UavControlType;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
-
import javax.annotation.Resource;
-import java.net.InetSocketAddress;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-
-import static com.ard.work.component.netty.TyTcpClientManager.channels;
/**
* TyServiceImpl
@@ -900,25 +895,12 @@
ErrorCodeEnum.getDescByCode(errorCode));
}
+
/**
* 发送消息
*/
private void sendMessage(String key, BaseCmd msg) {
- try {
- ChannelFuture future = channels.get(key).getChannelFuture();
- Channel channel = future.channel();
- if (channel.isActive()) {
- InetSocketAddress ipSocket = (InetSocketAddress) channel.remoteAddress();
- String host = ipSocket.getHostString();
- int port = ipSocket.getPort();
- channel.writeAndFlush(msg); // 发送编码后的数据
- log.debug("向六合服务器【{}:{}】发送数据，命令码：{}", host, port, String.format("0x%02X", msg.getCmdCode() & 0xFF));
- } else {
- log.error("与六合服务器链接断开，无法发送数据！");
- }
- } catch (Exception e) {
- log.error("发送消息异常：{}", e.getMessage());
- }
+ TcpClientUtil.sendMessage(key, msg);
}
}
\ No newline at end of file
--
Gitblit v1.9.3