第一个基本可用版本

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/ProxyServer.java

@@ -0,0 +1,40 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server;
+
+import fun.asgc.neutrino.core.annotation.NeutrinoApplication;
+import fun.asgc.neutrino.core.launcher.NeutrinoLauncher;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@NeutrinoApplication
+public class ProxyServer {
+
+	public static void main(String[] args) {
+		NeutrinoLauncher.runSync(ProxyServer.class, args);
+	}
+
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/config/ProxyConfig.java

@@ -0,0 +1,78 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.config;
+
+import fun.asgc.neutrino.core.annotation.Configuration;
+import fun.asgc.neutrino.core.annotation.Value;
+import lombok.Data;
+
+import java.util.Map;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Data
+@Configuration(prefix = "proxy")
+public class ProxyConfig {
+	private Protocol protocol;
+	private Server server;
+	@Value("license")
+	private Map<String, Integer> licenseMap;
+
+	@Data
+	public static class Protocol {
+		@Value("max-frame-length")
+		private Integer maxFrameLength;
+		@Value("length-field-offset")
+		private Integer lengthFieldOffset;
+		@Value("length-field-length")
+		private Integer lengthFieldLength;
+		@Value("initial-bytes-to-strip")
+		private Integer initialBytesToStrip;
+		@Value("length-adjustment")
+		private Integer lengthAdjustment;
+		@Value("read-idle-time")
+		private Integer readIdleTime;
+		@Value("write-idle-time")
+		private Integer writeIdleTime;
+		@Value("all-idle-time-seconds")
+		private Integer allIdleTimeSeconds;
+	}
+
+	@Data
+	public static class Server {
+		@Value("port")
+		private Integer port;
+		@Value("ssl-port")
+		private Integer sslPort;
+		@Value("key-store-password")
+		private String keyStorePassword;
+		@Value("key-manager-password")
+		private String keyManagerPassword;
+		@Value("jks-path")
+		private String jksPath;
+	}
+
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/config/ProxyServerConfig.java

@@ -0,0 +1,125 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.config;
+
+import fun.asgc.neutrino.proxy.core.ProxyClientConfig;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.*;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+public class ProxyServerConfig implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 更新配置后保证在其他线程即时生效
+     */
+    private static ProxyServerConfig instance = new ProxyServerConfig();;
+
+    /**
+     * 代理服务器为各个代理客户端（key）开启对应的端口列表（value）
+     */
+    private volatile Map<String, List<Integer>> clientInetPortMapping = new HashMap<String, List<Integer>>();
+
+    /**
+     * 代理服务器上的每个对外端口（key）对应的代理客户端背后的真实服务器信息（value）
+     */
+    private volatile Map<Integer, String> inetPortLanInfoMapping = new HashMap<Integer, String>();
+
+    public void addClientConfig(ProxyClientConfig clientConfig) {
+        String clientKey = clientConfig.getClientKey();
+        List<Integer> ports = new ArrayList<>();
+        for (ProxyClientConfig.Proxy proxy : clientConfig.getProxy()) {
+            ports.add(proxy.getServerPort());
+            inetPortLanInfoMapping.put(proxy.getServerPort(), proxy.getClientInfo());
+        }
+        clientInetPortMapping.put(clientKey, ports);
+    }
+
+    /**
+     * 获取代理客户端对应的代理服务器端口
+     *
+     * @param clientKey
+     * @return
+     */
+    public List<Integer> getClientInetPorts(String clientKey) {
+        return clientInetPortMapping.get(clientKey);
+    }
+
+    /**
+     * 根据代理服务器端口获取后端服务器代理信息
+     *
+     * @param port
+     * @return
+     */
+    public String getLanInfo(Integer port) {
+        return inetPortLanInfoMapping.get(port);
+    }
+
+    /**
+     * 返回需要绑定在代理服务器的端口（用于用户请求）
+     *
+     * @return
+     */
+    public List<Integer> getUserPorts() {
+        List<Integer> ports = new ArrayList<Integer>();
+        Iterator<Integer> ite = inetPortLanInfoMapping.keySet().iterator();
+        while (ite.hasNext()) {
+            ports.add(ite.next());
+        }
+
+        return ports;
+    }
+
+    public static ProxyServerConfig getInstance() {
+        return instance;
+    }
+
+
+    /**
+     * 代理客户端与其后面真实服务器映射关系
+     *
+     * @author fengfei
+     *
+     */
+    @Data
+    public static class ClientProxyMapping {
+
+        /**
+         * 代理服务器端口
+         */
+        private Integer inetPort;
+
+        /**
+         * 需要代理的网络信息（代理客户端能够访问），格式 192.168.1.99:80 (必须带端口)
+         */
+        private String lan;
+
+    }
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/core/BytesMetricsHandler.java

@@ -0,0 +1,72 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.core;
+
+import fun.asgc.neutrino.proxy.server.monitor.MetricsCollector;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+
+import java.net.InetSocketAddress;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+public class BytesMetricsHandler extends ChannelDuplexHandler {
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        InetSocketAddress sa = (InetSocketAddress) ctx.channel().localAddress();
+        MetricsCollector metricsCollector = MetricsCollector.getCollector(sa.getPort());
+        metricsCollector.incrementReadBytes(((ByteBuf) msg).readableBytes());
+        metricsCollector.incrementReadMsgs(1);
+        ctx.fireChannelRead(msg);
+    }
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        InetSocketAddress sa = (InetSocketAddress) ctx.channel().localAddress();
+        MetricsCollector metricsCollector = MetricsCollector.getCollector(sa.getPort());
+        metricsCollector.incrementWroteBytes(((ByteBuf) msg).readableBytes());
+        metricsCollector.incrementWroteMsgs(1);
+        super.write(ctx, msg, promise);
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        InetSocketAddress sa = (InetSocketAddress) ctx.channel().localAddress();
+        MetricsCollector.getCollector(sa.getPort()).getChannels().incrementAndGet();
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        InetSocketAddress sa = (InetSocketAddress) ctx.channel().localAddress();
+        MetricsCollector.getCollector(sa.getPort()).getChannels().decrementAndGet();
+        super.channelInactive(ctx);
+    }
+
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/core/ProxyServerRunner.java

@@ -0,0 +1,151 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.core;
+
+import fun.asgc.neutrino.core.annotation.Autowired;
+import fun.asgc.neutrino.core.annotation.Bean;
+import fun.asgc.neutrino.core.annotation.Component;
+import fun.asgc.neutrino.core.runner.ApplicationRunner;
+import fun.asgc.neutrino.core.util.FileUtil;
+import fun.asgc.neutrino.proxy.core.IdleCheckHandler;
+import fun.asgc.neutrino.proxy.core.ProxyMessageDecoder;
+import fun.asgc.neutrino.proxy.core.ProxyMessageEncoder;
+import fun.asgc.neutrino.proxy.server.config.ProxyConfig;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.ssl.SslHandler;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.net.ssl.*;
+import java.io.InputStream;
+import java.security.KeyStore;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Slf4j
+@Component
+public class ProxyServerRunner implements ApplicationRunner {
+	@Autowired
+	private ProxyConfig proxyConfig;
+	@Autowired("serverBossGroup")
+	private NioEventLoopGroup serverBossGroup;
+	@Autowired("serverWorkerGroup")
+	private NioEventLoopGroup serverWorkerGroup;
+
+	@Override
+	public void run(String[] args) {
+		startProxyServer();
+		startProxyServerForSSL();
+	}
+
+	/**
+	 * 启动代理服务
+	 */
+	private void startProxyServer() {
+		ServerBootstrap bootstrap = new ServerBootstrap();
+		bootstrap.group(serverBossGroup, serverWorkerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
+
+			@Override
+			public void initChannel(SocketChannel ch) throws Exception {
+				proxyServerCommonInitHandler(ch);
+			}
+		});
+		try {
+			bootstrap.bind(proxyConfig.getServer().getPort()).sync();
+			log.info("代理服务启动，端口：{}", proxyConfig.getServer().getPort());
+		} catch (Exception e) {
+			log.error("代理服务异常", e);
+		}
+	}
+
+	private void startProxyServerForSSL() {
+		if (null == proxyConfig.getServer().getSslPort()) {
+			return;
+		}
+ 		ServerBootstrap bootstrap = new ServerBootstrap();
+		bootstrap.group(serverBossGroup, serverWorkerGroup)
+			.channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
+			@Override
+			public void initChannel(SocketChannel ch) throws Exception {
+				ch.pipeline().addLast(createSslHandler());
+				proxyServerCommonInitHandler(ch);
+			}
+		});
+		try {
+			bootstrap.bind(proxyConfig.getServer().getSslPort()).sync();
+			log.info("代理服务启动，SSL端口： {}", proxyConfig.getServer().getSslPort());
+		} catch (Exception e) {
+			log.error("代理服务异常", e);
+		}
+	}
+
+	private ChannelHandler createSslHandler() {
+		try {
+			InputStream jksInputStream = FileUtil.getInputStream(proxyConfig.getServer().getJksPath());
+			SSLContext serverContext = SSLContext.getInstance("TLS");
+			final KeyStore ks = KeyStore.getInstance("JKS");
+			ks.load(jksInputStream, proxyConfig.getServer().getKeyStorePassword().toCharArray());
+			final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+			kmf.init(ks, proxyConfig.getServer().getKeyManagerPassword().toCharArray());
+			TrustManager[] trustManagers = null;
+
+			serverContext.init(kmf.getKeyManagers(), trustManagers, null);
+
+			SSLEngine sslEngine = serverContext.createSSLEngine();
+			sslEngine.setUseClientMode(false);
+			sslEngine.setNeedClientAuth(false);
+
+			return new SslHandler(sslEngine);
+		} catch (Exception e) {
+			log.error("创建SSL处理器失败", e);
+			e.printStackTrace();
+		}
+		return null;
+	}
+
+	private void proxyServerCommonInitHandler(SocketChannel ch) {
+		ch.pipeline().addLast(new ProxyMessageDecoder(proxyConfig.getProtocol().getMaxFrameLength(),
+			proxyConfig.getProtocol().getLengthFieldOffset(), proxyConfig.getProtocol().getLengthFieldLength(),
+			proxyConfig.getProtocol().getLengthAdjustment(), proxyConfig.getProtocol().getInitialBytesToStrip()));
+		ch.pipeline().addLast(new ProxyMessageEncoder());
+		ch.pipeline().addLast(new IdleCheckHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds()));
+		ch.pipeline().addLast(new ServerChannelHandler());
+	}
+
+	@Bean
+	public NioEventLoopGroup serverBossGroup() {
+		return new NioEventLoopGroup();
+	}
+
+	@Bean
+	public NioEventLoopGroup serverWorkerGroup() {
+		return new NioEventLoopGroup();
+	}
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/core/ServerChannelHandler.java

