diff --git a/src/main/java/com/tongran/agent/client/netty/MultiTargetNettyClient.java b/src/main/java/com/tongran/agent/client/netty/MultiTargetNettyClient.java new file mode 100644 index 0000000..2b0aea0 --- /dev/null +++ b/src/main/java/com/tongran/agent/client/netty/MultiTargetNettyClient.java @@ -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 CONNECTION_KEY = AttributeKey.valueOf("connectionKey"); + + // 连接管理器 + private final Map connectionMap = new ConcurrentHashMap<>(); + private final Map 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() { + @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 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 { +// @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 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; + } +} \ No newline at end of file