v1.0.2初始化
This commit is contained in:
qiminbao
@@ -0,0 +1,301 @@
|
||||
package com.tongran.agent.client.netty;
|
||||
|
||||
import com.tongran.agent.client.netty.handler.AgentDecoderHandler;
|
||||
import com.tongran.agent.client.netty.handler.AgentDispatcherHandler;
|
||||
import com.tongran.agent.client.netty.handler.AgentEncoderHandler;
|
||||
import com.tongran.agent.client.netty.model.Message;
|
||||
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.util.AttributeKey;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
@Component
|
||||
public class MultiTargetNettyClient {
|
||||
// 连接标识属性键
|
||||
private static final AttributeKey<String> CONNECTION_KEY = AttributeKey.valueOf("connectionKey");
|
||||
|
||||
// 连接管理器
|
||||
private final Map<String, Channel> connectionMap = new ConcurrentHashMap<>();
|
||||
private final Map<Channel, String> reverseConnectionMap = new ConcurrentHashMap<>();
|
||||
private EventLoopGroup workerGroup;
|
||||
|
||||
@Resource
|
||||
private AgentDecoderHandler decoderHandler;
|
||||
|
||||
@Resource
|
||||
private AgentEncoderHandler encoderHandler;
|
||||
|
||||
@Resource
|
||||
private AgentDispatcherHandler dispatcherHandler;
|
||||
|
||||
/**
|
||||
* 动态创建TCP连接
|
||||
* @param connectionKey 连接唯一标识（格式：ip:port 或自定义）
|
||||
* @param host 目标主机
|
||||
* @param port 目标端口
|
||||
* @param timeout 超时时间（秒）
|
||||
* @return 是否创建成功
|
||||
*/
|
||||
public boolean createConnection(String connectionKey, String host, int port, int timeout) {
|
||||
if (workerGroup == null) {
|
||||
workerGroup = new NioEventLoopGroup();
|
||||
}
|
||||
|
||||
if (connectionMap.containsKey(connectionKey)) {
|
||||
System.out.println("连接已存在: " + connectionKey);
|
||||
return true;
|
||||
}
|
||||
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
bootstrap.group(workerGroup)
|
||||
.channel(NioSocketChannel.class)
|
||||
.option(ChannelOption.TCP_NODELAY, true)
|
||||
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout * 1000)
|
||||
.handler(new ChannelInitializer<SocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(SocketChannel ch) {
|
||||
ChannelPipeline pipeline = ch.pipeline();
|
||||
// 添加编解码器
|
||||
pipeline.addLast(decoderHandler);
|
||||
pipeline.addLast(encoderHandler);
|
||||
// 添加心跳机制
|
||||
// pipeline.addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
|
||||
// 添加自定义处理器
|
||||
pipeline.addLast(dispatcherHandler);
|
||||
}
|
||||
});
|
||||
|
||||
try {
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
final boolean[] success = {false};
|
||||
|
||||
bootstrap.connect(host, port).addListener((ChannelFuture future) -> {
|
||||
if (future.isSuccess()) {
|
||||
Channel channel = future.channel();
|
||||
connectionMap.put(connectionKey, channel);
|
||||
reverseConnectionMap.put(channel, connectionKey);
|
||||
success[0] = true;
|
||||
System.out.println("连接建立成功: " + connectionKey + " -> " + host + ":" + port);
|
||||
} else {
|
||||
System.err.println("连接建立失败: " + connectionKey + " - " + future.cause().getMessage());
|
||||
success[0] = false;
|
||||
}
|
||||
latch.countDown();
|
||||
});
|
||||
|
||||
// 等待连接结果
|
||||
return latch.await(timeout, TimeUnit.SECONDS) && success[0];
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 向指定连接发送消息
|
||||
* @param connectionKey 连接标识
|
||||
* @param message 消息内容
|
||||
* @return 是否发送成功
|
||||
*/
|
||||
public boolean sendMessage(String connectionKey, String message) {
|
||||
Channel channel = connectionMap.get(connectionKey);
|
||||
if (channel == null) {
|
||||
System.err.println("连接不存在: " + connectionKey);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!channel.isActive()) {
|
||||
System.err.println("连接已断开: " + connectionKey);
|
||||
removeConnection(connectionKey);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
ChannelFuture future = channel.writeAndFlush(message).sync();
|
||||
System.out.println("消息发送成功到: " + connectionKey + " - " + message);
|
||||
return future.isSuccess();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public boolean sendMessages(String connectionKey, Message message) {
|
||||
Channel channel = connectionMap.get(connectionKey);
|
||||
if (channel == null) {
|
||||
System.err.println("连接不存在: " + connectionKey);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!channel.isActive()) {
|
||||
System.err.println("连接已断开: " + connectionKey);
|
||||
removeConnection(connectionKey);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
ChannelFuture future = channel.writeAndFlush(message).sync();
|
||||
System.out.println("消息发送成功到: " + connectionKey + " - " + message);
|
||||
return future.isSuccess();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量发送消息到不同连接
|
||||
* @param messages Map<连接标识, 消息内容>
|
||||
*/
|
||||
public void sendMessages(Map<String, String> messages) {
|
||||
messages.forEach((connectionKey, message) -> {
|
||||
new Thread(() -> {
|
||||
if (sendMessage(connectionKey, message)) {
|
||||
System.out.println("批量发送成功: " + connectionKey);
|
||||
} else {
|
||||
System.err.println("批量发送失败: " + connectionKey);
|
||||
}
|
||||
}).start();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭指定连接
|
||||
* @param connectionKey 连接标识
|
||||
*/
|
||||
public void closeConnection(String connectionKey) {
|
||||
Channel channel = connectionMap.get(connectionKey);
|
||||
if (channel != null) {
|
||||
channel.close();
|
||||
removeConnection(connectionKey);
|
||||
System.out.println("连接已关闭: " + connectionKey);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 移除连接记录
|
||||
* @param connectionKey 连接标识
|
||||
*/
|
||||
private void removeConnection(String connectionKey) {
|
||||
Channel channel = connectionMap.remove(connectionKey);
|
||||
if (channel != null) {
|
||||
reverseConnectionMap.remove(channel);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭所有连接
|
||||
*/
|
||||
public void shutdown() {
|
||||
// 关闭所有连接
|
||||
connectionMap.forEach((key, channel) -> {
|
||||
if (channel.isActive()) {
|
||||
channel.close();
|
||||
}
|
||||
});
|
||||
|
||||
connectionMap.clear();
|
||||
reverseConnectionMap.clear();
|
||||
|
||||
// 关闭EventLoopGroup
|
||||
if (workerGroup != null) {
|
||||
workerGroup.shutdownGracefully();
|
||||
}
|
||||
System.out.println("所有连接已关闭");
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取连接状态
|
||||
* @return 当前活跃连接数
|
||||
*/
|
||||
public int getActiveConnections() {
|
||||
return (int) connectionMap.values().stream()
|
||||
.filter(Channel::isActive)
|
||||
.count();
|
||||
}
|
||||
|
||||
// /**
|
||||
// * 客户端处理器
|
||||
// */
|
||||
// private class ClientHandler extends SimpleChannelInboundHandler<String> {
|
||||
// @Override
|
||||
// protected void channelRead0(ChannelHandlerContext ctx, String msg) {
|
||||
// String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
|
||||
// System.out.println("收到来自 " + connectionKey + " 的消息: " + msg);
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
|
||||
// String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
|
||||
// System.err.println("连接 " + connectionKey + " 发生异常: " + cause.getMessage());
|
||||
// ctx.close();
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void channelInactive(ChannelHandlerContext ctx) {
|
||||
// String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
|
||||
// System.out.println("连接断开: " + connectionKey);
|
||||
// removeConnection(connectionKey);
|
||||
// }
|
||||
// }
|
||||
|
||||
/**
|
||||
* 使用示例
|
||||
*/
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
MultiTargetNettyClient client = new MultiTargetNettyClient();
|
||||
|
||||
try {
|
||||
// 动态创建多个不同IP和端口的连接
|
||||
boolean success1 = client.createConnection("server1", "127.0.0.1", 8080, 5);
|
||||
boolean success2 = client.createConnection("server2", "192.168.1.100", 8081, 5);
|
||||
boolean success3 = client.createConnection("server3", "10.0.0.50", 8082, 5);
|
||||
|
||||
// 等待连接建立
|
||||
Thread.sleep(2000);
|
||||
|
||||
if (success1) {
|
||||
client.sendMessage("server1", "Hello Server 1 from 8080");
|
||||
}
|
||||
|
||||
if (success2) {
|
||||
client.sendMessage("server2", "Hello Server 2 from 8081");
|
||||
}
|
||||
|
||||
if (success3) {
|
||||
client.sendMessage("server3", "Hello Server 3 from 8082");
|
||||
}
|
||||
|
||||
// // 批量发送示例
|
||||
// Map<String, String> batchMessages = Map.of(
|
||||
// "server1", "Batch message to server 1",
|
||||
// "server2", "Batch message to server 2",
|
||||
// "server3", "Batch message to server 3"
|
||||
// );
|
||||
//
|
||||
// client.sendMessages(batchMessages);
|
||||
|
||||
// 保持运行一段时间
|
||||
Thread.sleep(5000);
|
||||
|
||||
System.out.println("当前活跃连接数: " + client.getActiveConnections());
|
||||
|
||||
} finally {
|
||||
client.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
public Channel get(String connectionKey){
|
||||
Channel channel = connectionMap.get(connectionKey);
|
||||
return channel;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user