@@ -0,0 +1,92 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.core;
+
+import fun.asgc.neutrino.core.base.DefaultDispatcher;
+import fun.asgc.neutrino.core.base.Dispatcher;
+import fun.asgc.neutrino.core.util.BeanManager;
+import fun.asgc.neutrino.core.util.LockUtil;
+import fun.asgc.neutrino.proxy.core.*;
+import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.*;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+public class ServerChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
+    private static volatile Dispatcher<ChannelHandlerContext, ProxyMessage> dispatcher;
+
+    public ServerChannelHandler() {
+        LockUtil.doubleCheckProcess(() -> null == dispatcher,
+            ServerChannelHandler.class,
+            () -> {
+                dispatcher = new DefaultDispatcher<>("消息调度器",
+                    BeanManager.getBeanListBySuperClass(ProxyMessageHandler.class),
+                    proxyMessage -> ProxyDataTypeEnum.of((int)proxyMessage.getType()) == null ? null : ProxyDataTypeEnum.of((int)proxyMessage.getType()).getName());
+         });
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, ProxyMessage proxyMessage) throws Exception {
+        dispatcher.dispatch(ctx, proxyMessage);
+    }
+
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+        Channel userChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
+        if (userChannel != null) {
+            userChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
+        }
+
+        super.channelWritabilityChanged(ctx);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        Channel userChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
+        if (userChannel != null && userChannel.isActive()) {
+            String clientKey = ctx.channel().attr(Constants.CLIENT_KEY).get();
+            String userId = ctx.channel().attr(Constants.USER_ID).get();
+            Channel cmdChannel = ProxyChannelManager.getCmdChannel(clientKey);
+            if (cmdChannel != null) {
+                ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, userId);
+            }
+
+            // 数据发送完成后再关闭连接，解决http1.0数据传输问题
+            userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+            userChannel.close();
+        } else {
+            ProxyChannelManager.removeCmdChannel(ctx.channel());
+        }
+
+        super.channelInactive(ctx);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        super.exceptionCaught(ctx, cause);
+    }
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/core/UserChannelHandler.java

