初始化
This commit is contained in:
qiminbao
@@ -0,0 +1,127 @@
|
||||
package com.tongran.agentserver.server.netty;
|
||||
|
||||
import com.tongran.agentserver.server.netty.config.AgentNettyConfig;
|
||||
import com.tongran.agentserver.server.netty.handler.DecoderHandler;
|
||||
import com.tongran.agentserver.server.netty.handler.NettyClientHandler;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.*;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.SocketChannel;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.codec.string.StringEncoder;
|
||||
import io.netty.handler.timeout.IdleStateHandler;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.annotation.PreDestroy;
|
||||
import javax.annotation.Resource;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
@Component
|
||||
public class NettyTcpClient {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(NettyTcpClient.class);
|
||||
|
||||
private EventLoopGroup group = new NioEventLoopGroup();
|
||||
|
||||
private Channel channel;
|
||||
|
||||
@Resource
|
||||
private AgentNettyConfig config;
|
||||
|
||||
// 目标服务器IP和端口
|
||||
// private final String host = "www.tz2.xyz"; // 替换为实际IP
|
||||
// private final int port = 6610; // 替换为实际端口
|
||||
|
||||
/**
|
||||
* 启动客户端连接
|
||||
*/
|
||||
@PostConstruct
|
||||
public void start() {
|
||||
try {
|
||||
connect();
|
||||
} catch (Exception e) {
|
||||
logger.error("Netty客户端启动失败", e);
|
||||
}
|
||||
}
|
||||
|
||||
public void connect() throws InterruptedException {
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
bootstrap.group(group)
|
||||
.channel(NioSocketChannel.class)
|
||||
.option(ChannelOption.TCP_NODELAY, true)
|
||||
.option(ChannelOption.SO_KEEPALIVE, true)
|
||||
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
|
||||
.handler(new ChannelInitializer<SocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(SocketChannel ch) {
|
||||
ChannelPipeline pipeline = ch.pipeline();
|
||||
// 添加编解码器
|
||||
pipeline.addLast("decoder", new DecoderHandler());
|
||||
pipeline.addLast("encoder", new StringEncoder());
|
||||
// 添加心跳机制
|
||||
pipeline.addLast("idleStateHandler",
|
||||
new IdleStateHandler(0, 0, 90, TimeUnit.SECONDS));
|
||||
// 添加自定义处理器
|
||||
pipeline.addLast("clientHandler", new NettyClientHandler());
|
||||
}
|
||||
});
|
||||
|
||||
// 连接服务器
|
||||
ChannelFuture future = bootstrap.connect(config.getHost(), config.getPort()).sync();
|
||||
|
||||
// 监听连接状态
|
||||
future.addListener((ChannelFutureListener) f -> {
|
||||
if (f.isSuccess()) {
|
||||
logger.info("成功连接到TCP服务器 {}:{}", config.getHost(), config.getPort());
|
||||
channel = f.channel();
|
||||
} else {
|
||||
logger.error("连接TCP服务器失败 {}:{}, 5秒后尝试重连...", config.getHost(), config.getPort());
|
||||
f.channel().eventLoop().schedule(() -> {
|
||||
try {
|
||||
connect();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}, 5, TimeUnit.SECONDS);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 发送消息
|
||||
* @param message 要发送的消息内容
|
||||
*/
|
||||
public void sendMessage(String message) {
|
||||
if (channel != null && channel.isActive()) {
|
||||
channel.writeAndFlush(message);
|
||||
logger.info("发送消息: {}", message);
|
||||
} else {
|
||||
logger.warn("TCP连接未建立，消息发送失败: {}", message);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭连接
|
||||
*/
|
||||
@PreDestroy
|
||||
public void stop() {
|
||||
if (channel != null) {
|
||||
channel.close();
|
||||
}
|
||||
group.shutdownGracefully();
|
||||
logger.info("Netty客户端已关闭");
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查连接状态
|
||||
* @return true表示连接正常
|
||||
*/
|
||||
public boolean isConnected() {
|
||||
return channel != null && channel.isActive();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package com.tongran.agentserver.server.netty.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
@Data
|
||||
@Configuration
|
||||
@ConfigurationProperties(prefix = "netty.server")
|
||||
public class AgentNettyConfig {
|
||||
|
||||
private String host;
|
||||
|
||||
private int port;
|
||||
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package com.tongran.agentserver.server.netty.handler;
|
||||
|
||||
import cn.hutool.cache.Cache;
|
||||
import cn.hutool.cache.CacheUtil;
|
||||
import com.tongran.agentserver.utils.AssertLog;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
||||
import io.netty.util.CharsetUtil;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Component
|
||||
@ChannelHandler.Sharable
|
||||
public class DecoderHandler extends ChannelInboundHandlerAdapter {
|
||||
|
||||
|
||||
|
||||
// 用来临时保留没有处理过的请求报文
|
||||
private final Cache<String, ByteBuf> lruCache = CacheUtil.newLRUCache(3000);
|
||||
|
||||
/**
|
||||
* 消息解码器
|
||||
*/
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
||||
// 检查是否为 ByteBuf（未解码的原始数据）
|
||||
if (msg instanceof ByteBuf) {
|
||||
ByteBuf byteBuf = (ByteBuf) msg;
|
||||
String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码
|
||||
AssertLog.info("<<[up]:[up-content]==>{}", messages);
|
||||
|
||||
// JSONObject jsonObject = JSONObject.parseObject(messages);
|
||||
// String clientId = jsonObject.getString("clientId");
|
||||
// Message electricMessage = Message.builder().build();
|
||||
// electricMessage.setClientId(clientId);
|
||||
// JSONObject object = new JSONObject();
|
||||
// object.put("dateType","register");
|
||||
// object.put("stats","1");
|
||||
// electricMessage.setContent(object.toString());
|
||||
// ctx.fireChannelRead(electricMessage);//传递到下一个handler
|
||||
|
||||
byteBuf.release(); // 释放 ByteBuf 资源（重要！）
|
||||
} else {
|
||||
System.out.println("Unexpected message type: " + msg.getClass());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public byte[] subByte(byte[] b, int off, int length) {
|
||||
byte[] bytes = new byte[length];
|
||||
System.arraycopy(b, off, bytes, 0, length);
|
||||
return bytes;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,214 @@
|
||||
package com.tongran.agentserver.server.netty.handler;
|
||||
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import com.tongran.agentserver.server.netty.model.Message;
|
||||
import com.tongran.agentserver.utils.AgentUtil;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.*;
|
||||
import io.netty.channel.socket.SocketChannel;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.codec.string.StringEncoder;
|
||||
import io.netty.handler.timeout.IdleState;
|
||||
import io.netty.handler.timeout.IdleStateEvent;
|
||||
import io.netty.util.AttributeKey;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
@ChannelHandler.Sharable
|
||||
public class NettyClientHandler extends SimpleChannelInboundHandler<Message> {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);
|
||||
|
||||
// 目标服务器 IP 和端口（从 Channel 的 RemoteAddress 动态获取）
|
||||
private String serverIp;
|
||||
private int serverPort;
|
||||
|
||||
// 重连状态标记（原子变量保证线程安全）
|
||||
private AtomicBoolean isReconnecting = new AtomicBoolean(false);
|
||||
private int retryCount = 0; // 当前重试次数
|
||||
|
||||
// 重连策略配置（可根据需求调整）
|
||||
private int maxRetries = 3; // 最大重试次数（3次）
|
||||
private long initialInterval = 1000; // 初始重连间隔（1秒）
|
||||
private long intervalMultiplier = 2; // 间隔倍数（指数退避）
|
||||
private long maxInterval = 4000; // 最大重连间隔（4秒）
|
||||
|
||||
// 异步调度器（独立线程池，避免阻塞 Netty I/O 线程）
|
||||
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
|
||||
|
||||
// AttributeKey 用于传递重连上下文（可选）
|
||||
private static final AttributeKey<NettyClientHandler> RECONNECT_HANDLER_KEY =
|
||||
AttributeKey.valueOf("nettyClientHandler");
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
|
||||
logger.info("收到服务器消息: {}", msg);
|
||||
// 这里可以添加消息处理逻辑
|
||||
}
|
||||
|
||||
@Override
|
||||
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
|
||||
if (evt instanceof IdleStateEvent) {
|
||||
IdleStateEvent event = (IdleStateEvent) evt;
|
||||
if (event.state() == IdleState.WRITER_IDLE) {
|
||||
// 发送心跳包
|
||||
JSONObject object = new JSONObject();
|
||||
object.put("strength","31");
|
||||
String clientId = AgentUtil.getMotherboardUUID();
|
||||
Message message = Message.builder().clientId(clientId).dataType("HEARTBEAT").data(object.toString()).build();
|
||||
// 将对象转为 JSON 字符串
|
||||
String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran";
|
||||
System.out.println(json);
|
||||
ctx.writeAndFlush(json);
|
||||
logger.debug("发送心跳包");
|
||||
}
|
||||
} else {
|
||||
super.userEventTriggered(ctx, evt);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelActive(ChannelHandlerContext ctx) {
|
||||
logger.info("与服务器建立连接: {}", ctx.channel().remoteAddress());
|
||||
// 连接建立时，记录目标服务器 IP 和端口（从 RemoteAddress 解析）
|
||||
InetSocketAddress remoteAddr = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||
this.serverIp = remoteAddr.getHostString();
|
||||
this.serverPort = remoteAddr.getPort();
|
||||
System.out.println("连接建立成功，目标服务器: " + serverIp + ":" + serverPort);
|
||||
retryCount = 0; // 连接成功时重置重试次数
|
||||
|
||||
// 连接建立后可以发送登录认证消息
|
||||
String clientId = AgentUtil.getMotherboardUUID();
|
||||
JSONObject object = new JSONObject();
|
||||
object.put("uuid",clientId);
|
||||
Message message = Message.builder().clientId(clientId).dataType("LOGIN").data(object.toString()).build();
|
||||
// 将对象转为 JSON 字符串
|
||||
String json = "agent-tcp:"+JSON.toJSONString(message)+"@tong-ran";
|
||||
System.out.println("连接建立成功，发送登录认证="+json);
|
||||
ctx.writeAndFlush(json);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelInactive(ChannelHandlerContext ctx) {
|
||||
logger.warn("与服务器断开连接: {}", ctx.channel().remoteAddress());
|
||||
// 连接断开时触发重连逻辑
|
||||
System.out.println("连接断开，触发重连...");
|
||||
if (isReconnecting.compareAndSet(false, true)) {
|
||||
startReconnectTask(ctx); // 启动异步重连任务
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动异步重连任务（指数退避策略）
|
||||
*/
|
||||
private void startReconnectTask(ChannelHandlerContext ctx) {
|
||||
// 检查是否超过最大重试次数
|
||||
if (retryCount >= maxRetries) {
|
||||
System.out.println("已达到最大重试次数（" + maxRetries + "），停止重连");
|
||||
isReconnecting.set(false);
|
||||
return;
|
||||
}
|
||||
|
||||
// 计算当前重连间隔（指数退避公式：initialInterval * (intervalMultiplier ^ retryCount)）
|
||||
long currentInterval = calculateCurrentInterval();
|
||||
System.out.printf("第 %d 次重连，将在 %dms 后尝试...%n", retryCount + 1, currentInterval);
|
||||
|
||||
// 调度重连任务（使用独立线程池）
|
||||
ScheduledFuture<?> reconnectFuture = scheduler.schedule(() -> {
|
||||
try {
|
||||
// 关闭已断开的 Channel（避免资源泄漏）
|
||||
if (ctx.channel().isActive()) {
|
||||
ctx.channel().close().sync();
|
||||
}
|
||||
|
||||
// 重新初始化 Bootstrap 并连接服务器
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
bootstrap.group(ctx.channel().eventLoop()) // 复用原 EventLoopGroup
|
||||
.channel(NioSocketChannel.class)
|
||||
.handler(new ChannelInitializer<SocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(SocketChannel ch) {
|
||||
// 重新添加编解码器和业务处理器（与初始连接一致）
|
||||
ch.pipeline().addLast(new DecoderHandler()); // 自定义解码器
|
||||
ch.pipeline().addLast(new StringEncoder()); // 自定义编码器
|
||||
ch.pipeline().addLast(new NettyClientHandler()); // 绑定重连处理器
|
||||
// 其他业务处理器...
|
||||
}
|
||||
});
|
||||
|
||||
// 执行连接（阻塞直到完成）
|
||||
ChannelFuture connectFuture = bootstrap.connect(serverIp, serverPort).sync();
|
||||
Channel newChannel = connectFuture.channel();
|
||||
|
||||
// 重连成功后的处理
|
||||
onReconnectSuccess(newChannel);
|
||||
} catch (Exception e) {
|
||||
// 重连失败处理
|
||||
onReconnectFailure(ctx,e);
|
||||
}
|
||||
}, currentInterval, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* 计算当前重连间隔（指数退避）
|
||||
*/
|
||||
private long calculateCurrentInterval() {
|
||||
int currentRetry = retryCount;
|
||||
long interval = (long) (initialInterval * Math.pow(intervalMultiplier, currentRetry));
|
||||
return Math.min(interval, maxInterval); // 限制最大间隔
|
||||
}
|
||||
|
||||
/**
|
||||
* 重连成功回调
|
||||
*/
|
||||
private void onReconnectSuccess(Channel newChannel) {
|
||||
System.out.println("重连成功！新 Channel ID: " + newChannel.id().asLongText());
|
||||
// 重置重连状态
|
||||
retryCount = 0;
|
||||
isReconnecting.set(false);
|
||||
// 可选：同步业务状态（如会话恢复）
|
||||
}
|
||||
|
||||
/**
|
||||
* 重连失败回调
|
||||
*/
|
||||
private void onReconnectFailure(ChannelHandlerContext ctx,Exception e) {
|
||||
retryCount++; // 重试次数+1
|
||||
System.err.printf("重连失败（第 %d 次）: %s%n", retryCount, e.getMessage());
|
||||
|
||||
// 检查是否超过最大重试次数
|
||||
if (retryCount >= maxRetries) {
|
||||
System.out.println("已达到最大重试次数（" + maxRetries + "），停止重连");
|
||||
isReconnecting.set(false);
|
||||
return;
|
||||
}
|
||||
|
||||
// 继续下一次重连（触发 channelInactive 重新调度）
|
||||
if (ctx != null) {
|
||||
startReconnectTask(ctx);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
|
||||
logger.error("通信异常", cause);
|
||||
ctx.close();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handlerRemoved(ChannelHandlerContext ctx) {
|
||||
// 清理资源（如关闭调度器）
|
||||
scheduler.shutdown();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package com.tongran.agentserver.server.netty.model;
|
||||
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* @author egrias
|
||||
*/
|
||||
@Data
|
||||
@Builder
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public class Message implements Serializable {
|
||||
|
||||
private static final long serialVersionUID = -1267013167162440610L;
|
||||
|
||||
/**
|
||||
* clientId
|
||||
*/
|
||||
private String clientId;
|
||||
|
||||
/**
|
||||
* 数据类型：LOGIN、HEARTBEAT、CPU、MEMORY、SYSTEM、POINT、NET、DISK、DOCKER、SWITCHBOARD
|
||||
*/
|
||||
private String dataType;
|
||||
|
||||
/**
|
||||
* 发送内容
|
||||
*/
|
||||
private String data;
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user