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.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
|
@Slf4j
|
@ChannelHandler.Sharable
|
public class ClientHandler extends SimpleChannelInboundHandler<BaseCmd> {
|
|
@Resource
|
private ITyService tyService;
|
|
@Resource
|
private TyTcpClientManager tyTcpClientManager;
|
|
@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.debug("写空闲 -> 发送心跳包 时间戳={}", 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(host + ":" + port, false);
|
ctx.channel().eventLoop().schedule(() -> {
|
log.error("六合服务端链接不上，开始重连操作...");
|
try {
|
tyTcpClientManager.initConnection(serverId,host, port);
|
} catch (Exception e) {
|
log.error("六合服务端连接异常: {}", e.getMessage());
|
}
|
}, 5, TimeUnit.SECONDS);
|
|
super.channelInactive(ctx);
|
}
|
|
}
|