@@ -0,0 +1,153 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.core;
+
+import fun.asgc.neutrino.proxy.core.Constants;
+import fun.asgc.neutrino.proxy.core.ProxyMessage;
+import fun.asgc.neutrino.proxy.server.config.ProxyServerConfig;
+import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+public class UserChannelHandler extends SimpleChannelInboundHandler<ByteBuf> {
+
+    private static AtomicLong userIdProducer = new AtomicLong(0);
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+
+        // 当出现异常就关闭连接
+        ctx.close();
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {
+
+        // 通知代理客户端
+        Channel userChannel = ctx.channel();
+        Channel proxyChannel = userChannel.attr(Constants.NEXT_CHANNEL).get();
+        if (proxyChannel == null) {
+
+            // 该端口还没有代理客户端
+            ctx.channel().close();
+        } else {
+            byte[] bytes = new byte[buf.readableBytes()];
+            buf.readBytes(bytes);
+            String userId = ProxyChannelManager.getUserChannelUserId(userChannel);
+            proxyChannel.writeAndFlush(ProxyMessage.buildTransferMessage(userId, bytes));
+        }
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        Channel userChannel = ctx.channel();
+        InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress();
+        Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort());
+
+        if (cmdChannel == null) {
+            // 该端口还没有代理客户端
+            ctx.channel().close();
+        } else {
+            String userId = newUserId();
+            String lanInfo = ProxyServerConfig.getInstance().getLanInfo(sa.getPort());
+            // 用户连接到代理服务器时，设置用户连接不可读，等待代理后端服务器连接成功后再改变为可读状态
+            userChannel.config().setOption(ChannelOption.AUTO_READ, false);
+            ProxyChannelManager.addUserChannelToCmdChannel(cmdChannel, userId, userChannel);
+            cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(userId).setData(lanInfo.getBytes()));
+        }
+
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+
+        // 通知代理客户端
+        Channel userChannel = ctx.channel();
+        InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress();
+        Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort());
+        if (cmdChannel == null) {
+
+            // 该端口还没有代理客户端
+            ctx.channel().close();
+        } else {
+
+            // 用户连接断开，从控制连接中移除
+            String userId = ProxyChannelManager.getUserChannelUserId(userChannel);
+            ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, userId);
+            Channel proxyChannel = userChannel.attr(Constants.NEXT_CHANNEL).get();
+            if (proxyChannel != null && proxyChannel.isActive()) {
+                proxyChannel.attr(Constants.NEXT_CHANNEL).remove();
+                proxyChannel.attr(Constants.CLIENT_KEY).remove();
+                proxyChannel.attr(Constants.USER_ID).remove();
+
+                proxyChannel.config().setOption(ChannelOption.AUTO_READ, true);
+                // 通知客户端，用户连接已经断开
+                proxyChannel.writeAndFlush(ProxyMessage.buildDisconnectMessage(userId));
+            }
+        }
+
+        super.channelInactive(ctx);
+    }
+
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+
+        // 通知代理客户端
+        Channel userChannel = ctx.channel();
+        InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress();
+        Channel cmdChannel = ProxyChannelManager.getCmdChannel(sa.getPort());
+        if (cmdChannel == null) {
+
+            // 该端口还没有代理客户端
+            ctx.channel().close();
+        } else {
+            Channel proxyChannel = userChannel.attr(Constants.NEXT_CHANNEL).get();
+            if (proxyChannel != null) {
+                proxyChannel.config().setOption(ChannelOption.AUTO_READ, userChannel.isWritable());
+            }
+        }
+
+        super.channelWritabilityChanged(ctx);
+    }
+
+    /**
+     * 为用户连接产生ID
+     *
+     * @return
+     */
+    private static String newUserId() {
+        return String.valueOf(userIdProducer.incrementAndGet());
+    }
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/handler/ProxyMessageAuthHandler.java

