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

Use ExecutorService instead of event loop for Netty connection (#13904) 

* Use ExecutorService instead of event loop for Netty connection

This change addresses the issue of the event loop being blocked for an extended period, improving overall performance and responsiveness.

* fix log getConnectAddress

* applay format

* applay format

* applay format

* applay format

* applay format

* Use an independent ExecutorService

* Use ScheduledExecutorService for scheduling tasks

* delete unnecessary files

* Modify to stop ExecutorService using shutdownNow

* Modify to use ScheduledExecutor managed by FrameworkExecutorRepository

* Synchronize reconnectDuration with HeaderExchangeClient's reconnectDuration

* get framework model directly

use awaitility
This commit is contained in:
hanpen24 2024-04-07 11:31:23 +09:00 committed by GitHub
parent 40280a2c29
commit f03823eec1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 61 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractClient.java
View File

@ -21,6 +21,7 @@ import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.ChannelHandler;
@ -28,9 +29,11 @@ import org.apache.dubbo.remoting.Client;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.transport.dispatcher.ChannelHandlers;
import org.apache.dubbo.rpc.model.FrameworkModel;
import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@ -41,6 +44,11 @@ import static org.apache.dubbo.common.constants.CommonConstants.THREAD_NAME_KEY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CLOSE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CONNECT_PROVIDER;
import static org.apache.dubbo.config.Constants.CLIENT_THREAD_POOL_NAME;
import static org.apache.dubbo.remoting.Constants.HEARTBEAT_CHECK_TICK;
import static org.apache.dubbo.remoting.Constants.LEAST_HEARTBEAT_DURATION;
import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION;
import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION_KEY;
import static org.apache.dubbo.remoting.utils.UrlUtils.getIdleTimeout;
/**
* AbstractClient
@ -55,13 +63,23 @@ public abstract class AbstractClient extends AbstractEndpoint implements Client
protected volatile ExecutorService executor;
protected volatile ScheduledExecutorService connectivityExecutor;
private FrameworkModel frameworkModel;
protected long reconnectDuaration;
public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
super(url, handler);
// set default needReconnect true when channel is not connected
needReconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, true);
frameworkModel = url.getOrDefaultFrameworkModel();
initExecutor(url);
reconnectDuaration = getReconnectDuration(url);
try {
doOpen();
} catch (Throwable t) {
@ -134,6 +152,11 @@ public abstract class AbstractClient extends AbstractEndpoint implements Client
url = url.addParameter(THREAD_NAME_KEY, CLIENT_THREAD_POOL_NAME)
.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL);
executor = executorRepository.createExecutorIfAbsent(url);
connectivityExecutor = frameworkModel
.getBeanFactory()
.getBean(FrameworkExecutorRepository.class)
.getConnectivityScheduledExecutor();
}
protected static ChannelHandler wrapChannelHandler(URL url, ChannelHandler handler) {
@ -296,6 +319,25 @@ public abstract class AbstractClient extends AbstractEndpoint implements Client
}
}
private long getReconnectDuration(URL url) {
int idleTimeout = getIdleTimeout(url);
long heartbeatTimeoutTick = calculateLeastDuration(idleTimeout);
return calculateReconnectDuration(url, heartbeatTimeoutTick);
}
private long calculateLeastDuration(int time) {
if (time / HEARTBEAT_CHECK_TICK <= 0) {
return LEAST_HEARTBEAT_DURATION;
} else {
return time / HEARTBEAT_CHECK_TICK;
}
}
private long calculateReconnectDuration(URL url, long tick) {
long leastReconnectDuration = url.getParameter(LEAST_RECONNECT_DURATION_KEY, LEAST_RECONNECT_DURATION);
return Math.max(leastReconnectDuration, tick);
}
@Override
public void reconnect() throws RemotingException {
connectLock.lock();

21
dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java
View File

@ -43,7 +43,6 @@ import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoop;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.timeout.IdleStateHandler;
@ -75,6 +74,8 @@ public class NettyConnectionClient extends AbstractConnectionClient {
public static final AttributeKey<AbstractConnectionClient> CONNECTION = AttributeKey.valueOf("connection");
private AtomicBoolean isReconnecting;
public NettyConnectionClient(URL url, ChannelHandler handler) throws RemotingException {
super(url, handler);
}
@ -91,6 +92,7 @@ public class NettyConnectionClient extends AbstractConnectionClient {
this.closePromise = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
this.init = new AtomicBoolean(false);
this.increase();
this.isReconnecting = new AtomicBoolean(false);
}
@Override
@ -158,6 +160,10 @@ public class NettyConnectionClient extends AbstractConnectionClient {
@Override
protected void doConnect() throws RemotingException {
if (!isReconnecting.compareAndSet(false, true)) {
return;
}
if (isClosed()) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
@ -347,6 +353,11 @@ public class NettyConnectionClient extends AbstractConnectionClient {
@Override
public void operationComplete(ChannelFuture future) {
if (!isReconnecting.compareAndSet(true, false)) {
return;
}
if (future.isSuccess()) {
return;
}
@ -364,8 +375,8 @@ public class NettyConnectionClient extends AbstractConnectionClient {
"%s is reconnecting, attempt=%d cause=%s",
connectionClient, 0, future.cause().getMessage()));
}
final EventLoop loop = future.channel().eventLoop();
loop.schedule(
connectivityExecutor.schedule(
() -> {
try {
connectionClient.doConnect();
@ -377,8 +388,8 @@ public class NettyConnectionClient extends AbstractConnectionClient {
"Failed to connect to server: " + getConnectAddress());
}
},
1L,
TimeUnit.SECONDS);
reconnectDuaration,
TimeUnit.MILLISECONDS);
}
}
}

3
dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/ConnectionTest.java
View File

@ -28,6 +28,7 @@ import org.apache.dubbo.remoting.api.pu.DefaultPuHandler;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ModuleModel;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@ -39,6 +40,7 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT;
import static org.awaitility.Awaitility.await;
public class ConnectionTest {
@ -138,6 +140,7 @@ public class ConnectionTest {
nettyPortUnificationServer.bind();
// auto reconnect
await().atMost(Duration.ofSeconds(100)).until(() -> connectionClient.isAvailable());
Assertions.assertTrue(connectionClient.isAvailable());
connectionClient.close();