Merge remote-tracking branch 'origin/dev' into dev

zhaowenyu

2025-10-14 f905f4377db462640a4fdd53dc5daa1adfc8c542

 ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java

@@ -1,14 +1,17 @@
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.NettyTcpClient;
import com.ard.work.component.netty.command.BaseCmd;
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;
@@ -25,21 +28,29 @@
    private ITyService tyService;

    @Resource
    private NettyTcpClient nettyTcpClient;
    private TyTcpClientManager tyTcpClientManager;

    @Override
    public void channelRead0(ChannelHandlerContext ctx, BaseCmd command) {
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String host = ipSocket.getHostString();
        int port = ipSocket.getPort();
        tyService.processReplyMessage(host + ":" + port, command);
        Channel channel = ctx.channel();
        String serverId = (String) channel.attr(AttributeKey.valueOf("serverId")).get();
        tyService.processReplyMessage(serverId, command);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //当客户端connect成功后，可以在最后一个handler等待用户输入
        log.info("连接成功建立！");
        tyService.registerMessage();
        Channel channel = ctx.channel();
        String serverId = (String) channel.attr(AttributeKey.valueOf("serverId")).get();

        InetSocketAddress remoteAddress = (InetSocketAddress) channel.remoteAddress();
        String host = remoteAddress.getHostString();
        int port = remoteAddress.getPort();

        log.info("与边缘服务器 [{}] 连接成功 ({}:{})", serverId, host, port);

        // 针对当前连接的服务器进行注册
        tyService.registerMessage(serverId);
        super.channelActive(ctx);
    }

@@ -50,37 +61,39 @@
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object obj) {
        if (obj instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) obj;
            if (event.state() == IdleState.READER_IDLE) {
                log.info("客户端读超时");
            } else if (event.state() == IdleState.WRITER_IDLE) {
              //  log.info("写空闲发送⼼跳!");
               // HeartbeatCommand msg = new HeartbeatCommand(System.currentTimeMillis());
                //ctx.writeAndFlush(msg);
            } else if (event.state() == IdleState.ALL_IDLE) {
                log.info("客户端所有操作超时");
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            long timestamp = System.currentTimeMillis();

            switch (event.state()) {
                case READER_IDLE:
                    log.warn("读空闲超时 - 可能服务器未响应");
                    break;
                case WRITER_IDLE:
                    // 写空闲时主动发心跳
                    HeartbeatCmd heartbeat = new HeartbeatCmd(timestamp);
                    ctx.writeAndFlush(heartbeat);
                    log.debug("写空闲 -> 发送心跳包 时间戳={}", timestamp);
                    break;
                case ALL_IDLE:
                    log.info("读写都空闲 - 可考虑关闭连接或重连");
                    break;
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        int port = ipSocket.getPort();
        String host = ipSocket.getHostString();
        log.error("与天宇服务{}:{}连接断开!", host, port);
        TyTcpServerCache.putTcpServerCache(host + ":" + port, false);
        ctx.channel().eventLoop().schedule(() -> {
            log.error("天宇服务端链接不上，开始重连操作...");
            try {
                nettyTcpClient.initConnection(host, port);
            } catch (Exception e) {
                log.error("天宇服务端连接异常: {}", e.getMessage());
            }
        }, 5, TimeUnit.SECONDS);

        Channel channel = ctx.channel();
        String serverId = (String) channel.attr(AttributeKey.valueOf("serverId")).get();
        InetSocketAddress remoteAddress = (InetSocketAddress) channel.remoteAddress();
        String host = remoteAddress.getHostString();
        int port = remoteAddress.getPort();
        log.error("与六合服务{}:{}连接断开!", host, port);
        TyTcpServerCache.putTcpServerCache(serverId, false);
        super.channelInactive(ctx);
    }