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

Support client gracefully shutdown (#12504) 

* Support client gracefully shutdown

* fix check

* Fix init
This commit is contained in:
Albumen Kevin 2023-06-12 20:27:55 +08:00 committed by GitHub
parent fc6fbd60f9
commit e92f85d980
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 46 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
View File

@ -134,6 +134,8 @@ public interface Constants {
int DEFAULT_RECONNECT_PERIOD = 2000;
String CHANNEL_SHUTDOWN_TIMEOUT_KEY = "channel.shutdown.timeout";
String SEND_RECONNECT_KEY = "send.reconnect";
String CHECK_KEY = "check";

33
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.remoting.exchange.support;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.GlobalResourceInitializer;
import org.apache.dubbo.common.serialize.SerializationException;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.common.timer.Timeout;
import org.apache.dubbo.common.timer.Timer;
@ -26,7 +27,6 @@ import org.apache.dubbo.common.timer.TimerTask;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.common.serialize.SerializationException;
import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.remoting.exchange.Response;
@ -150,22 +150,39 @@ public class DefaultFuture extends CompletableFuture<Object> {
*
* @param channel channel to close
*/
public static void closeChannel(Channel channel) {
public static void closeChannel(Channel channel, long timeout) {
long deadline = timeout > 0 ? System.currentTimeMillis() + timeout : 0;
for (Map.Entry<Long, Channel> entry : CHANNELS.entrySet()) {
if (channel.equals(entry.getValue())) {
DefaultFuture future = getFuture(entry.getKey());
if (future != null && !future.isDone()) {
Response disconnectResponse = new Response(future.getId());
disconnectResponse.setStatus(Response.CHANNEL_INACTIVE);
disconnectResponse.setErrorMessage("Channel " +
channel + " is inactive. Directly return the unFinished request : " +
(logger.isDebugEnabled() ? future.getRequest() : future.getRequest().copyWithoutData()));
DefaultFuture.received(channel, disconnectResponse);
long restTime = deadline - System.currentTimeMillis();
if (restTime > 0) {
try {
future.get(restTime, TimeUnit.MILLISECONDS);
} catch (java.util.concurrent.TimeoutException ignore) {
logger.warn(PROTOCOL_TIMEOUT_SERVER, "", "",
"Trying to close channel " + channel + ", but response is not received in "
+ timeout + "ms, and the request id is " + future.id);
} catch (Throwable ignore) {}
}
if (!future.isDone()) {
respInactive(channel, future);
}
}
}
}
}
private static void respInactive(Channel channel, DefaultFuture future) {
Response disconnectResponse = new Response(future.getId());
disconnectResponse.setStatus(Response.CHANNEL_INACTIVE);
disconnectResponse.setErrorMessage("Channel " +
channel + " is inactive. Directly return the unFinished request : " +
(logger.isDebugEnabled() ? future.getRequest() : future.getRequest().copyWithoutData()));
DefaultFuture.received(channel, disconnectResponse);
}
public static void received(Channel channel, Response response) {
received(channel, response, false);
}

10
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java
View File

@ -18,6 +18,7 @@ package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.remoting.Channel;
@ -30,6 +31,7 @@ import org.apache.dubbo.remoting.exchange.Response;
import org.apache.dubbo.remoting.exchange.support.DefaultFuture;
import java.net.InetSocketAddress;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@ -48,6 +50,8 @@ final class HeaderExchangeChannel implements ExchangeChannel {
private final Channel channel;
private final long shutdownTimeout;
private volatile boolean closed = false;
HeaderExchangeChannel(Channel channel) {
@ -55,6 +59,10 @@ final class HeaderExchangeChannel implements ExchangeChannel {
throw new IllegalArgumentException("channel == null");
}
this.channel = channel;
this.shutdownTimeout = Optional.ofNullable(channel.getUrl())
.map(URL::getOrDefaultApplicationModel)
.map(ConfigurationUtils::getServerShutdownTimeout)
.orElse(DEFAULT_TIMEOUT);
}
static HeaderExchangeChannel getOrAddChannel(Channel ch) {
@ -159,7 +167,7 @@ final class HeaderExchangeChannel implements ExchangeChannel {
closed = true;
try {
// graceful close
DefaultFuture.closeChannel(channel);
DefaultFuture.closeChannel(channel, shutdownTimeout);
} catch (Exception e) {
logger.warn(TRANSPORT_FAILED_CLOSE, "", "", e.getMessage(), e);
}

10
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.NetUtils;
@ -132,6 +133,8 @@ public class HeaderExchangeHandler implements ChannelHandlerDelegate {
public void connected(Channel channel) throws RemotingException {
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
handler.connected(exchangeChannel);
channel.setAttribute(Constants.CHANNEL_SHUTDOWN_TIMEOUT_KEY,
ConfigurationUtils.getServerShutdownTimeout(channel.getUrl().getOrDefaultApplicationModel()));
}
@Override
@ -140,7 +143,12 @@ public class HeaderExchangeHandler implements ChannelHandlerDelegate {
try {
handler.disconnected(exchangeChannel);
} finally {
DefaultFuture.closeChannel(channel);
int shutdownTimeout = 0;
Object timeoutObj = channel.getAttribute(Constants.CHANNEL_SHUTDOWN_TIMEOUT_KEY);
if (timeoutObj instanceof Integer) {
shutdownTimeout = (Integer) timeoutObj;
}
DefaultFuture.closeChannel(channel, shutdownTimeout);
HeaderExchangeChannel.removeChannel(channel);
}
}

2
dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/DefaultFutureTest.java
View File

@ -167,7 +167,7 @@ class DefaultFutureTest {
ExecutorService executor = ExtensionLoader.getExtensionLoader(ExecutorRepository.class)
.getDefaultExtension().createExecutorIfAbsent(URL.valueOf("dubbo://127.0.0.1:23456"));
DefaultFuture.newFuture(channel, request, 1000, executor);
DefaultFuture.closeChannel(channel);
DefaultFuture.closeChannel(channel, 0);
Assertions.assertFalse(executor.isTerminated());
}