增加格式化存储卡接口

liusuyi

2025-11-24 2a729cd545ddbbab1eae6cbcf819783ef11da4ab

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

@@ -1,12 +1,12 @@
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.IArdDroneInfoService;
import com.ard.work.service.IArdServerInfoService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
@@ -18,25 +18,24 @@
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();

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

    @Resource
    private IArdDroneInfoService ardDroneService;

    @Resource
    private ClientHandler clientHandler;
@@ -54,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);
@@ -63,7 +62,9 @@
        return bootstrap;
    }

    /** 新增或重连服务器 */
    /**
     * 新增或重连服务器
     */
    public void connectServer(ArdServerInfo serverInfo) {
        String serverId = serverInfo.getId();
        String host = serverInfo.getHost();
@@ -73,65 +74,48 @@
        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);
                // 修改状态为： 1 - 在线
                updateServerAndDroneStatus(serverId, "1");

                // 成功后再加入 channels
                TyTcpServerCache.putChannel(serverId, channel);

            } else {
                log.warn("❌ 服务器 [{}] 连接失败", serverId);
                TyTcpServerCache.putTcpServerCache(serverId, false);
                // 断开服务器，修改状态为： 0 - 离线
                updateServerAndDroneStatus(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 - 离线
            updateServerAndDroneStatus(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());
        }
    private void updateServerAndDroneStatus(String serverId, String status) {
        ardServerService.updateServerStatus(serverId, status);
        ardDroneService.updateDroneStatus(serverId, status);
    }

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

    /**
     * 定时向所有已连接的六合服务器发送心跳
     * 定时向所有已连接的边缘服务器服务器发送心跳（写空闲已经发心跳了，这里暂时取消）
     */
    @Scheduled(fixedRate = 5000) // 每5秒执行一次
   // @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("❤\uFE0F 向边缘服务器 [{}]发送心跳，时间戳={}", serverId, timestamp);
                });
    }

}