package com.ard.work.component.netty.handler; 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.IdleStateEvent; import io.netty.util.AttributeKey; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.net.InetSocketAddress; @Component @Slf4j @ChannelHandler.Sharable public class ClientHandler extends SimpleChannelInboundHandler { @Resource private ITyService tyService; @Override public void channelRead0(ChannelHandlerContext ctx, BaseCmd 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等待用户输入 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); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { log.error("发生异常: ", cause); // 使用日志记录异常 // ctx.close(); } @Override 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.info("写空闲 -> 发送心跳包 时间戳={}", timestamp); break; case ALL_IDLE: log.info("读写都空闲 - 可考虑关闭连接或重连"); break; } } else { super.userEventTriggered(ctx, evt); } } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { 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); } }