From 8600e9825b235b665849d2b15b0fa02b2ac533ac Mon Sep 17 00:00:00 2001 From: Jason Huang Date: Mon, 26 Jun 2017 19:24:15 +0800 Subject: [PATCH] =?UTF-8?q?fix=20bug=EF=BC=9A=20=E7=8E=B0=E5=AE=9E?= =?UTF-8?q?=E8=BF=90=E7=94=A8=E4=B8=AD=E5=87=BA=E7=8E=B0=EF=BC=9Aclient?= =?UTF-8?q?=E5=B7=B2=E7=BB=8F=E4=B8=8D=E5=AD=98=E5=9C=A8=E4=BD=86server?= =?UTF-8?q?=E7=AB=AF=E7=9A=84=E8=BF=9E=E6=8E=A5=E4=BB=8D=E7=84=B6=E6=98=AF?= =?UTF-8?q?=E5=AD=98=E6=B4=BB=E7=8A=B6=E6=80=81=E3=80=82=20=E5=9C=A8?= =?UTF-8?q?=E6=AD=A4=E7=A7=8D=E7=BD=91=E7=BB=9C=E5=BC=82=E5=B8=B8=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E6=97=B6=EF=BC=8Cstop=20canal=20server=E4=BC=9A?= =?UTF-8?q?=E4=B8=80=E7=9B=B4hang=E4=BD=8F=E3=80=82=20=E5=8E=9F=E5=9B=A0?= =?UTF-8?q?=EF=BC=9A=E5=BC=82=E5=B8=B8=E7=BD=91=E7=BB=9C=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E7=94=B1=E4=BA=8E=E6=97=A0=E6=95=B0=E6=8D=AE=E4=BC=A0=E8=BE=93?= =?UTF-8?q?=EF=BC=8C=E5=AF=BC=E8=87=B4=20io=20worker=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E4=B8=80=E7=9B=B4loop=20selector=E8=80=8C=E6=97=A0=E6=B3=95?= =?UTF-8?q?=E9=80=80=E5=87=BA=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 解决方案：1. 设置keepalive 由os 监控。 2. 设置netty idle 机制，定时检测并主动close socket。(fix 时间单位bug) 3. 每次新创建socket channel时存储在一个容器内，在stop canal server时，从容器内主动释放所有socket channel。 --- .../server/netty/CanalServerWithNetty.java | 25 ++++++++++++++++++- .../handler/ClientAuthenticationHandler.java | 5 +++- .../HandshakeInitializationHandler.java | 12 +++++++++ 3 files changed, 40 insertions(+), 2 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java index 873bea70..57289f7a 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java @@ -9,6 +9,8 @@ import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.ChannelPipeline; import org.jboss.netty.channel.ChannelPipelineFactory; import org.jboss.netty.channel.Channels; +import org.jboss.netty.channel.group.ChannelGroup; +import org.jboss.netty.channel.group.DefaultChannelGroup; import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; @@ -32,6 +34,7 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana private int port; private Channel serverChannel = null; private ServerBootstrap bootstrap = null; + private ChannelGroup childGroups = null; //socket channel container, used to close sockets explicitly. private static class SingletonHolder { @@ -40,6 +43,7 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana private CanalServerWithNetty(){ this.embeddedServer = CanalServerWithEmbedded.instance(); + this.childGroups = new DefaultChannelGroup(); } public static CanalServerWithNetty instance() { @@ -55,6 +59,19 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool())); + /* + * enable keep-alive mechanism, handle abnormal network connection scenarios on OS level. + * the threshold parameters are depended on OS. + * e.g. On Linux: + * net.ipv4.tcp_keepalive_time = 300 + * net.ipv4.tcp_keepalive_probes = 2 + * net.ipv4.tcp_keepalive_intvl = 30 + */ + bootstrap.setOption("child.keepAlive", true); + /* + * optional parameter. + */ + bootstrap.setOption("child.tcpNoDelay", true); // 构造对应的pipeline bootstrap.setPipelineFactory(new ChannelPipelineFactory() { @@ -62,7 +79,8 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana public ChannelPipeline getPipeline() throws Exception { ChannelPipeline pipelines = Channels.pipeline(); pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder()); - pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler()); + //support to maintain child socket channel. + pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler(childGroups)); pipelines.addLast(ClientAuthenticationHandler.class.getName(), new ClientAuthenticationHandler(embeddedServer)); @@ -87,6 +105,11 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana this.serverChannel.close().awaitUninterruptibly(1000); } + //close sockets explicitly to reduce socket channel hung in complicated network environment. + if (this.childGroups != null) { + this.childGroups.close().awaitUninterruptibly(5000); + } + if (this.bootstrap != null) { this.bootstrap.releaseExternalResources(); } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java index 0656e8aa..73bcd5d2 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.server.netty.handler; +import java.util.concurrent.TimeUnit; + import org.apache.commons.lang.StringUtils; import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.channel.ChannelFuture; @@ -87,10 +89,11 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler { if (clientAuth.getNetWriteTimeout() > 0) { writeTimeout = clientAuth.getNetWriteTimeout(); } + //fix bug: soTimeout parameter's unit from connector is millseconds. IdleStateHandler idleStateHandler = new IdleStateHandler(NettyUtils.hashedWheelTimer, readTimeout, writeTimeout, - 0); + 0, TimeUnit.MILLISECONDS); ctx.getPipeline().addBefore(SessionHandler.class.getName(), IdleStateHandler.class.getName(), idleStateHandler); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/HandshakeInitializationHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/HandshakeInitializationHandler.java index b65ff2a1..438e508f 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/HandshakeInitializationHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/HandshakeInitializationHandler.java @@ -3,6 +3,7 @@ package com.alibaba.otter.canal.server.netty.handler; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelStateEvent; import org.jboss.netty.channel.SimpleChannelHandler; +import org.jboss.netty.channel.group.ChannelGroup; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,10 +19,21 @@ import com.alibaba.otter.canal.server.netty.NettyUtils; * @version 1.0.0 */ public class HandshakeInitializationHandler extends SimpleChannelHandler { + //support to maintain socket channel. + private ChannelGroup childGroups; + + public HandshakeInitializationHandler(ChannelGroup childGroups) { + this.childGroups = childGroups; + } private static final Logger logger = LoggerFactory.getLogger(HandshakeInitializationHandler.class); public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { + //add new socket channel in channel container, used to manage sockets. + if(childGroups != null) { + childGroups.add(ctx.getChannel()); + } + byte[] body = Packet.newBuilder() .setType(CanalPacket.PacketType.HANDSHAKE) .setBody(Handshake.newBuilder().build().toByteString())