From 8aa9e121554b84735c8ec6139bd095ea3acb05e7 Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Tue, 14 Oct 2025 16:10:49 +0800
Subject: [PATCH] 更改六合边缘服务器动态连接
---
ard-work/src/main/java/com/ard/work/domain/cache/TyTcpServerCache.java | 21 +++++--
ard-work/src/main/java/com/ard/work/service/impl/ArdServerServiceImpl.java | 25 +++++++-
ard-work/src/main/java/com/ard/work/service/impl/TyServiceImpl.java | 21 -------
ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java | 89 ++++++++++++++++++++---------
ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java | 11 ---
5 files changed, 100 insertions(+), 67 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 fccf3db..7fbf709 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,6 +1,7 @@
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;
@@ -15,6 +16,7 @@
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
@@ -61,39 +63,28 @@
return bootstrap;
}
- /**
- * 初始化与六合服务器的连接
- */
- public void initConnection(String serverId, String host, Integer port) {
- String key = host + ":" + port;
- log.info("初始化连接边缘服务器 [{}] ({})", serverId, key);
+ /** 新增或重连服务器 */
+ public void connectServer(ArdServerInfo serverInfo) {
+ String serverId = serverInfo.getId();
+ String host = serverInfo.getHost();
+ Integer port = serverInfo.getTcpPort();
- // 如果存在旧连接则关闭
- if (channels.containsKey(serverId)) {
- TcpClientSession oldSession = channels.get(serverId);
- if (oldSession != null && oldSession.getChannelFuture() != null) {
- oldSession.getChannelFuture().channel().close();
- }
- }
+ // 如果已有旧连接，先断开
+ disconnectServer(serverId);
- // 异步连接 TCP 服务端
Bootstrap bootstrap = getBootstrap();
- ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture futureListener) -> {
- if (futureListener.isSuccess()) {
- log.info("✅ 边缘服务器 [{}] 连接成功 ({})", serverId, key);
- Channel channel = futureListener.channel();
- // 绑定 serverId 到 channel
+ ChannelFuture future = 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(key, true);
+ TyTcpServerCache.putTcpServerCache(serverId, true);
} else {
- log.warn("❌ 边缘服务器 [{}] 连接失败 ({})，10 秒后重试", serverId, key);
- TyTcpServerCache.putTcpServerCache(key, false);
- final EventLoop eventLoop = futureListener.channel().eventLoop();
- eventLoop.schedule(() -> initConnection(serverId, host, port), 10, TimeUnit.SECONDS);
+ log.warn("❌ 服务器 [{}] 连接失败", serverId);
+ TyTcpServerCache.putTcpServerCache(serverId, false);
}
});
- // 存入连接缓存
TcpClientSession session = TcpClientSession.builder()
.ClientId("TY_" + serverId)
.channelFuture(future)
@@ -104,6 +95,16 @@
.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();
+ log.info("断开服务器 [{}] 连接", serverId);
+ }
+ TyTcpServerCache.removeTcpServerCache(serverId);
}
/**
@@ -137,7 +138,7 @@
* 启动时从数据库加载所有边缘服务器并建立 TCP 连接
*/
@PostConstruct
- public void createAllTyConnections() {
+ public void initConnectServers() {
try {
var serverList = ardServerService.selectArdServerManagementList(new ArdServerInfo()); // ← 从数据库查询所有服务器配置
if (serverList == null || serverList.isEmpty()) {
@@ -146,7 +147,7 @@
}
for (ArdServerInfo serverInfo : serverList) {
- initConnection(serverInfo.getId(), serverInfo.getHost(), serverInfo.getTcpPort());
+ connectServer(serverInfo);
}
log.info("共初始化 {} 个边缘服务器连接", serverList.size());
@@ -154,4 +155,38 @@
log.error("创建边缘服务器连接异常：{}", e.getMessage(), e);
}
}
+
+ @Scheduled(fixedDelay = 10000) // 每 10 秒检查一次
+ public void reconnectFailedServers() {
+ TyTcpServerCache.getTcpServerCacheMap().forEach((key, state) -> {
+ if (Boolean.FALSE.equals(state)) { // 连接失败的服务器
+ ArdServerInfo serverInfo = ardServerService.selectArdServerManagementById(key);
+ if (serverInfo != null) {
+ connectServer(serverInfo);
+ }
+ }
+ });
+ }
+
+ /**
+ * 定时向所有已连接的六合服务器发送心跳
+ */
+ @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 -> {
+ HeartbeatCmd heartbeat = new HeartbeatCmd(timestamp);
+ ChannelFuture future = session.getChannelFuture();
+ future.channel().writeAndFlush(heartbeat);
+ log.info("→ 向边缘服务器 [{}] ({}:{}) 发送心跳，时间戳={}",
+ session.getServerId(), session.getHost(), session.getPort(), timestamp);
+ });
+ }
+
}
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 c3eab84..86f6795 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
@@ -93,16 +93,7 @@
String host = remoteAddress.getHostString();
int port = remoteAddress.getPort();
log.error("与六合服务{}:{}连接断开!", host, port);
- TyTcpServerCache.putTcpServerCache(host + ":" + port, false);
- ctx.channel().eventLoop().schedule(() -> {
- log.error("六合服务端链接不上，开始重连操作...");
- try {
- tyTcpClientManager.initConnection(serverId,host, port);
- } catch (Exception e) {
- log.error("六合服务端连接异常: {}", e.getMessage());
- }
- }, 5, TimeUnit.SECONDS);
-
+ TyTcpServerCache.putTcpServerCache(serverId, false);
super.channelInactive(ctx);
}
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
index ec7d9d6..e03a482 100644
--- 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
@@ -1,5 +1,9 @@
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;
/**
@@ -13,17 +17,22 @@
private static final ConcurrentHashMap<String, Boolean> tcpServerCache = new ConcurrentHashMap<>();
// 存储TCP服务连接状态到缓存
- public static void putTcpServerCache(String tcpServer,Boolean state) {
- tcpServerCache.put(tcpServer, state);
+ public static void putTcpServerCache(String serverId,Boolean state) {
+ tcpServerCache.put(serverId, state);
}
// 从缓存中获取TCP服务连接状态
- public static Boolean getTcpServerCache(String tcpServer) {
- return tcpServerCache.getOrDefault(tcpServer,false);
+ public static Boolean getTcpServerCache(String serverId) {
+ return tcpServerCache.getOrDefault(serverId,false);
}
// 删除缓存中的TCP服务连接状态
- public static void removeTcpServerCache(String tcpServer) {
- tcpServerCache.remove(tcpServer);
+ 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/ArdServerServiceImpl.java b/ard-work/src/main/java/com/ard/work/service/impl/ArdServerServiceImpl.java
index a86bdeb..dddd4ab 100644
--- a/ard-work/src/main/java/com/ard/work/service/impl/ArdServerServiceImpl.java
+++ b/ard-work/src/main/java/com/ard/work/service/impl/ArdServerServiceImpl.java
@@ -2,6 +2,8 @@
import com.ard.common.utils.uuid.IdUtils;
import com.ard.work.component.forest.TyClient;
+import com.ard.work.component.netty.TyTcpClientManager;
+import com.ard.work.domain.cache.TyTcpServerCache;
import com.ard.work.domain.dto.TokenDTO;
import com.ard.work.domain.entity.ArdServerInfo;
import com.ard.work.domain.receiver.AccessTokenReceiver;
@@ -31,6 +33,8 @@
private ArdServerInfoMapper ardServerInfoMapper;
@Resource
private TyClient tyClient;
+ @Resource
+ private TyTcpClientManager tyTcpClientManager;
/**
* 查询服务器管理
@@ -63,7 +67,11 @@
@Override
public int insertArdServerManagement(ArdServerInfo ardServerInfo) {
ardServerInfo.setId(IdUtils.fastUUID());
- return ardServerInfoMapper.insertArdServerManagement(ardServerInfo);
+ int i = ardServerInfoMapper.insertArdServerManagement(ardServerInfo);
+ if (i > 0) {
+ tyTcpClientManager.connectServer(ardServerInfo);
+ }
+ return i;
}
/**
@@ -74,7 +82,11 @@
*/
@Override
public int updateArdServerManagement(ArdServerInfo ardServerInfo) {
- return ardServerInfoMapper.updateArdServerManagement(ardServerInfo);
+ int i = ardServerInfoMapper.updateArdServerManagement(ardServerInfo);
+ if (i > 0) {
+ tyTcpClientManager.connectServer(ardServerInfo);
+ }
+ return i;
}
/**
@@ -85,7 +97,14 @@
*/
@Override
public int deleteArdServerManagementByIds(String[] ids) {
- return ardServerInfoMapper.deleteArdServerManagementByIds(ids);
+ int i = ardServerInfoMapper.deleteArdServerManagementByIds(ids);
+ if (i > 0) {
+ for (String id : ids) {
+ //移除tcp连接
+ tyTcpClientManager.disconnectServer(id);
+ }
+ }
+ return i;
}
/**
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 cc586d0..d27d7ed 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
@@ -883,27 +883,6 @@
/**
- * 定时向所有已连接的六合服务器发送心跳
- */
- @Scheduled(fixedRate = 5000) // 每5秒执行一次
- public void sendHeartBeat() {
- long timestamp = System.currentTimeMillis();
-
- for (TcpClientSession session : channels.values()) {
- ChannelFuture future = session.getChannelFuture();
- if (future == null || !future.channel().isActive()) {
- log.warn("边缘服务器 [{}] 连接未激活，跳过心跳发送", session.getServerId());
- continue;
- }
-
- HeartbeatCmd heartbeat = new HeartbeatCmd(timestamp);
- future.channel().writeAndFlush(heartbeat);
-
- log.info("→ 向边缘服务器 [{}] ({}:{}) 发送心跳，时间戳={}",
- session.getServerId(), session.getHost(), session.getPort(), timestamp);
- }
- }
- /**
* 发送消息
*/
private void sendMessage(String key, BaseCmd msg) {
--
Gitblit v1.9.3