From 0d9fb1b9336da142d070109782a8c260744f5e57 Mon Sep 17 00:00:00 2001
From: zhaowenyu <35589726@qq.com>
Date: Mon, 08 Dec 2025 12:16:01 +0800
Subject: [PATCH] Merge remote-tracking branch 'origin/dev' into dev
---
ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java | 84 +++++++++++++++++++++++-------------------
1 files changed, 46 insertions(+), 38 deletions(-)
diff --git a/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java b/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java
index 33298be..0613572 100644
--- a/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java
+++ b/ard-work/src/main/java/com/ard/work/component/netty/handler/ClientHandler.java
@@ -1,19 +1,19 @@
package com.ard.work.component.netty.handler;
-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.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
@@ -24,22 +24,27 @@
@Resource
private ITyService tyService;
- @Resource
- private NettyTcpClient nettyTcpClient;
-
@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 +55,40 @@
}
@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.debug("读空闲超时 - 可能服务器未响应");
+ ctx.close(); // <-- 必须加这句
+ break;
+ case WRITER_IDLE:
+ // 写空闲时主动发心跳
+ HeartbeatCmd heartbeat = new HeartbeatCmd(timestamp);
+ ctx.writeAndFlush(heartbeat);
+ log.debug("写空闲 -> 发送心跳包 时间戳={}", timestamp);
+ break;
+ case ALL_IDLE:
+ log.debug("读写都空闲 - 可考虑关闭连接或重连");
+ 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);
}
--
Gitblit v1.9.3