~nishiwei/ard-uav-server.git

			@@ -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 连接
			@@ -162,6 +135,7 @@
			if (Boolean.FALSE.equals(state)) { // 连接失败的服务器
			ArdServerInfo serverInfo = ardServerService.selectArdServerInfoById(key);
			if (serverInfo != null) {
			log.warn("服务器 [{}] 连接失败，尝试重连", key);
			connectServer(serverInfo);
			}
			}
			@@ -169,24 +143,19 @@
			}

			/**
			* 定时向所有已连接的六合服务器发送心跳
			* 定时向所有已连接的边缘服务器服务器发送心跳
			*/
			@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);
			});
			}

			}