@@ -0,0 +1,126 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.handler;
+
+import com.alibaba.fastjson.JSONObject;
+import fun.asgc.neutrino.core.annotation.Autowired;
+import fun.asgc.neutrino.core.annotation.Component;
+import fun.asgc.neutrino.core.annotation.Match;
+import fun.asgc.neutrino.core.util.BeanManager;
+import fun.asgc.neutrino.proxy.core.*;
+import fun.asgc.neutrino.proxy.server.config.ProxyConfig;
+import fun.asgc.neutrino.proxy.server.config.ProxyServerConfig;
+import fun.asgc.neutrino.proxy.server.core.BytesMetricsHandler;
+import fun.asgc.neutrino.proxy.server.core.UserChannelHandler;
+import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import lombok.extern.slf4j.Slf4j;
+
+import java.net.BindException;
+import java.util.List;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Slf4j
+@Match(type = Constants.ProxyDataTypeName.AUTH)
+@Component
+public class ProxyMessageAuthHandler implements ProxyMessageHandler {
+	@Autowired("serverBossGroup")
+	private NioEventLoopGroup serverBossGroup;
+	@Autowired("serverWorkerGroup")
+	private NioEventLoopGroup serverWorkerGroup;
+	@Autowired
+	private ProxyConfig proxyConfig;
+
+	@Override
+	public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
+		ProxyClientConfig clientConfig = JSONObject.parseObject(proxyMessage.getInfo(), ProxyClientConfig.class);
+		String clientKey = clientConfig.getClientKey();
+		if (!proxyConfig.getLicenseMap().containsKey(clientKey)) {
+			ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.AUTH_FAILED, "无效的clientKey"));
+			ctx.channel().close();
+			return;
+		}
+
+		if (proxyConfig.getLicenseMap().get(clientKey) != -1 && clientConfig.getProxy().size() > proxyConfig.getLicenseMap().get(clientKey)) {
+			ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.AUTH_FAILED, "代理端口数超过license限制"));
+			ctx.channel().close();
+			return;
+		}
+
+		ProxyServerConfig.getInstance().addClientConfig(clientConfig);
+		List<Integer> ports = ProxyServerConfig.getInstance().getClientInetPorts(clientKey);
+		if (ports == null) {
+			ctx.channel().close();
+			return;
+		}
+
+		Channel channel = ProxyChannelManager.getCmdChannel(clientKey);
+		if (channel != null) {
+			ctx.channel().close();
+			return;
+		}
+
+		ProxyChannelManager.addCmdChannel(ports, clientKey, ctx.channel());
+
+		startUserPortServer(ports);
+	}
+
+	@Override
+	public String name() {
+		return ProxyDataTypeEnum.AUTH.getDesc();
+	}
+
+	private void startUserPortServer(List<Integer> ports) {
+		ServerBootstrap bootstrap = new ServerBootstrap();
+		bootstrap.group(serverBossGroup, serverWorkerGroup)
+			.channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
+			@Override
+			public void initChannel(SocketChannel ch) throws Exception {
+				ch.pipeline().addFirst(new BytesMetricsHandler());
+				ch.pipeline().addLast(new UserChannelHandler());
+			}
+		});
+
+		for (int port : ports) {
+			try {
+				bootstrap.bind(port).get();
+				log.info("绑定用户端口： {}", port);
+			} catch (Exception ex) {
+				// BindException表示该端口已经绑定过
+				if (!(ex.getCause() instanceof BindException)) {
+					throw new RuntimeException(ex);
+				}
+			}
+		}
+	}
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/handler/ProxyMessageConnectHandler.java

