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连接
---
ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java | 101 +++++++++++++++++---------------------------------
1 files changed, 35 insertions(+), 66 deletions(-)
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 3a1ec4b..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();
@@ -63,7 +58,9 @@
return bootstrap;
}
- /** 新增或重连服务器 */
+ /**
+ * 新增或重连服务器
+ */
public void connectServer(ArdServerInfo serverInfo) {
String serverId = serverInfo.getId();
String host = serverInfo.getHost();
@@ -73,66 +70,42 @@
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();
channel.attr(AttributeKey.valueOf("serverId")).set(serverId);
TyTcpServerCache.putTcpServerCache(serverId, true);
+ // 断开服务器，修改状态为： 0 - 在线
+ ardServerService.updateServerStatus(serverId, "1");
+
+ // 成功后再加入 channels
+ TyTcpServerCache.putChannel(serverId, channel);
+
} else {
log.warn("❌ 服务器 [{}] 连接失败", serverId);
TyTcpServerCache.putTcpServerCache(serverId, false);
+ // 断开服务器，修改状态为： 0 - 离线
+ 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 连接
@@ -140,7 +113,7 @@
@PostConstruct
public void initConnectServers() {
try {
- var serverList = ardServerService.selectArdServerManagementList(new ArdServerInfo()); // ← 从数据库查询所有服务器配置
+ var serverList = ardServerService.selectArdServerInfoList(new ArdServerInfo()); // ← 从数据库查询所有服务器配置
if (serverList == null || serverList.isEmpty()) {
log.warn("未在数据库中找到边缘服务器配置，跳过连接初始化");
return;
@@ -160,8 +133,9 @@
public void reconnectFailedServers() {
TyTcpServerCache.getTcpServerCacheMap().forEach((key, state) -> {
if (Boolean.FALSE.equals(state)) { // 连接失败的服务器
- ArdServerInfo serverInfo = ardServerService.selectArdServerManagementById(key);
+ ArdServerInfo serverInfo = ardServerService.selectArdServerInfoById(key);
if (serverInfo != null) {
+ log.warn("服务器 [{}] 连接失败，尝试重连", key);
connectServer(serverInfo);
}
}
@@ -174,19 +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);
});
}
-
}
--
Gitblit v1.9.3