优化tcp连接

liusuyi

2025-10-17 f9ee3fa619d478e5e592dfaa99a5b993feb6c67a

优化tcp连接

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

File was deleted

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

New file
@@ -0,0 +1,42 @@
package com.ard.work.component.netty;

import com.ard.work.component.netty.command.BaseCmd;
import com.ard.work.component.netty.cache.TyTcpServerCache;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;

/**
 * Description:
 * Author: lsy
 * Date: 2025年10月17日14:27
 **/
@Slf4j
public class TcpClientUtil {

    /**
     * 向指定边缘服务器发送消息
     */
    public static void sendMessage(String serverId, BaseCmd msg) {
        try {
            Channel channel = TyTcpServerCache.getChannel(serverId);
            if (channel == null) {
                log.error("未找到边缘服务器 [{}] 的连接", serverId);
                return;
            }

            if (!channel.isActive()) {
                log.warn("边缘服务器 [{}] 连接未激活，无法发送消息", serverId);
                return;
            }

            InetSocketAddress remote = (InetSocketAddress) channel.remoteAddress();
            channel.writeAndFlush(msg);
            log.warn("→ 向边缘服务器 [{}] ({}:{}) 发送消息，命令码：0x{}",
                    serverId, remote.getHostString(), remote.getPort(),
                    String.format("%02X", msg.getCmdCode() & 0xFF));
        } catch (Exception e) {
            log.error("发送消息到边缘服务器 [{}] 异常：{}", serverId, e.getMessage());
        }
    }
}

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

@@ -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();

@@ -75,7 +70,7 @@
        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();
@@ -83,6 +78,10 @@
                TyTcpServerCache.putTcpServerCache(serverId, true);
                // 断开服务器，修改状态为： 0 - 在线
                ardServerService.updateServerStatus(serverId, "1");

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

            } else {
                log.warn("❌ 服务器 [{}] 连接失败", serverId);
                TyTcpServerCache.putTcpServerCache(serverId, false);
@@ -90,59 +89,23 @@
                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 连接
@@ -172,6 +135,7 @@
            if (Boolean.FALSE.equals(state)) { // 连接失败的服务器
                ArdServerInfo serverInfo = ardServerService.selectArdServerInfoById(key);
                if (serverInfo != null) {
                    log.warn("服务器 [{}] 连接失败，尝试重连", key);
                    connectServer(serverInfo);
                }
            }
@@ -184,18 +148,14 @@
    @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);
                });
    }
}

 ard-work/src/main/java/com/ard/work/component/netty/cache/TyTcpServerCache.java

New file
@@ -0,0 +1,68 @@
package com.ard.work.component.netty.cache;

import io.netty.channel.Channel;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 六合tcp服务器缓存
 *
 * @author liusuyi
 * @date 2024年11月01日13:53
 **/
public class TyTcpServerCache {
    /**
     * 以 serverId 作为唯一 key 存储 Channel
     */
    private static Map<String, Channel> channels = new ConcurrentHashMap<>();

    //存储通道
    public static void putChannel(String serverId, Channel channel) {
        channels.put(serverId, channel);
    }

    //获取通道
    public static Channel getChannel(String serverId) {
        return channels.get(serverId);
    }

    //删除通道
    public static void removeChannel(String serverId) {
        channels.remove(serverId);
    }

    //获取所有
    public static Collection<Channel> getChannels() {
        return Collections.unmodifiableCollection(channels.values());
    }


    /**
     * 以 serverId 作为唯一 key 存储 连接状态
     */
    private static final ConcurrentHashMap<String, Boolean> tcpServerCache = new ConcurrentHashMap<>();

    // 存储TCP服务连接状态到缓存
    public static void putTcpServerCache(String serverId, Boolean state) {
        tcpServerCache.put(serverId, state);
    }

    // 从缓存中获取TCP服务连接状态
    public static Boolean getTcpServerCache(String serverId) {
        return tcpServerCache.getOrDefault(serverId, false);
    }

    // 删除缓存中的TCP服务连接状态
    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/component/netty/handler/ClientHandler.java

@@ -1,22 +1,19 @@
package com.ard.work.component.netty.handler;

import com.ard.work.component.netty.TyTcpClientManager;
import com.ard.work.component.netty.command.register.HeartbeatCmd;
import com.ard.work.domain.cache.TyTcpServerCache;
import com.ard.work.component.netty.command.BaseCmd;
import com.ard.work.component.netty.command.register.HeartbeatCmd;
import com.ard.work.component.netty.cache.TyTcpServerCache;
import com.ard.work.service.ITyService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;


@Component
@@ -26,9 +23,6 @@

    @Resource
    private ITyService tyService;

    @Resource
    private TyTcpClientManager tyTcpClientManager;

