增加无人机在线离线事件监听 增加飞行轨迹业务处理

liusuyi

2025-12-19 e0a6ad18bb69cdba48064327a439efca5673dcf1

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

@@ -6,6 +6,7 @@
import com.ard.work.component.netty.handler.MessageProtocolEncoder;
import com.ard.work.component.netty.cache.TyTcpServerCache;
import com.ard.work.domain.entity.ArdServerInfo;
import com.ard.work.service.IArdDroneInfoService;
import com.ard.work.service.IArdServerInfoService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
@@ -34,6 +35,9 @@
    private IArdServerInfoService ardServerService; // ← 从数据库读取边缘服务器配置

    @Resource
    private IArdDroneInfoService ardDroneService;

    @Resource
    private ClientHandler clientHandler;

    /**
@@ -49,7 +53,7 @@
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast("heart", new IdleStateHandler(0, 3, 0, TimeUnit.SECONDS));
                        pipeline.addLast("heart", new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
                        pipeline.addLast("encoder", new MessageProtocolEncoder());
                        pipeline.addLast("decoder", new MessageProtocolDecoder());
                        pipeline.addLast("handler", clientHandler);
@@ -76,8 +80,8 @@
                Channel channel = f.channel();
                channel.attr(AttributeKey.valueOf("serverId")).set(serverId);
                TyTcpServerCache.putTcpServerCache(serverId, true);
                // 断开服务器，修改状态为： 0 - 在线
                ardServerService.updateServerStatus(serverId, "1");
                // 修改状态为： 1 - 在线
                updateServerAndDroneStatus(serverId, "1");

                // 成功后再加入 channels
                TyTcpServerCache.putChannel(serverId, channel);
@@ -86,7 +90,7 @@
                log.warn("❌ 服务器 [{}] 连接失败", serverId);
                TyTcpServerCache.putTcpServerCache(serverId, false);
                // 断开服务器，修改状态为： 0 - 离线
                ardServerService.updateServerStatus(serverId, "0");
                updateServerAndDroneStatus(serverId, "0");
            }
        });
    }
@@ -100,12 +104,19 @@
            channel.close();
            log.info("断开服务器 [{}] 连接", serverId);
            // 断开服务器，修改状态为： 0 - 离线
            ardServerService.updateServerStatus(serverId, "0");
            updateServerAndDroneStatus(serverId, "0");
            TyTcpServerCache.removeChannel(serverId);
        }
        TyTcpServerCache.removeTcpServerCache(serverId);
    }

    /**
     * 同步修改服务和无人机状态
     */
    private void updateServerAndDroneStatus(String serverId, String status) {
        ardServerService.updateServerStatus(serverId, status);
        ardDroneService.updateDroneStatus(serverId, status);
    }

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

    @Scheduled(fixedDelay = 10000) // 每 10 秒检查一次
    @Scheduled(fixedDelay = 5000) // 每 5 秒检查一次
    public void reconnectFailedServers() {
        TyTcpServerCache.getTcpServerCacheMap().forEach((key, state) -> {
            if (Boolean.FALSE.equals(state)) { // 连接失败的服务器
@@ -143,9 +154,9 @@
    }

    /**
     * 定时向所有已连接的边缘服务器服务器发送心跳
     * 定时向所有已连接的边缘服务器服务器发送心跳（写空闲已经发心跳了，这里暂时取消）
     */
    @Scheduled(fixedRate = 5000) // 每5秒执行一次
   // @Scheduled(fixedRate = 5000) // 每5秒执行一次
    public void sendHeartBeat() {
        long timestamp = System.currentTimeMillis();
        TyTcpServerCache.getChannels().stream()
@@ -155,7 +166,7 @@
                    String serverId = (String) channel.attr(AttributeKey.valueOf("serverId")).get();
                    HeartbeatCmd heartbeat = new HeartbeatCmd(timestamp);
                    channel.writeAndFlush(heartbeat);
                    log.info("→ 向边缘服务器 [{}]发送心跳，时间戳={}", serverId, timestamp);
                    log.info("❤\uFE0F 向边缘服务器 [{}]发送心跳，时间戳={}", serverId, timestamp);
                });
    }
}