1.修改服务器状态接口优化。 2.无人机盒子对象增加盒子别名字段。 3.动态修改服务状态。

Xiajijian

2025-10-16 56062b177a71cf781f8e480c0ec6f5e8394e7a8e

 ard-work/src/main/java/com/ard/work/component/netty/TyTcpClientManager.java

@@ -1,12 +1,13 @@
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.domain.entity.ArdServerInfo;
import com.ard.work.service.IArdServerService;
import com.ard.work.service.IArdServerInfoService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
@@ -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;
@@ -34,7 +36,7 @@
    private final EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

    @Resource
    private IArdServerService ardServerService; // ← 从数据库读取边缘服务器配置
    private IArdServerInfoService ardServerService; // ← 从数据库读取边缘服务器配置

    @Resource
    private ClientHandler clientHandler;
@@ -62,38 +64,33 @@
    }

    /**
     * 初始化与六合服务器的连接
     * 新增或重连服务器
     */
    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);
                // 断开服务器，修改状态为： 0 - 在线
                ardServerService.updateServerStatus(serverId, "1");
            } 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);
                // 断开服务器，修改状态为： 0 - 离线
                ardServerService.updateServerStatus(serverId, "0");
            }
        });

        // 存入连接缓存
        TcpClientSession session = TcpClientSession.builder()
                .ClientId("TY_" + serverId)
                .channelFuture(future)
@@ -104,6 +101,20 @@
                .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);
            // 断开服务器，修改状态为： 0 - 离线
            ardServerService.updateServerStatus(serverId, "0");
        }
        TyTcpServerCache.removeTcpServerCache(serverId);
    }

    /**
@@ -137,16 +148,16 @@
     * 启动时从数据库加载所有边缘服务器并建立 TCP 连接
     */
    @PostConstruct
    public void createAllTyConnections() {
    public void initConnectServers() {
        try {
            var serverList = ardServerService.selectArdServerManagementList(new ArdServerInfo()); // ← 从数据库查询所有服务器配置
            var serverList = ardServerService.selectArdServerInfoList(new ArdServerInfo()); // ← 从数据库查询所有服务器配置
            if (serverList == null || serverList.isEmpty()) {
                log.warn("未在数据库中找到边缘服务器配置，跳过连接初始化");
                return;
            }

            for (ArdServerInfo serverInfo : serverList) {
                initConnection(serverInfo.getId(), serverInfo.getHost(), serverInfo.getTcpPort());
                connectServer(serverInfo);
            }

            log.info("共初始化 {} 个边缘服务器连接", serverList.size());
@@ -154,4 +165,37 @@
            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.selectArdServerInfoById(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);
                });
    }
}