From a93646619e4686a61acd69c1b308e3c03b7d09a5 Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Mon, 26 Jan 2026 18:04:09 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=B0=B4=E4=BD=8D=E7=BA=BF?= =?UTF-8?q?=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/config/ProxyConfig.java | 41 ++++++++++ .../client/config/ProxyConfiguration.java | 17 ++++- .../neutrinoproxy/client/util/StringUtil.java | 75 +++++++++++++++++++ .../server/base/proxy/ProxyConfig.java | 40 ++++++++++ .../server/base/proxy/ProxyConfiguration.java | 11 ++- .../server/proxy/core/ProxyTunnelServer.java | 8 ++ 6 files changed, 184 insertions(+), 8 deletions(-) create mode 100644 neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/StringUtil.java diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfig.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfig.java index a33f3520..3a1a729d 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfig.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfig.java @@ -1,6 +1,10 @@ package org.dromara.neutrinoproxy.client.config; +import cn.hutool.core.util.StrUtil; +import io.netty.channel.WriteBufferWaterMark; import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.dromara.neutrinoproxy.client.util.StringUtil; import org.noear.solon.annotation.Component; import org.noear.solon.annotation.Inject; @@ -9,6 +13,7 @@ import org.noear.solon.annotation.Inject; * @author: aoshiguchen * @date: 2022/6/16 */ +@Slf4j @Data @Component public class ProxyConfig { @@ -29,6 +34,8 @@ public class ProxyConfig { private Integer readIdleTime; private Integer writeIdleTime; private Integer allIdleTimeSeconds; + // 水位线 + private String waterMark; } @Data @@ -71,4 +78,38 @@ public class ProxyConfig { private String puppetPortRange; private Boolean transferLogEnable; } + + + private WriteBufferWaterMark waterMark; + private boolean isParseWaterMark = false; + + public synchronized WriteBufferWaterMark getWaterMark() { + if (isParseWaterMark) { + return waterMark; + } + isParseWaterMark = true; + if (null == protocol || StrUtil.isBlank(protocol.getWaterMark())) { + return null; + } + String[] tmp = protocol.getWaterMark().split("/"); + if (tmp.length != 2) { + log.info("[配置解析] 水位线配置参数格式有误! config={}", protocol.getWaterMark()); + return null; + } + String lowStr = tmp[0].trim(); + String highStr = tmp[1].trim(); + if (!StringUtil.isBytesDesc(lowStr) || !StringUtil.isBytesDesc(highStr)) { + log.info("[配置解析] 水位线配置参数格式有误! config={}", protocol.getWaterMark()); + return null; + } + Long low = StringUtil.parseBytes(lowStr); + Long high = StringUtil.parseBytes(highStr); + if (null == low || null == high || low >= high) { + log.info("[配置解析] 水位线配置参数格式或大小有误! config={}", protocol.getWaterMark()); + return null; + } + waterMark = new WriteBufferWaterMark(low.intValue(), high.intValue()); + log.info("[配置解析] 水位线配置 config={},low={},high={}", protocol.getWaterMark(), low.intValue(), high.intValue()); + return waterMark; + } } diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java index 87dda221..16b8f156 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java @@ -1,8 +1,6 @@ package org.dromara.neutrinoproxy.client.config; -import io.netty.channel.ChannelInitializer; -import io.netty.channel.ChannelOption; -import io.netty.channel.ChannelPipeline; +import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioDatagramChannel; @@ -16,7 +14,6 @@ import org.dromara.neutrinoproxy.core.aot.NeutrinoCoreRuntimeNativeRegistrar; import org.dromara.neutrinoproxy.core.dispatcher.DefaultDispatcher; import org.dromara.neutrinoproxy.core.dispatcher.Dispatcher; import io.netty.bootstrap.Bootstrap; -import io.netty.channel.ChannelHandlerContext; import org.noear.solon.Solon; import org.noear.solon.annotation.Bean; import org.noear.solon.annotation.Configuration; @@ -108,6 +105,12 @@ public class ProxyConfiguration implements LifecycleBean { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(tunnelWorkGroup); bootstrap.channel(NioSocketChannel.class); + + WriteBufferWaterMark waterMark = proxyConfig.getWaterMark(); + if (null != waterMark) { + bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, waterMark); + } + bootstrap.remoteAddress(InetSocketAddress.createUnresolved(proxyConfig.getTunnel().getServerIp(), proxyConfig.getTunnel().getServerPort())); bootstrap.handler(new ChannelInitializer() { @@ -165,6 +168,12 @@ public class ProxyConfiguration implements LifecycleBean { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(tcpRealServerWorkGroup); bootstrap.channel(NioSocketChannel.class); + + WriteBufferWaterMark waterMark = proxyConfig.getWaterMark(); + if (null != waterMark) { + bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, waterMark); + } + bootstrap.handler(new ChannelInitializer() { @Override diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/StringUtil.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/StringUtil.java new file mode 100644 index 00000000..4cac749e --- /dev/null +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/StringUtil.java @@ -0,0 +1,75 @@ +package org.dromara.neutrinoproxy.client.util; + +import cn.hutool.core.util.StrUtil; + +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * @author: aoshiguchen + * @date: 2023/12/15 + */ +public class StringUtil { + private static final Integer BYTES_MUL_KB = 1024; + private static final Integer BYTES_MUL_MB = BYTES_MUL_KB * 1024; + private static final Integer BYTES_MUL_GB = BYTES_MUL_MB * 1024; + private static final String[] BYTES_UNIT_STR = {"B", "K", "KB", "M", "MB", "G", "GB"}; + private static final Integer[] BYTES_UNIT_MUL = {1, BYTES_MUL_KB, BYTES_MUL_KB, BYTES_MUL_MB, BYTES_MUL_MB, BYTES_MUL_GB, BYTES_MUL_GB}; + private static final String BYTES_DESC_REGEX = "\\s*(\\d+\\.*\\d*)\\s*(B|K|KB|M|MB|G|GB)\\s*"; + private static final Pattern BYTES_DESC_PATTERN = Pattern.compile(BYTES_DESC_REGEX); + + /** + * 校验是否符合字节描述 + * @param desc + * @return + */ + public static boolean isBytesDesc(String desc) { + if (StrUtil.isBlank(desc)) { + return false; + } + return desc.toUpperCase().matches(BYTES_DESC_REGEX); + } + + /** + * 解析字节数 + * 支持B、K、KB、M、MB、G、GB 忽略大小写、忽略首尾空格、忽略数字与单位之间的空格 + * @param desc + * @return + */ + public static Long parseBytes(String desc) { + try { + if (!isBytesDesc(desc)) { + return null; + } + Matcher matcher = BYTES_DESC_PATTERN.matcher(desc.toUpperCase()); + boolean found = matcher.find(); + if (!found) { + return null; + } + Double n = Double.parseDouble(matcher.group(1)); + String unit = matcher.group(2); + Integer unitIndex = getBytesUnitIndex(unit); + if (null == unitIndex) { + return null; + } + + return (long)(n * BYTES_UNIT_MUL[unitIndex]); + } catch (Exception e) { + // ignore + } + + return null; + } + + private static Integer getBytesUnitIndex(String unit) { + if (StrUtil.isBlank(unit)) { + return null; + } + for (int i = 0; i < BYTES_UNIT_STR.length; i++) { + if (unit.equals(BYTES_UNIT_STR[i])) { + return i; + } + } + return null; + } +} diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfig.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfig.java index eb48ca3b..0e14aac1 100644 --- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfig.java +++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfig.java @@ -1,6 +1,10 @@ package org.dromara.neutrinoproxy.server.base.proxy; +import cn.hutool.core.util.StrUtil; +import io.netty.channel.WriteBufferWaterMark; import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.dromara.neutrinoproxy.server.util.StringUtil; import org.noear.solon.annotation.Component; import org.noear.solon.annotation.Inject; @@ -9,6 +13,7 @@ import org.noear.solon.annotation.Inject; * @author: aoshiguchen * @date: 2022/6/16 */ +@Slf4j @Data @Component public class ProxyConfig { @@ -38,6 +43,8 @@ public class ProxyConfig { private Integer readIdleTime; private Integer writeIdleTime; private Integer allIdleTimeSeconds; + // 水位线 + private String waterMark; } @Data @@ -76,4 +83,37 @@ public class ProxyConfig { private Integer workThreadCount; private Boolean transferLogEnable; } + + private WriteBufferWaterMark waterMark; + private boolean isParseWaterMark = false; + + public synchronized WriteBufferWaterMark getWaterMark() { + if (isParseWaterMark) { + return waterMark; + } + isParseWaterMark = true; + if (null == protocol || StrUtil.isBlank(protocol.getWaterMark())) { + return null; + } + String[] tmp = protocol.getWaterMark().split("/"); + if (tmp.length != 2) { + log.info("[配置解析] 水位线配置参数格式有误! config={}", protocol.getWaterMark()); + return null; + } + String lowStr = tmp[0].trim(); + String highStr = tmp[1].trim(); + if (!StringUtil.isBytesDesc(lowStr) || !StringUtil.isBytesDesc(highStr)) { + log.info("[配置解析] 水位线配置参数格式有误! config={}", protocol.getWaterMark()); + return null; + } + Long low = StringUtil.parseBytes(lowStr); + Long high = StringUtil.parseBytes(highStr); + if (null == low || null == high || low >= high) { + log.info("[配置解析] 水位线配置参数格式或大小有误! config={}", protocol.getWaterMark()); + return null; + } + waterMark = new WriteBufferWaterMark(low.intValue(), high.intValue()); + log.info("[配置解析] 水位线配置 config={},low={},high={}", protocol.getWaterMark(), low.intValue(), high.intValue()); + return waterMark; + } } diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfiguration.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfiguration.java index a9a69fb8..693fceeb 100644 --- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfiguration.java +++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/base/proxy/ProxyConfiguration.java @@ -1,10 +1,9 @@ package org.dromara.neutrinoproxy.server.base.proxy; +import cn.hutool.core.util.StrUtil; import io.netty.bootstrap.Bootstrap; import io.netty.bootstrap.ServerBootstrap; -import io.netty.channel.ChannelInitializer; -import io.netty.channel.ChannelOption; -import io.netty.channel.ChannelPipeline; +import io.netty.channel.*; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioDatagramChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; @@ -15,7 +14,6 @@ import org.dromara.neutrinoproxy.core.ProxyMessageHandler; import org.dromara.neutrinoproxy.core.aot.NeutrinoCoreRuntimeNativeRegistrar; import org.dromara.neutrinoproxy.core.dispatcher.DefaultDispatcher; import org.dromara.neutrinoproxy.core.dispatcher.Dispatcher; -import io.netty.channel.ChannelHandlerContext; import io.netty.channel.nio.NioEventLoopGroup; import org.dromara.neutrinoproxy.server.proxy.core.TcpVisitorChannelHandler; import org.dromara.neutrinoproxy.server.proxy.core.UdpVisitorChannelHandler; @@ -69,6 +67,11 @@ public class ProxyConfiguration implements LifecycleBean { @Inject ProxyConfig proxyConfig ) { ServerBootstrap bootstrap = new ServerBootstrap(); + + WriteBufferWaterMark waterMark = proxyConfig.getWaterMark(); + if (null != waterMark) { + bootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, waterMark); + } bootstrap.group(tcpServerBossGroup, tcpServerWorkerGroup) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer() { diff --git a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelServer.java b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelServer.java index 12c5a429..721a3dbd 100644 --- a/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelServer.java +++ b/neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/ProxyTunnelServer.java @@ -1,5 +1,7 @@ package org.dromara.neutrinoproxy.server.proxy.core; +import io.netty.channel.ChannelOption; +import io.netty.channel.WriteBufferWaterMark; import io.netty.handler.logging.LoggingHandler; import org.dromara.neutrinoproxy.core.ProxyMessageDecoder; import org.dromara.neutrinoproxy.core.ProxyMessageEncoder; @@ -54,6 +56,12 @@ public class ProxyTunnelServer implements EventListener { */ private void startProxyServer() { ServerBootstrap bootstrap = new ServerBootstrap(); + + WriteBufferWaterMark waterMark = proxyConfig.getWaterMark(); + if (null != waterMark) { + bootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, waterMark); + } + bootstrap.group(serverBossGroup, serverWorkerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer() { @Override