@@ -0,0 +1,80 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.handler;
+
+import fun.asgc.neutrino.core.annotation.Component;
+import fun.asgc.neutrino.core.annotation.Match;
+import fun.asgc.neutrino.proxy.core.Constants;
+import fun.asgc.neutrino.proxy.core.ProxyDataTypeEnum;
+import fun.asgc.neutrino.proxy.core.ProxyMessage;
+import fun.asgc.neutrino.proxy.core.ProxyMessageHandler;
+import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOption;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Match(type = Constants.ProxyDataTypeName.CONNECT)
+@Component
+public class ProxyMessageConnectHandler implements ProxyMessageHandler {
+
+	@Override
+	public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
+		String info = proxyMessage.getInfo();
+		if (info == null) {
+			ctx.channel().close();
+			return;
+		}
+
+		String[] tokens = info.split("@");
+		if (tokens.length != 2) {
+			ctx.channel().close();
+			return;
+		}
+
+		Channel cmdChannel = ProxyChannelManager.getCmdChannel(tokens[1]);
+		if (cmdChannel == null) {
+			ctx.channel().close();
+			return;
+		}
+
+		Channel userChannel = ProxyChannelManager.getUserChannel(cmdChannel, tokens[0]);
+		if (userChannel != null) {
+			ctx.channel().attr(Constants.USER_ID).set(tokens[0]);
+			ctx.channel().attr(Constants.CLIENT_KEY).set(tokens[1]);
+			ctx.channel().attr(Constants.NEXT_CHANNEL).set(userChannel);
+			userChannel.attr(Constants.NEXT_CHANNEL).set(ctx.channel());
+			// 代理客户端与后端服务器连接成功，修改用户连接为可读状态
+			userChannel.config().setOption(ChannelOption.AUTO_READ, true);
+		}
+	}
+
+	@Override
+	public String name() {
+		return ProxyDataTypeEnum.CONNECT.getDesc();
+	}
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/handler/ProxyMessageDisconnectHandler.java

