forked from uav/ard-uav-server
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); }); } } 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); }
@@ -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); // 返回一个副本，避免外部修改原始缓存 } } 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; } /** 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) {