    @Override
    public void channelRead0(ChannelHandlerContext ctx, BaseCmd command) {

 ard-work/src/main/java/com/ard/work/domain/cache/TyTcpServerCache.java

File was deleted

 ard-work/src/main/java/com/ard/work/service/impl/TyServiceImpl.java

@@ -1,66 +1,61 @@
package com.ard.work.service.impl;

import com.ard.common.utils.ByteUtils;
import com.ard.work.aspect.annotation.CheckTakeoffAltitude;
import com.ard.work.aspect.annotation.CheckControlRight;
import com.ard.work.aspect.annotation.CheckTakeoffAltitude;
import com.ard.work.component.forest.TyClient;
import com.ard.work.component.netty.TcpClientUtil;
import com.ard.work.component.netty.command.BaseCmd;
import com.ard.work.component.netty.command.camera.*;
import com.ard.work.component.netty.command.manual.*;
import com.ard.work.component.netty.command.register.*;
import com.ard.work.component.netty.command.settings.*;
import com.ard.work.component.netty.command.speaker.*;
import com.ard.work.component.netty.command.wayline.*;
import com.ard.work.domain.dto.camera.*;
import com.ard.work.domain.dto.manual.*;
import com.ard.work.domain.dto.speaker.*;
import com.ard.work.domain.dto.wayline.*;
import com.ard.work.domain.entity.ArdDroneInfo;
import com.ard.work.domain.entity.ArdServerInfo;
import com.ard.work.domain.receiver.*;
import com.ard.work.component.forest.TyClient;
import com.ard.work.domain.cache.UavSignalInfoCache;
import com.ard.work.domain.cache.UavStateCache;
import com.ard.work.domain.cache.UavTelemetryDataCache;
import com.ard.work.domain.dto.BoxHistoryDTO;
import com.ard.work.domain.dto.BoxPhotosDTO;
import com.ard.work.domain.dto.BoxSettingsDTO;
import com.ard.work.domain.dto.control.AbsAngleControlDTO;
import com.ard.work.domain.dto.control.AngleControlDTO;
import com.ard.work.domain.dto.control.SpeedControlDTO;
import com.ard.work.domain.dto.converter.wayline.PlanLineDataConverter;
import com.ard.work.domain.dto.settings.ReturnHeightSettingDTO;
import com.ard.work.domain.dto.settings.ReturnPointSettingDTO;
import com.ard.work.domain.dto.settings.VideoStreamSettingDTO;
import com.ard.work.domain.dto.signal.SignalInfo;
import com.ard.work.domain.dto.state.UavState;
import com.ard.work.domain.dto.telemetry.TelemetryData;
import com.ard.work.component.netty.command.control.AbsAngleControlCmd;
import com.ard.work.component.netty.command.control.AngleControlCmd;
import com.ard.work.component.netty.command.control.SpeedControlCmd;
import com.ard.work.component.netty.command.manual.*;
import com.ard.work.component.netty.command.notice.BoxPowerStatusReplyCmd;
import com.ard.work.component.netty.command.notice.FlyWayEndReplyCmd;
import com.ard.work.component.netty.command.notice.UavReadyReplyCmd;
import com.ard.work.component.netty.command.real.SignalInfoCmd;
import com.ard.work.component.netty.command.real.UavStateCmd;
import com.ard.work.component.netty.command.real.UavTelemetryDataCmd;
import com.ard.work.component.netty.command.register.*;
import com.ard.work.component.netty.command.settings.*;
import com.ard.work.component.netty.command.speaker.*;
import com.ard.work.component.netty.command.wayline.*;
import com.ard.work.component.netty.handler.ErrorCodeEnum;
import com.ard.work.component.netty.protobuf.TelemetryDataPB;
import com.ard.work.domain.cache.UavSignalInfoCache;
import com.ard.work.domain.cache.UavStateCache;
import com.ard.work.domain.cache.UavTelemetryDataCache;
import com.ard.work.domain.dto.BoxHistoryDTO;
import com.ard.work.domain.dto.BoxPhotosDTO;
import com.ard.work.domain.dto.BoxSettingsDTO;
import com.ard.work.domain.dto.camera.*;
import com.ard.work.domain.dto.control.AbsAngleControlDTO;
import com.ard.work.domain.dto.control.AngleControlDTO;
import com.ard.work.domain.dto.control.SpeedControlDTO;
import com.ard.work.domain.dto.converter.wayline.PlanLineDataConverter;
import com.ard.work.domain.dto.manual.*;
import com.ard.work.domain.dto.settings.ReturnHeightSettingDTO;
import com.ard.work.domain.dto.settings.ReturnPointSettingDTO;
import com.ard.work.domain.dto.settings.VideoStreamSettingDTO;
import com.ard.work.domain.dto.signal.SignalInfo;
import com.ard.work.domain.dto.speaker.*;
import com.ard.work.domain.dto.state.UavState;
import com.ard.work.domain.dto.telemetry.TelemetryData;
import com.ard.work.domain.dto.wayline.*;
import com.ard.work.domain.entity.ArdDroneInfo;
import com.ard.work.domain.entity.ArdServerInfo;
import com.ard.work.domain.message.UavControlType;
import com.ard.work.domain.receiver.*;
import com.ard.work.service.IArdDroneInfoService;
import com.ard.work.service.IArdServerInfoService;
import com.ard.work.service.ITyService;
import com.ard.work.domain.message.UavControlType;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.ard.work.component.netty.TyTcpClientManager.channels;

/**
 * TyServiceImpl
@@ -900,25 +895,12 @@
                ErrorCodeEnum.getDescByCode(errorCode));
    }


    /**
     * 发送消息
     */
    private void sendMessage(String key, BaseCmd msg) {
        try {
            ChannelFuture future = channels.get(key).getChannelFuture();
            Channel channel = future.channel();
            if (channel.isActive()) {
                InetSocketAddress ipSocket = (InetSocketAddress) channel.remoteAddress();
                String host = ipSocket.getHostString();
                int port = ipSocket.getPort();
                channel.writeAndFlush(msg); // 发送编码后的数据
                log.debug("向六合服务器【{}:{}】发送数据，命令码：{}", host, port, String.format("0x%02X", msg.getCmdCode() & 0xFF));
            } else {
                log.error("与六合服务器链接断开，无法发送数据！");
            }
        } catch (Exception e) {
            log.error("发送消息异常：{}", e.getMessage());
        }
        TcpClientUtil.sendMessage(key, msg);
    }

}