You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2022/06/17 06:27:53 UTC
[dubbo] branch 3.0 updated: [optimization]RestProtocol opt, share client pool among services and avoid potential memory leak. (#10023)
This is an automated email from the ASF dual-hosted git repository.
albumenj 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 ab63595800 [optimization]RestProtocol opt, share client pool among services and avoid potential memory leak. (#10023)
ab63595800 is described below
commit ab63595800cba18b8144613dba6fd66477869abe
Author: ken.lj <ke...@gmail.com>
AuthorDate: Fri Jun 17 14:27:46 2022 +0800
[optimization]RestProtocol opt, share client pool among services and avoid potential memory leak. (#10023)
* rest protocol client share
* rest protocol optimization
* reference count support
* check and clear
---
.../common/reference/ReferenceCountedResource.java | 77 +++++++++++
.../apache/dubbo/metadata/ServiceNameMapping.java | 4 +-
.../dubbo/rpc/protocol/AbstractProxyProtocol.java | 8 +-
.../rpc/protocol/rest/ReferenceCountedClient.java | 50 +++++++
.../dubbo/rpc/protocol/rest/RestProtocol.java | 144 +++++++++++++--------
5 files changed, 225 insertions(+), 58 deletions(-)
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/reference/ReferenceCountedResource.java b/dubbo-common/src/main/java/org/apache/dubbo/common/reference/ReferenceCountedResource.java
new file mode 100755
index 0000000000..3288a514d6
--- /dev/null
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/reference/ReferenceCountedResource.java
@@ -0,0 +1,77 @@
+/*
+ * 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.common.reference;
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+/**
+ * inspired by Netty
+ */
+public abstract class ReferenceCountedResource implements AutoCloseable {
+ private static final Logger logger = LoggerFactory.getLogger(ReferenceCountedResource.class);
+ private static final AtomicLongFieldUpdater<ReferenceCountedResource> COUNTER_UPDATER
+ = AtomicLongFieldUpdater.newUpdater(ReferenceCountedResource.class, "counter");
+
+ private volatile long counter = 1;
+
+ /**
+ * Increments the reference count by 1.
+ */
+ public final ReferenceCountedResource retain() {
+ long oldCount = COUNTER_UPDATER.getAndIncrement(this);
+ if (oldCount <= 0) {
+ COUNTER_UPDATER.getAndDecrement(this);
+ throw new AssertionError("This instance has been destroyed");
+ }
+ return this;
+ }
+
+ /**
+ * Decreases the reference count by 1 and calls {@link this#destroy} if the reference count reaches 0.
+ */
+ public final boolean release() {
+ long remainingCount = COUNTER_UPDATER.decrementAndGet(this);
+
+ if (remainingCount == 0) {
+ destroy();
+ return true;
+ } else if (remainingCount <= -1) {
+ logger.warn("This instance has been destroyed");
+ return false;
+ } else {
+ return false;
+ }
+ }
+
+ /**
+ * Useful when used together with try-with-resources pattern
+ */
+ @Override
+ public final void close() {
+ release();
+ }
+
+
+ /**
+ * This method will be invoked when counter reaches 0, override this method to destroy materials related to the specific resource.
+ */
+ protected abstract void destroy();
+
+}
diff --git a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMapping.java b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMapping.java
index 934999ba07..a8ddd6c2c2 100644
--- a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMapping.java
+++ b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMapping.java
@@ -90,9 +90,9 @@ public interface ServiceNameMapping extends Destroyable {
}
/**
- * Init the mapping data from local storage and url parameter.
+ * Init mapping from local storage and url parameter.
*
- * @return app list that current interface maps to, in sequence determined by:
+ * @return app list the current interface maps to, in sequence determined by:
* 1. PROVIDED_BY specified by user
* 2. snapshot in local file
*/
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
index bc5415f1c8..3af93813a4 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java
@@ -137,15 +137,21 @@ public abstract class AbstractProxyProtocol extends AbstractProtocol {
super.destroy();
target.destroy();
invokers.remove(this);
+ AbstractProxyProtocol.this.destroyInternal(url);
}
};
invokers.add(invoker);
return invoker;
}
+ // used to destroy unused clients and other resource
+ protected void destroyInternal(URL url) {
+ // subclass override
+ }
+
protected RpcException getRpcException(Class<?> type, URL url, Invocation invocation, Throwable e) {
RpcException re = new RpcException("Failed to invoke remote service: " + type + ", method: "
- + invocation.getMethodName() + ", cause: " + e.getMessage(), e);
+ + invocation.getMethodName() + ", cause: " + e.getMessage(), e);
re.setCode(getErrorCode(e));
return re;
}
diff --git a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/ReferenceCountedClient.java b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/ReferenceCountedClient.java
new file mode 100644
index 0000000000..e3c15debe4
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/ReferenceCountedClient.java
@@ -0,0 +1,50 @@
+/*
+ * 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.rpc.protocol.rest;
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.reference.ReferenceCountedResource;
+
+import org.jboss.resteasy.client.jaxrs.ResteasyClient;
+
+public class ReferenceCountedClient extends ReferenceCountedResource {
+ private static final Logger logger = LoggerFactory.getLogger(ReferenceCountedClient.class);
+
+ private final ResteasyClient resteasyClient;
+
+ public ReferenceCountedClient(ResteasyClient resteasyClient) {
+ this.resteasyClient = resteasyClient;
+ }
+
+ public ResteasyClient getClient() {
+ return resteasyClient;
+ }
+
+ public boolean isDestroyed() {
+ return resteasyClient.isClosed();
+ }
+
+ @Override
+ protected void destroy() {
+ try {
+ resteasyClient.close();
+ } catch (Exception e) {
+ logger.error("Close resteasy client error", e);
+ }
+ }
+}
diff --git a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
index 9deed4bd92..267fcb18f0 100644
--- a/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
+++ b/dubbo-rpc/dubbo-rpc-rest/src/main/java/org/apache/dubbo/rpc/protocol/rest/RestProtocol.java
@@ -43,10 +43,8 @@ import org.jboss.resteasy.util.GetRestful;
import javax.servlet.ServletContext;
import javax.ws.rs.ProcessingException;
import javax.ws.rs.WebApplicationException;
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
@@ -72,8 +70,7 @@ public class RestProtocol extends AbstractProxyProtocol {
private final RestServerFactory serverFactory = new RestServerFactory();
- // TODO in the future maybe we can just use a single rest client and connection manager
- private final List<ResteasyClient> clients = Collections.synchronizedList(new LinkedList<>());
+ private final Map<String, ReferenceCountedClient> clients = new ConcurrentHashMap<>();
private volatile ConnectionMonitor connectionMonitor;
@@ -136,65 +133,77 @@ public class RestProtocol extends AbstractProxyProtocol {
@Override
protected <T> T doRefer(Class<T> serviceType, URL url) throws RpcException {
+ ReferenceCountedClient referenceCountedClient = clients.computeIfAbsent(url.getAddress(), _key -> {
+ // TODO more configs to add
+ return createReferenceCountedClient(url);
+ });
+
+ if (referenceCountedClient.isDestroyed()) {
+ referenceCountedClient = createReferenceCountedClient(url);
+ clients.put(url.getAddress(), referenceCountedClient);
+ }
+ referenceCountedClient.retain();
+
+ ResteasyClient resteasyClient = referenceCountedClient.getClient();
+ for (String clazz : COMMA_SPLIT_PATTERN.split(url.getParameter(EXTENSION_KEY, ""))) {
+ if (!StringUtils.isEmpty(clazz)) {
+ try {
+ resteasyClient.register(Thread.currentThread().getContextClassLoader().loadClass(clazz.trim()));
+ } catch (ClassNotFoundException e) {
+ throw new RpcException("Error loading JAX-RS extension class: " + clazz.trim(), e);
+ }
+ }
+ }
+
+ // TODO protocol
+ ResteasyWebTarget target = resteasyClient.target("http://" + url.getAddress() + "/" + getContextPath(url));
+ return target.proxy(serviceType);
+ }
- // TODO more configs to add
+ private ReferenceCountedClient createReferenceCountedClient(URL url) {
PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
// 20 is the default maxTotal of current PoolingClientConnectionManager
connectionManager.setMaxTotal(url.getParameter(CONNECTIONS_KEY, HTTPCLIENTCONNECTIONMANAGER_MAXTOTAL));
connectionManager.setDefaultMaxPerRoute(url.getParameter(CONNECTIONS_KEY, HTTPCLIENTCONNECTIONMANAGER_MAXPERROUTE));
-
if (connectionMonitor == null) {
connectionMonitor = new ConnectionMonitor();
connectionMonitor.start();
}
- connectionMonitor.addConnectionManager(connectionManager);
+ connectionMonitor.addConnectionManager(url.getAddress(), connectionManager);
+
RequestConfig requestConfig = RequestConfig.custom()
- .setConnectTimeout(url.getParameter(CONNECT_TIMEOUT_KEY, DEFAULT_CONNECT_TIMEOUT))
- .setSocketTimeout(url.getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT))
- .build();
+ .setConnectTimeout(url.getParameter(CONNECT_TIMEOUT_KEY, DEFAULT_CONNECT_TIMEOUT))
+ .setSocketTimeout(url.getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT))
+ .build();
SocketConfig socketConfig = SocketConfig.custom()
- .setSoKeepAlive(true)
- .setTcpNoDelay(true)
- .build();
+ .setSoKeepAlive(true)
+ .setTcpNoDelay(true)
+ .build();
CloseableHttpClient httpClient = HttpClientBuilder.create()
- .setConnectionManager(connectionManager)
- .setKeepAliveStrategy((response, context) -> {
- HeaderElementIterator it = new BasicHeaderElementIterator(response.headerIterator(HTTP.CONN_KEEP_ALIVE));
- while (it.hasNext()) {
- HeaderElement he = it.nextElement();
- String param = he.getName();
- String value = he.getValue();
- if (value != null && param.equalsIgnoreCase(TIMEOUT_KEY)) {
- return Long.parseLong(value) * 1000;
- }
+ .setConnectionManager(connectionManager)
+ .setKeepAliveStrategy((response, context) -> {
+ HeaderElementIterator it = new BasicHeaderElementIterator(response.headerIterator(HTTP.CONN_KEEP_ALIVE));
+ while (it.hasNext()) {
+ HeaderElement he = it.nextElement();
+ String param = he.getName();
+ String value = he.getValue();
+ if (value != null && param.equalsIgnoreCase(TIMEOUT_KEY)) {
+ return Long.parseLong(value) * 1000;
}
- return HTTPCLIENT_KEEPALIVEDURATION;
- })
- .setDefaultRequestConfig(requestConfig)
- .setDefaultSocketConfig(socketConfig)
- .build();
+ }
+ return HTTPCLIENT_KEEPALIVEDURATION;
+ })
+ .setDefaultRequestConfig(requestConfig)
+ .setDefaultSocketConfig(socketConfig)
+ .build();
ApacheHttpClient4Engine engine = new ApacheHttpClient4Engine(httpClient/*, localContext*/);
- ResteasyClient client = new ResteasyClientBuilder().httpEngine(engine).build();
- clients.add(client);
-
- client.register(RpcContextFilter.class);
- for (String clazz : COMMA_SPLIT_PATTERN.split(url.getParameter(EXTENSION_KEY, ""))) {
- if (!StringUtils.isEmpty(clazz)) {
- try {
- client.register(Thread.currentThread().getContextClassLoader().loadClass(clazz.trim()));
- } catch (ClassNotFoundException e) {
- throw new RpcException("Error loading JAX-RS extension class: " + clazz.trim(), e);
- }
- }
- }
-
- // TODO protocol
- ResteasyWebTarget target = client.target("http://" + url.getHost() + ":" + url.getPort() + "/" + getContextPath(url));
- return target.proxy(serviceType);
+ ResteasyClient resteasyClient = new ResteasyClientBuilder().httpEngine(engine).build();
+ resteasyClient.register(RpcContextFilter.class);
+ return new ReferenceCountedClient(resteasyClient);
}
@Override
@@ -229,9 +238,10 @@ public class RestProtocol extends AbstractProxyProtocol {
if (logger.isInfoEnabled()) {
logger.info("Closing rest clients");
}
- for (ResteasyClient client : clients) {
+ for (ReferenceCountedClient client : clients.values()) {
try {
- client.close();
+ // destroy directly regardless of the current reference count.
+ client.destroy();
} catch (Throwable t) {
logger.warn("Error closing rest client", t);
}
@@ -240,9 +250,9 @@ public class RestProtocol extends AbstractProxyProtocol {
}
/**
- * getPath() will return: [contextpath + "/" +] path
- * 1. contextpath is empty if user does not set through ProtocolConfig or ProviderConfig
- * 2. path will never be empty, it's default value is the interface name.
+ * getPath() will return: [contextpath + "/" +] path
+ * 1. contextpath is empty if user does not set through ProtocolConfig or ProviderConfig
+ * 2. path will never be empty, its default value is the interface name.
*
* @return return path only if user has explicitly gave then a value.
*/
@@ -261,12 +271,28 @@ public class RestProtocol extends AbstractProxyProtocol {
}
}
+ @Override
+ protected void destroyInternal(URL url) {
+ try {
+ ReferenceCountedClient referenceCountedClient = clients.get(url.getAddress());
+ if (referenceCountedClient != null && referenceCountedClient.release()) {
+ clients.remove(url.getAddress());
+ connectionMonitor.destroyManager(url);
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to close unused resources in rest protocol. interfaceName [" + url.getServiceInterface() + "]", e);
+ }
+ }
+
protected class ConnectionMonitor extends Thread {
private volatile boolean shutdown;
- private final List<PoolingHttpClientConnectionManager> connectionManagers = Collections.synchronizedList(new LinkedList<>());
+ /**
+ * The lifecycle of {@code PoolingHttpClientConnectionManager} instance is bond with ReferenceCountedClient
+ */
+ private final Map<String, PoolingHttpClientConnectionManager> connectionManagers = new ConcurrentHashMap<>();
- public void addConnectionManager(PoolingHttpClientConnectionManager connectionManager) {
- connectionManagers.add(connectionManager);
+ public void addConnectionManager(String address, PoolingHttpClientConnectionManager connectionManager) {
+ connectionManagers.putIfAbsent(address, connectionManager);
}
@Override
@@ -275,7 +301,7 @@ public class RestProtocol extends AbstractProxyProtocol {
while (!shutdown) {
synchronized (this) {
wait(HTTPCLIENTCONNECTIONMANAGER_CLOSEWAITTIME_MS);
- for (PoolingHttpClientConnectionManager connectionManager : connectionManagers) {
+ for (PoolingHttpClientConnectionManager connectionManager : connectionManagers.values()) {
connectionManager.closeExpiredConnections();
connectionManager.closeIdleConnections(HTTPCLIENTCONNECTIONMANAGER_CLOSEIDLETIME_S, TimeUnit.SECONDS);
}
@@ -293,5 +319,13 @@ public class RestProtocol extends AbstractProxyProtocol {
notifyAll();
}
}
+
+ // destroy the connection manager of a specific address when ReferenceCountedClient is destroyed.
+ private void destroyManager(URL url) {
+ PoolingHttpClientConnectionManager connectionManager = connectionManagers.remove(url.getAddress());
+ if (connectionManager != null) {
+ connectionManager.close();
+ }
+ }
}
}