@@ -0,0 +1,81 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.handler;
+
+import fun.asgc.neutrino.core.annotation.Component;
+import fun.asgc.neutrino.core.annotation.Match;
+import fun.asgc.neutrino.proxy.core.Constants;
+import fun.asgc.neutrino.proxy.core.ProxyDataTypeEnum;
+import fun.asgc.neutrino.proxy.core.ProxyMessage;
+import fun.asgc.neutrino.proxy.core.ProxyMessageHandler;
+import fun.asgc.neutrino.proxy.server.util.ProxyChannelManager;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Match(type = Constants.ProxyDataTypeName.DISCONNECT)
+@Component
+public class ProxyMessageDisconnectHandler implements ProxyMessageHandler {
+
+	@Override
+	public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
+		String clientKey = ctx.channel().attr(Constants.CLIENT_KEY).get();
+
+		// 代理连接没有连上服务器由控制连接发送用户端断开连接消息
+		if (clientKey == null) {
+			String userId = proxyMessage.getInfo();
+			Channel userChannel = ProxyChannelManager.removeUserChannelFromCmdChannel(ctx.channel(), userId);
+			if (userChannel != null) {
+				// 数据发送完成后再关闭连接，解决http1.0数据传输问题
+				userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+			}
+			return;
+		}
+
+		Channel cmdChannel = ProxyChannelManager.getCmdChannel(clientKey);
+		if (cmdChannel == null) {
+			return;
+		}
+
+		Channel userChannel = ProxyChannelManager.removeUserChannelFromCmdChannel(cmdChannel, ctx.channel().attr(Constants.USER_ID).get());
+		if (userChannel != null) {
+			// 数据发送完成后再关闭连接，解决http1.0数据传输问题
+			userChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+			ctx.channel().attr(Constants.NEXT_CHANNEL).remove();
+			ctx.channel().attr(Constants.CLIENT_KEY).remove();
+			ctx.channel().attr(Constants.USER_ID).remove();
+		}
+	}
+
+	@Override
+	public String name() {
+		return ProxyDataTypeEnum.DISCONNECT.getDesc();
+	}
+
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/handler/ProxyMessageHeartbeatHandler.java

@@ -0,0 +1,52 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.handler;
+
+import fun.asgc.neutrino.core.annotation.Component;
+import fun.asgc.neutrino.core.annotation.Match;
+import fun.asgc.neutrino.proxy.core.Constants;
+import fun.asgc.neutrino.proxy.core.ProxyDataTypeEnum;
+import fun.asgc.neutrino.proxy.core.ProxyMessage;
+import fun.asgc.neutrino.proxy.core.ProxyMessageHandler;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Match(type = Constants.ProxyDataTypeName.HEARTBEAT)
+@Component
+public class ProxyMessageHeartbeatHandler implements ProxyMessageHandler {
+
+	@Override
+	public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
+		ctx.channel().writeAndFlush(ProxyMessage.buildHeartbeatMessage());
+	}
+
+	@Override
+	public String name() {
+		return ProxyDataTypeEnum.HEARTBEAT.getDesc();
+	}
+
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/handler/ProxyMessageTransferHandler.java

@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.handler;
+
+import fun.asgc.neutrino.core.annotation.Component;
+import fun.asgc.neutrino.core.annotation.Match;
+import fun.asgc.neutrino.proxy.core.Constants;
+import fun.asgc.neutrino.proxy.core.ProxyDataTypeEnum;
+import fun.asgc.neutrino.proxy.core.ProxyMessage;
+import fun.asgc.neutrino.proxy.core.ProxyMessageHandler;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Match(type = Constants.ProxyDataTypeName.TRANSFER)
+@Component
+public class ProxyMessageTransferHandler implements ProxyMessageHandler {
+
+	@Override
+	public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
+		Channel userChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
+		if (userChannel != null) {
+			ByteBuf buf = ctx.alloc().buffer(proxyMessage.getData().length);
+			buf.writeBytes(proxyMessage.getData());
+			userChannel.writeAndFlush(buf);
+		}
+	}
+
+	@Override
+	public String name() {
+		return ProxyDataTypeEnum.TRANSFER.getDesc();
+	}
+
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/monitor/Metrics.java

