You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by gu...@apache.org on 2021/04/09 11:53:14 UTC
[dubbo] branch 3.0 updated: Simplify connection reconnect (#7526)
This is an automated email from the ASF dual-hosted git repository.
guohao pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new 31563ea Simplify connection reconnect (#7526)
31563ea is described below
commit 31563eafb746681ee75436e0c45e412b054fda0b
Author: GuoHao <gu...@gmail.com>
AuthorDate: Fri Apr 9 19:52:51 2021 +0800
Simplify connection reconnect (#7526)
---
.../org/apache/dubbo/remoting/api/Connection.java | 21 ++--
.../dubbo/remoting/api/ConnectionHandler.java | 132 ++-------------------
.../dubbo/remoting/api/ConnectionListener.java | 48 ++++++++
3 files changed, 70 insertions(+), 131 deletions(-)
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 66bebfe..24833aa 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> 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> 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<SocketChannel>() {
@@ -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 ee19b9c..47dca24 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<Boolean> 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();
- }
-
- 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);
- }
- }
+ reconnect(ctx.channel());
+ super.channelInactive(ctx);
}
- 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 0000000..a379ce0
--- /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