natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Simplify connection reconnect (#7526)

This commit is contained in:
GuoHao 2021-04-09 19:52:51 +08:00 committed by GitHub
parent be54ec206a
commit 31563eafb7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 70 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java
View File

@ -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));
}

132
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java
View File

@ -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();
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);
}
}

48
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java Normal file
View File

@ -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);
}
}
}