From 31563eafb746681ee75436e0c45e412b054fda0b Mon Sep 17 00:00:00 2001 From: GuoHao Date: Fri, 9 Apr 2021 19:52:51 +0800 Subject: [PATCH] Simplify connection reconnect (#7526) --- .../apache/dubbo/remoting/api/Connection.java | 21 ++- .../dubbo/remoting/api/ConnectionHandler.java | 132 ++---------------- .../remoting/api/ConnectionListener.java | 48 +++++++ 3 files changed, 70 insertions(+), 131 deletions(-) create mode 100644 dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java index 66bebfe456..24833aab80 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java @@ -21,7 +21,6 @@ import org.apache.dubbo.common.extension.ExtensionLoader; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.ExecutorUtil; -import org.apache.dubbo.common.utils.NamedThreadFactory; import org.apache.dubbo.common.utils.NetUtils; import org.apache.dubbo.remoting.Constants; import org.apache.dubbo.remoting.RemotingException; @@ -33,20 +32,16 @@ import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; -import io.netty.channel.ChannelPromise; import io.netty.channel.socket.SocketChannel; import io.netty.util.AbstractReferenceCounted; import io.netty.util.AttributeKey; -import io.netty.util.HashedWheelTimer; import io.netty.util.ReferenceCountUtil; import io.netty.util.ReferenceCounted; -import io.netty.util.Timer; import io.netty.util.concurrent.DefaultPromise; import io.netty.util.concurrent.GlobalEventExecutor; import io.netty.util.concurrent.Promise; import java.net.InetSocketAddress; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -56,8 +51,6 @@ import static org.apache.dubbo.remoting.api.NettyEventLoopFactory.socketChannelC public class Connection extends AbstractReferenceCounted implements ReferenceCounted { - public static final Timer TIMER = new HashedWheelTimer( - new NamedThreadFactory("dubbo-network-timer", true), 30, TimeUnit.MILLISECONDS); public static final AttributeKey CONNECTION = AttributeKey.valueOf("connection"); private static final Logger logger = LoggerFactory.getLogger(Connection.class); private final URL url; @@ -67,7 +60,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou private final InetSocketAddress remote; private final AtomicBoolean closed = new AtomicBoolean(false); private final AtomicReference channel = new AtomicReference<>(); - private final ChannelPromise initPromise; + private final ChannelFuture initPromise; public Connection(URL url) { url = ExecutorUtil.setThreadName(url, "DubboClientHandler"); @@ -89,7 +82,10 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou } - public ChannelPromise connect() { + public ChannelFuture connect() { + if (isClosed()) { + return null; + } final Bootstrap bootstrap = new Bootstrap(); bootstrap.group(NettyEventLoopFactory.NIO_EVENT_LOOP_GROUP) .option(ChannelOption.SO_KEEPALIVE, true) @@ -98,7 +94,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou .remoteAddress(getConnectAddress()) .channel(socketChannelClass()); - final ConnectionHandler connectionHandler = new ConnectionHandler(this, bootstrap, TIMER); + final ConnectionHandler connectionHandler = new ConnectionHandler(this); bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout); bootstrap.handler(new ChannelInitializer() { @@ -116,7 +112,9 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou // TODO support Socks5 } }); - return connectionHandler.connect(); + final ChannelFuture promise = bootstrap.connect(); + promise.addListener(new ConnectionListener(this)); + return promise; } public Channel getChannel() { @@ -139,7 +137,6 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou public void onConnected(Channel channel) { this.channel.set(channel); channel.attr(CONNECTION).set(this); - this.initPromise.trySuccess(); if (logger.isInfoEnabled()) { logger.info(String.format("Connection:%s connected ", this)); } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java index ee19b9c7d3..47dca2489e 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java @@ -19,51 +19,23 @@ package org.apache.dubbo.remoting.api; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; -import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; -import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.channel.ChannelPromise; -import io.netty.channel.DefaultChannelPromise; +import io.netty.channel.EventLoop; import io.netty.util.AttributeKey; -import io.netty.util.Timer; -import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @ChannelHandler.Sharable public class ConnectionHandler extends ChannelInboundHandlerAdapter { private static final Logger log = LoggerFactory.getLogger(ConnectionHandler.class); - - private static final int MIN_FAST_RECONNECT_INTERVAL = 4000; - private static final int BACKOFF_CAP = 13; private static final AttributeKey GO_AWAY_KEY = AttributeKey.valueOf("dubbo_channel_goaway"); - private final Timer timer; - private final Bootstrap bootstrap; private final Connection connection; - private final Semaphore permit = new Semaphore(1); - private volatile long lastReconnect; - public ConnectionHandler(Connection connection, Bootstrap bootstrap, Timer timer) { + public ConnectionHandler(Connection connection) { this.connection = connection; - this.bootstrap = bootstrap; - this.timer = timer; - - } - - public ChannelPromise connect() { - final ChannelFuture init = bootstrap.connect(); - final DefaultChannelPromise promise = new DefaultChannelPromise(init.channel()); - init.addListener(future -> { - if (!future.isSuccess()) { - promise.tryFailure(future.cause()); - tryReconnect(connection); - } - }); - return promise; } public void onGoAway(Channel channel) { @@ -71,15 +43,16 @@ public class ConnectionHandler extends ChannelInboundHandlerAdapter { if (connection != null) { connection.onGoaway(channel); } - tryReconnect(connection); + reconnect(channel); } @Override - public void channelActive(ChannelHandlerContext ctx) throws Exception { + public void channelActive(ChannelHandlerContext ctx) { ctx.fireChannelActive(); - final Connection connection = Connection.getConnectionFromChannel(ctx.channel()); - if (connection != null) { + if (!connection.isClosed()) { connection.onConnected(ctx.channel()); + } else { + ctx.close(); } } @@ -90,97 +63,18 @@ public class ConnectionHandler extends ChannelInboundHandlerAdapter { ctx.close(); } - public boolean shouldFastReconnect() { - final long period = System.currentTimeMillis() - lastReconnect; - return period > MIN_FAST_RECONNECT_INTERVAL; - } - - - private boolean isGoAway(Channel channel) { - return Boolean.TRUE.equals(channel.attr(GO_AWAY_KEY).get()); - } - @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - // Reconnect event will be triggered by Connection.init(); - if (isGoAway(ctx.channel())) { - ctx.fireChannelInactive(); - return; - } - if (connection.getChannel() == null || connection.getChannel().equals(ctx.channel())) { - tryReconnect(connection); - } - ctx.fireChannelInactive(); + reconnect(ctx.channel()); + super.channelInactive(ctx); } - private void tryReconnect(Connection connection) { - if (connection != null && !connection.isClosed()) { - if (shouldFastReconnect()) { - if (log.isInfoEnabled()) { - log.info(String.format("Connection %s inactive, schedule fast reconnect", connection)); - } - reconnect(connection, 4); - } else { - if (log.isInfoEnabled()) { - log.info(String.format("Connection %s inactive, schedule normal reconnect", connection)); - } - reconnect(connection, BACKOFF_CAP); - } - } - } - - private void reconnect(final Connection connection, final int attempts) { - this.lastReconnect = System.currentTimeMillis(); - - int timeout = 2 << attempts; - if (bootstrap.config().group().isShuttingDown()) { - return; - } - - int nextAttempt = Math.min(BACKOFF_CAP, attempts + 1); - if (permit.tryAcquire()) { - timer.newTimeout(timeout1 -> tryReconnect(connection, nextAttempt), timeout, TimeUnit.MILLISECONDS); - } - } - - - private void tryReconnect(final Connection connection, final int nextAttempt) { - permit.release(); - - if (connection.isClosed() || bootstrap.config().group().isShuttingDown()) { - return; - } + private void reconnect(Channel channel) { if (log.isInfoEnabled()) { - log.info(String.format("Connection %s is reconnecting, attempt=%d", connection, nextAttempt)); + log.info(String.format("Connection %s is reconnecting, attempt=%d", connection, 1)); } - - bootstrap.connect().addListener((ChannelFutureListener) future -> { - if (connection.isClosed() || bootstrap.config().group().isShuttingDown()) { - if (future.isSuccess()) { - Channel ch = future.channel(); - Connection con = Connection.getConnectionFromChannel(ch); - if (con != null) { - con.close(); - } - } - return; - } - - if (future.isSuccess()) { - final Channel channel = future.channel(); - if (!connection.isClosed()) { - if (connection.getChannel() == null || !connection.getChannel().isActive()) { - connection.onConnected(channel); - } else { - channel.close(); - } - } else { - channel.close(); - } - } else { - reconnect(connection, nextAttempt); - } - }); + final EventLoop eventLoop = channel.eventLoop(); + eventLoop.schedule(connection::connect, 1, TimeUnit.SECONDS); } } \ No newline at end of file diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java new file mode 100644 index 0000000000..a379ce00bf --- /dev/null +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dubbo.remoting.api; + +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; + +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.EventLoop; + +import java.util.concurrent.TimeUnit; + +public class ConnectionListener implements ChannelFutureListener { + private static final Logger log = LoggerFactory.getLogger(ConnectionListener.class); + + private final Connection connection; + + public ConnectionListener(Connection connection) { + this.connection = connection; + } + + @Override + public void operationComplete(ChannelFuture channelFuture) { + if (!channelFuture.isSuccess()) { + if (log.isInfoEnabled()) { + log.info(String.format("Connection %s is reconnecting, attempt=%d", connection, 0)); + } + final EventLoop loop = channelFuture.channel().eventLoop(); + loop.schedule((Runnable) connection::connect, 1L, TimeUnit.SECONDS); + } + } +} \ No newline at end of file