From 2ec87abda973c5bbcc3974f5682dc21dc11860c8 Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Fri, 2 Jun 2023 22:09:01 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF=E9=87=8D=E8=BF=9E?= =?UTF-8?q?=E6=9C=BA=E5=88=B6=E8=B0=83=E6=95=B4=EF=BC=8C=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E9=A6=96=E6=AC=A1=E8=BF=9E=E6=8E=A5=E5=A4=B1=E8=B4=A5=E4=B9=9F?= =?UTF-8?q?=E8=BF=9B=E8=A1=8C=E9=87=8D=E8=BF=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/core/ProxyClientService.java | 199 +++++++++--------- 1 file changed, 104 insertions(+), 95 deletions(-) diff --git a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyClientService.java b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyClientService.java index cf138ffa..48430cb6 100644 --- a/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyClientService.java +++ b/neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyClientService.java @@ -49,15 +49,15 @@ public class ProxyClientService { /** * 重连间隔（秒） */ - private static final long RECONNECT_INTERVAL_SECONDS = 5; + private static final long RECONNECT_INTERVAL_SECONDS = 8; /** * 重连次数 */ private volatile int reconnectCount = 0; - /** - * 启用重连服务 - */ - private volatile boolean reconnectServiceEnable = false; +// /** +// * 启用重连服务 +// */ +// private volatile boolean reconnectServiceEnable = false; /** * 重连服务执行器 */ @@ -80,7 +80,7 @@ public class ProxyClientService { bootstrap.group(workerGroup); bootstrap.channel(NioSocketChannel.class); - bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000); bootstrap.option(ChannelOption.SO_KEEPALIVE, true); /** * TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。（一个连接会设置MSS参数，因此，TCP/IP希望每次都能够以MSS尺寸的数据块来发送数据）。 @@ -104,103 +104,112 @@ public class ProxyClientService { ch.pipeline().addLast(new ClientChannelHandler()); } }); - this.start(); - } - public void start() { - if (StrUtil.isEmpty(proxyConfig.getClient().getServerIp())) { - log.error("not found server-ip config."); - Solon.stop(); - return; + try { + this.start(); + } catch (Exception e) { + // 启动连不上也做一下重连，因此先catch异常 + log.error("启动异常", e); + } } - if (null == proxyConfig.getClient().getServerPort()) { - log.error("not found server-port config."); - Solon.stop(); - return; + + public void start() { + if (StrUtil.isEmpty(proxyConfig.getClient().getServerIp())) { + log.error("not found server-ip config."); + Solon.stop(); + return; + } + if (null == proxyConfig.getClient().getServerPort()) { + log.error("not found server-port config."); + Solon.stop(); + return; + } + if (null != proxyConfig.getClient().getSslEnable() && proxyConfig.getClient().getSslEnable() + && StrUtil.isEmpty(proxyConfig.getClient().getJksPath())) { + log.error("not found jks-path config."); + Solon.stop(); + return; + } + if (StrUtil.isEmpty(proxyConfig.getClient().getLicenseKey())) { + log.error("not found license-key config."); + Solon.stop(); + return; + } + if (null == channel || !channel.isActive()) { + try { + connectProxyServer(); + } catch (Exception e) { + log.error("client start error", e); + } + } else { + channel.writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey())); + } } - if (null != proxyConfig.getClient().getSslEnable() && proxyConfig.getClient().getSslEnable() - && StrUtil.isEmpty(proxyConfig.getClient().getJksPath())) { - log.error("not found jks-path config."); - Solon.stop(); - return; + + /** + * 连接代理服务器 + */ + private void connectProxyServer() throws InterruptedException { + bootstrap.connect(proxyConfig.getClient().getServerIp(), proxyConfig.getClient().getServerPort()) + .addListener(new ChannelFutureListener() { + + @Override + public void operationComplete(ChannelFuture future) throws Exception { + if (future.isSuccess()) { + channel = future.channel(); + // 连接成功，向服务器发送客户端认证信息（licenseKey） + ProxyUtil.setCmdChannel(future.channel()); + future.channel().writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey())); + log.info("连接代理服务成功. channelId:{}", future.channel().id().asLongText()); + +// reconnectServiceEnable = true; + reconnectCount = 0; + } else { + log.info("连接代理服务失败!"); + } + } + }).sync(); } - if (StrUtil.isEmpty(proxyConfig.getClient().getLicenseKey())) { - log.error("not found license-key config."); - Solon.stop(); - return; + + private ChannelHandler createSslHandler() { + try { + InputStream jksInputStream = FileUtil.getInputStream(proxyConfig.getClient().getJksPath()); + + SSLContext clientContext = SSLContext.getInstance("TLS"); + final KeyStore ks = KeyStore.getInstance("JKS"); + ks.load(jksInputStream, proxyConfig.getClient().getKeyStorePassword().toCharArray()); + TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(ks); + TrustManager[] trustManagers = tmf.getTrustManagers(); + clientContext.init(null, trustManagers, null); + + SSLEngine sslEngine = clientContext.createSSLEngine(); + sslEngine.setUseClientMode(true); + + return new SslHandler(sslEngine); + } catch (Exception e) { + log.error("创建SSL处理器失败", e); + e.printStackTrace(); + } + return null; } - if (null == channel || !channel.isActive()) { + + protected synchronized void reconnect() { +// if (!reconnectServiceEnable) { +// return; +// } + if (null != channel) { + if (channel.isActive()) { + return; + } + channel.close(); + } + + log.info("客户端重连 seq:{}", ++reconnectCount); try { connectProxyServer(); } catch (Exception e) { - log.error("client start error", e); + log.error("重连异常", e); } - } else { - channel.writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey())); } - } - - /** - * 连接代理服务器 - */ - private void connectProxyServer() throws InterruptedException { - bootstrap.connect(proxyConfig.getClient().getServerIp(), proxyConfig.getClient().getServerPort()) - .addListener(new ChannelFutureListener() { - - @Override - public void operationComplete(ChannelFuture future) throws Exception { - if (future.isSuccess()) { - channel = future.channel(); - // 连接成功，向服务器发送客户端认证信息（licenseKey） - ProxyUtil.setCmdChannel(future.channel()); - future.channel().writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey())); - log.info("连接代理服务成功. channelId:{}", future.channel().id().asLongText()); - - reconnectServiceEnable = true; - reconnectCount = 0; - } else { - log.info("连接代理服务失败!"); - } - } - }).sync(); - } - - private ChannelHandler createSslHandler() { - try { - InputStream jksInputStream = FileUtil.getInputStream(proxyConfig.getClient().getJksPath()); - - SSLContext clientContext = SSLContext.getInstance("TLS"); - final KeyStore ks = KeyStore.getInstance("JKS"); - ks.load(jksInputStream, proxyConfig.getClient().getKeyStorePassword().toCharArray()); - TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); - tmf.init(ks); - TrustManager[] trustManagers = tmf.getTrustManagers(); - clientContext.init(null, trustManagers, null); - - SSLEngine sslEngine = clientContext.createSSLEngine(); - sslEngine.setUseClientMode(true); - - return new SslHandler(sslEngine); - } catch (Exception e) { - log.error("创建SSL处理器失败", e); - e.printStackTrace(); - } - return null; - } - - protected synchronized void reconnect() { - if (!reconnectServiceEnable || null == channel) { - return; - } - if (channel.isActive()) { - return; - } - channel.close(); - log.info("客户端重连 seq:{}", ++reconnectCount); - try { - connectProxyServer(); - } catch (Exception e) { - log.error("重连异常", e); - } - } }