@@ -0,0 +1,44 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.monitor;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+@Data
+public class Metrics implements Serializable {
+    private static final long serialVersionUID = 1L;
+    private int port;
+    private long readBytes;
+    private long wroteBytes;
+    private long readMsgs;
+    private long wroteMsgs;
+    private int channels;
+    private long timestamp;
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/monitor/MetricsCollector.java

@@ -0,0 +1,148 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.monitor;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+public class MetricsCollector {
+
+    private static Map<Integer, MetricsCollector> metricsCollectors = new ConcurrentHashMap<Integer, MetricsCollector>();
+
+    private Integer port;
+
+    private AtomicLong readBytes = new AtomicLong();
+
+    private AtomicLong wroteBytes = new AtomicLong();
+
+    private AtomicLong readMsgs = new AtomicLong();
+
+    private AtomicLong wroteMsgs = new AtomicLong();
+
+    private AtomicInteger channels = new AtomicInteger();
+
+    private MetricsCollector() {
+    }
+
+    public static MetricsCollector getCollector(Integer port) {
+        MetricsCollector collector = metricsCollectors.get(port);
+        if (collector == null) {
+            synchronized (metricsCollectors) {
+                collector = metricsCollectors.get(port);
+                if (collector == null) {
+                    collector = new MetricsCollector();
+                    collector.setPort(port);
+                    metricsCollectors.put(port, collector);
+                }
+            }
+        }
+
+        return collector;
+    }
+
+    public static List<Metrics> getAndResetAllMetrics() {
+        List<Metrics> allMetrics = new ArrayList<Metrics>();
+        Iterator<Entry<Integer, MetricsCollector>> ite = metricsCollectors.entrySet().iterator();
+        while (ite.hasNext()) {
+            allMetrics.add(ite.next().getValue().getAndResetMetrics());
+        }
+
+        return allMetrics;
+    }
+
+    public static List<Metrics> getAllMetrics() {
+        List<Metrics> allMetrics = new ArrayList<Metrics>();
+        Iterator<Entry<Integer, MetricsCollector>> ite = metricsCollectors.entrySet().iterator();
+        while (ite.hasNext()) {
+            allMetrics.add(ite.next().getValue().getMetrics());
+        }
+
+        return allMetrics;
+    }
+
+    public Metrics getAndResetMetrics() {
+        Metrics metrics = new Metrics();
+        metrics.setChannels(channels.get());
+        metrics.setPort(port);
+        metrics.setReadBytes(readBytes.getAndSet(0));
+        metrics.setWroteBytes(wroteBytes.getAndSet(0));
+        metrics.setTimestamp(System.currentTimeMillis());
+        metrics.setReadMsgs(readMsgs.getAndSet(0));
+        metrics.setWroteMsgs(wroteMsgs.getAndSet(0));
+
+        return metrics;
+    }
+
+    public Metrics getMetrics() {
+        Metrics metrics = new Metrics();
+        metrics.setChannels(channels.get());
+        metrics.setPort(port);
+        metrics.setReadBytes(readBytes.get());
+        metrics.setWroteBytes(wroteBytes.get());
+        metrics.setTimestamp(System.currentTimeMillis());
+        metrics.setReadMsgs(readMsgs.get());
+        metrics.setWroteMsgs(wroteMsgs.get());
+
+        return metrics;
+    }
+
+    public void incrementReadBytes(long bytes) {
+        readBytes.addAndGet(bytes);
+    }
+
+    public void incrementWroteBytes(long bytes) {
+        wroteBytes.addAndGet(bytes);
+    }
+
+    public void incrementReadMsgs(long msgs) {
+        readMsgs.addAndGet(msgs);
+    }
+
+    public void incrementWroteMsgs(long msgs) {
+        wroteMsgs.addAndGet(msgs);
+    }
+
+    public AtomicInteger getChannels() {
+        return channels;
+    }
+
+    public Integer getPort() {
+        return port;
+    }
+
+    public void setPort(Integer port) {
+        this.port = port;
+    }
+
+}

neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyChannelManager.java

@@ -0,0 +1,201 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package fun.asgc.neutrino.proxy.server.util;
+
+import fun.asgc.neutrino.proxy.core.Constants;
+import fun.asgc.neutrino.proxy.server.config.ProxyServerConfig;
+import io.netty.channel.Channel;
+import io.netty.util.AttributeKey;
+
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ *
+ * @author: aoshiguchen
+ * @date: 2022/6/16
+ */
+public class ProxyChannelManager {
+
+    private static final AttributeKey<Map<String, Channel>> USER_CHANNELS = AttributeKey.newInstance("user_channels");
+
+    private static final AttributeKey<String> REQUEST_LAN_INFO = AttributeKey.newInstance("request_lan_info");
+
+    private static final AttributeKey<List<Integer>> CHANNEL_PORT = AttributeKey.newInstance("channel_port");
+
+    private static final AttributeKey<String> CHANNEL_CLIENT_KEY = AttributeKey.newInstance("channel_client_key");
+
+    private static Map<Integer, Channel> portCmdChannelMapping = new ConcurrentHashMap<Integer, Channel>();
+
+    private static Map<String, Channel> cmdChannels = new ConcurrentHashMap<String, Channel>();
+
+    /**
+     * 增加代理服务器端口与代理控制客户端连接的映射关系
+     *
+     * @param ports
+     * @param channel
+     */
+    public static void addCmdChannel(List<Integer> ports, String clientKey, Channel channel) {
+        if (ports == null) {
+            throw new IllegalArgumentException("port can not be null");
+        }
+
+        // 客户端（proxy-client）相对较少，这里同步的比较重
+        // 保证服务器对外端口与客户端到服务器的连接关系在临界情况时调用removeChannel(Channel channel)时不出问题
+        synchronized (portCmdChannelMapping) {
+            for (int port : ports) {
+                portCmdChannelMapping.put(port, channel);
+            }
+        }
+
+        channel.attr(CHANNEL_PORT).set(ports);
+        channel.attr(CHANNEL_CLIENT_KEY).set(clientKey);
+        channel.attr(USER_CHANNELS).set(new ConcurrentHashMap<String, Channel>());
+        cmdChannels.put(clientKey, channel);
+    }
+
+    /**
+     * 代理客户端连接断开后清除关系
+     *
+     * @param channel
+     */
+    public static void removeCmdChannel(Channel channel) {
+        if (channel.attr(CHANNEL_PORT).get() == null) {
+            return;
+        }
+
+        String clientKey = channel.attr(CHANNEL_CLIENT_KEY).get();
+        Channel channel0 = cmdChannels.remove(clientKey);
+        if (channel != channel0) {
+            cmdChannels.put(clientKey, channel);
+        }
+
+        List<Integer> ports = channel.attr(CHANNEL_PORT).get();
+        for (int port : ports) {
+            Channel proxyChannel = portCmdChannelMapping.remove(port);
+            if (proxyChannel == null) {
+                continue;
+            }
+
+            // 在执行断连之前新的连接已经连上来了
+            if (proxyChannel != channel) {
+                portCmdChannelMapping.put(port, proxyChannel);
+            }
+        }
+
+        if (channel.isActive()) {
+            channel.close();
+        }
+
+        Map<String, Channel> userChannels = getUserChannels(channel);
+        Iterator<String> ite = userChannels.keySet().iterator();
+        while (ite.hasNext()) {
+            Channel userChannel = userChannels.get(ite.next());
+            if (userChannel.isActive()) {
+                userChannel.close();
+            }
+        }
+    }
+
+    public static Channel getCmdChannel(Integer port) {
+        return portCmdChannelMapping.get(port);
+    }
+
+    public static Channel getCmdChannel(String clientKey) {
+        return cmdChannels.get(clientKey);
+    }
+
+    /**
+     * 增加用户连接与代理客户端连接关系
+     *
+     * @param userId
+     * @param userChannel
+     */
+    public static void addUserChannelToCmdChannel(Channel cmdChannel, String userId, Channel userChannel) {
+        InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress();
+        String lanInfo = ProxyServerConfig.getInstance().getLanInfo(sa.getPort());
+        userChannel.attr(Constants.USER_ID).set(userId);
+        userChannel.attr(REQUEST_LAN_INFO).set(lanInfo);
+        cmdChannel.attr(USER_CHANNELS).get().put(userId, userChannel);
+    }
+
+    /**
+     * 删除用户连接与代理客户端连接关系
+     *
+     * @param userId
+     * @return
+     */
+    public static Channel removeUserChannelFromCmdChannel(Channel cmdChannel, String userId) {
+        if (cmdChannel.attr(USER_CHANNELS).get() == null) {
+            return null;
+        }
+
+        synchronized (cmdChannel) {
+            return cmdChannel.attr(USER_CHANNELS).get().remove(userId);
+        }
+    }
+
+    /**
+     * 根据代理客户端连接与用户编号获取用户连接
+     *
+     * @param userId
+     * @return
+     */
+    public static Channel getUserChannel(Channel cmdChannel, String userId) {
+        return cmdChannel.attr(USER_CHANNELS).get().get(userId);
+    }
+
+    /**
+     * 获取用户编号
+     *
+     * @param userChannel
+     * @return
+     */
+    public static String getUserChannelUserId(Channel userChannel) {
+        return userChannel.attr(Constants.USER_ID).get();
+    }
+
+    /**
+     * 获取用户请求的内网IP端口信息
+     *
+     * @param userChannel
+     * @return
+     */
+    public static String getUserChannelRequestLanInfo(Channel userChannel) {
+        return userChannel.attr(REQUEST_LAN_INFO).get();
+    }
+
+    /**
+     * 获取代理控制客户端连接绑定的所有用户连接
+     *
+     * @param cmdChannel
+     * @return
+     */
+    public static Map<String, Channel> getUserChannels(Channel cmdChannel) {
+        return cmdChannel.attr(USER_CHANNELS).get();
+    }
+
+}