From 4d0f9f4f4833317bfde6d73126e5dc386208ebdc Mon Sep 17 00:00:00 2001 From: stone lion Date: Fri, 12 Aug 2022 14:38:15 +0800 Subject: [PATCH 1/3] refactor: inspect protocol classes (#10444) --- .../rpc/protocol/dubbo/DubboProtocol.java | 69 +++++++------------ .../dubbo/LazyConnectExchangeClient.java | 2 +- .../dubbo/ReferenceCountExchangeClient.java | 6 +- 3 files changed, 28 insertions(+), 49 deletions(-) diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java index 2d35784a05..b2ccfbe2ac 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java @@ -48,10 +48,12 @@ import org.apache.dubbo.rpc.protocol.AbstractProtocol; import java.net.InetSocketAddress; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -406,35 +408,28 @@ public class DubboProtocol extends AbstractProtocol { } private ExchangeClient[] getClients(URL url) { - // whether to share connection - - boolean useShareConnect = false; - int connections = url.getParameter(CONNECTIONS_KEY, 0); - List shareClients = null; + // whether to share connection // if not configured, connection is shared, otherwise, one connection for one service if (connections == 0) { - useShareConnect = true; - /* * The xml configuration should have a higher priority than properties. */ - String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null); - connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigurationUtils.getProperty(url.getOrDefaultApplicationModel(), SHARE_CONNECTIONS_KEY, - DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr); - shareClients = getSharedClient(url, connections); + String shareConnectionsStr = StringUtils.isBlank(url.getParameter(SHARE_CONNECTIONS_KEY, (String) null)) + ? ConfigurationUtils.getProperty(url.getOrDefaultApplicationModel(), SHARE_CONNECTIONS_KEY, DEFAULT_SHARE_CONNECTIONS) + : url.getParameter(SHARE_CONNECTIONS_KEY, (String) null); + connections = Integer.parseInt(shareConnectionsStr); + + List shareClients = getSharedClient(url, connections); + ExchangeClient[] clients = new ExchangeClient[connections]; + Arrays.setAll(clients, shareClients::get); + return clients; } ExchangeClient[] clients = new ExchangeClient[connections]; for (int i = 0; i < clients.length; i++) { - if (useShareConnect) { - clients[i] = shareClients.get(i); - - } else { - clients[i] = initClient(url); - } + clients[i] = initClient(url); } - return clients; } @@ -461,6 +456,7 @@ public class DubboProtocol extends AbstractProtocol { synchronized (referenceClientMap) { for (; ; ) { + // guarantee just one thread in loading condition. And Other is waiting It had finished. clients = referenceClientMap.get(key); if (clients instanceof List) { @@ -513,7 +509,6 @@ public class DubboProtocol extends AbstractProtocol { } } return typedClients; - } /** @@ -527,14 +522,10 @@ public class DubboProtocol extends AbstractProtocol { return false; } - for (ReferenceCountExchangeClient referenceCountExchangeClient : referenceCountExchangeClients) { - // As long as one client is not available, you need to replace the unavailable client with the available one. - if (referenceCountExchangeClient == null || referenceCountExchangeClient.getCount() <= 0 || referenceCountExchangeClient.isClosed()) { - return false; - } - } - - return true; + // As long as one client is not available, you need to replace the unavailable client with the available one. + return referenceCountExchangeClients.stream() + .noneMatch(referenceCountExchangeClient -> referenceCountExchangeClient == null + || referenceCountExchangeClient.getCount() <= 0 || referenceCountExchangeClient.isClosed()); } /** @@ -546,12 +537,9 @@ public class DubboProtocol extends AbstractProtocol { if (CollectionUtils.isEmpty(referenceCountExchangeClients)) { return; } - - for (ReferenceCountExchangeClient referenceCountExchangeClient : referenceCountExchangeClients) { - if (referenceCountExchangeClient != null) { - referenceCountExchangeClient.incrementAndGetCount(); - } - } + referenceCountExchangeClients.stream() + .filter(Objects::nonNull) + .forEach(ReferenceCountExchangeClient::incrementAndGetCount); } /** @@ -592,8 +580,7 @@ public class DubboProtocol extends AbstractProtocol { * @param url */ private ExchangeClient initClient(URL url) { - - /** + /* * Instance of url is InstanceAddressURL, so addParameter actually adds parameters into ServiceInstance, * which means params are shared among different services. Since client is shared among services this is currently not a problem. */ @@ -605,7 +592,6 @@ public class DubboProtocol extends AbstractProtocol { " supported client type is " + StringUtils.join(url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).getSupportedExtensions(), " ")); } - ExchangeClient client; try { // Replace InstanceAddressURL with ServiceConfigURL. url = new ServiceConfigURL(DubboCodec.NAME, url.getUsername(), url.getPassword(), url.getHost(), url.getPort(), url.getPath(), url.getAllParameters()); @@ -614,17 +600,12 @@ public class DubboProtocol extends AbstractProtocol { url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT)); // connection should be lazy - if (url.getParameter(LAZY_CONNECT_KEY, false)) { - client = new LazyConnectExchangeClient(url, requestHandler); - } else { - client = Exchangers.connect(url, requestHandler); - } - + return url.getParameter(LAZY_CONNECT_KEY, false) + ? new LazyConnectExchangeClient(url, requestHandler) + : Exchangers.connect(url, requestHandler); } catch (RemotingException e) { throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e); } - - return client; } @Override diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java index fae3bf7295..7e2d5332c8 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java @@ -51,7 +51,7 @@ final class LazyConnectExchangeClient implements ExchangeClient { private final URL url; private final ExchangeHandler requestHandler; private final Lock connectLock = new ReentrantLock(); - private final int warningPeriod = 5000; + private static final int warningPeriod = 5000; private final boolean needReconnect; private volatile ExchangeClient client; private final AtomicLong warningCount = new AtomicLong(0); diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java index 2a904b00d7..b0a8f08e2f 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java @@ -43,7 +43,7 @@ final class ReferenceCountExchangeClient implements ExchangeClient { private final URL url; private final AtomicInteger referenceCount = new AtomicInteger(0); private final AtomicInteger disconnectCount = new AtomicInteger(0); - private final Integer warningPeriod = 50; + private static final Integer warningPeriod = 50; private ExchangeClient client; private int shutdownWaitTime = DEFAULT_SERVER_SHUTDOWN_TIMEOUT; @@ -202,9 +202,7 @@ final class ReferenceCountExchangeClient implements ExchangeClient { logger.warn(url.getAddress() + " " + url.getServiceKey() + " safe guard client , should not be called ,must have a bug."); } - /** - * the order of judgment in the if statement cannot be changed. - */ + // the order of judgment in the if statement cannot be changed. if (!(client instanceof LazyConnectExchangeClient)) { client = new LazyConnectExchangeClient(url, client.getExchangeHandler()); } From 01deaf9a0d0c45d89c3c6a99fa6c36bb16f58b6f Mon Sep 17 00:00:00 2001 From: Tianjc <931706361@qq.com> Date: Fri, 12 Aug 2022 14:38:38 +0800 Subject: [PATCH 2/3] Update MemoryStatusChecker.java (#10442) Alarm when spare memory < 2M,not <2K --- .../apache/dubbo/common/status/support/MemoryStatusChecker.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/status/support/MemoryStatusChecker.java b/dubbo-common/src/main/java/org/apache/dubbo/common/status/support/MemoryStatusChecker.java index 0f13000872..411fa2f3d9 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/status/support/MemoryStatusChecker.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/status/support/MemoryStatusChecker.java @@ -32,7 +32,7 @@ public class MemoryStatusChecker implements StatusChecker { long freeMemory = runtime.freeMemory(); long totalMemory = runtime.totalMemory(); long maxMemory = runtime.maxMemory(); - boolean ok = (maxMemory - (totalMemory - freeMemory) > 2048); // Alarm when spare memory < 2M + boolean ok = (maxMemory - (totalMemory - freeMemory) > 2*1024*1024); // Alarm when spare memory < 2M String msg = "max:" + (maxMemory / 1024 / 1024) + "M,total:" + (totalMemory / 1024 / 1024) + "M,used:" + ((totalMemory / 1024 / 1024) - (freeMemory / 1024 / 1024)) + "M,free:" + (freeMemory / 1024 / 1024) + "M"; return new Status(ok ? Status.Level.OK : Status.Level.WARN, msg); From b3f533d726953d0f3a97c207d1555bd4047eb165 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=BC=E5=8D=8E?= <43363120+BurningCN@users.noreply.github.com> Date: Fri, 12 Aug 2022 15:35:18 +0800 Subject: [PATCH 3/3] Optimize the logic of zkDataListener (#10445) --- .../TreePathDynamicConfiguration.java | 4 +- .../file/FileSystemDynamicConfiguration.java | 2 +- .../support/zookeeper/CacheListener.java | 96 +++++-------------- .../zookeeper/ZookeeperDataListener.java | 76 +++++++++++++++ .../ZookeeperDynamicConfiguration.java | 32 ++++--- .../ZookeeperDynamicConfigurationFactory.java | 9 -- 6 files changed, 125 insertions(+), 94 deletions(-) create mode 100644 dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDataListener.java diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/TreePathDynamicConfiguration.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/TreePathDynamicConfiguration.java index d311a908f5..cee63ea7db 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/TreePathDynamicConfiguration.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/TreePathDynamicConfiguration.java @@ -94,7 +94,7 @@ public abstract class TreePathDynamicConfiguration extends AbstractDynamicConfig @Override public final void addListener(String key, String group, ConfigurationListener listener) { String pathKey = buildPathKey(group, key); - doAddListener(pathKey, listener); + doAddListener(pathKey, listener, key, group); } @Override @@ -111,7 +111,7 @@ public abstract class TreePathDynamicConfiguration extends AbstractDynamicConfig protected abstract Collection doGetConfigKeys(String groupPath); - protected abstract void doAddListener(String pathKey, ConfigurationListener listener); + protected abstract void doAddListener(String pathKey, ConfigurationListener listener, String key, String group); protected abstract void doRemoveListener(String pathKey, ConfigurationListener listener); diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfiguration.java b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfiguration.java index 815e714025..cd43c54a31 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfiguration.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfiguration.java @@ -402,7 +402,7 @@ public class FileSystemDynamicConfiguration extends TreePathDynamicConfiguration } @Override - protected void doAddListener(String pathKey, ConfigurationListener listener) { + protected void doAddListener(String pathKey, ConfigurationListener listener, String key, String group) { doInListener(pathKey, (configFilePath, listeners) -> { if (listeners.isEmpty()) { // If no element, it indicates watchService was registered before ThrowableConsumer.execute(configFilePath, configFile -> { diff --git a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java index fe34bd5577..7c2bdf21fb 100644 --- a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java +++ b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java @@ -16,94 +16,50 @@ */ package org.apache.dubbo.configcenter.support.zookeeper; -import org.apache.dubbo.common.config.configcenter.ConfigChangeType; -import org.apache.dubbo.common.config.configcenter.ConfigChangedEvent; import org.apache.dubbo.common.config.configcenter.ConfigurationListener; import org.apache.dubbo.common.utils.CollectionUtils; -import org.apache.dubbo.common.utils.StringUtils; -import org.apache.dubbo.remoting.zookeeper.DataListener; -import org.apache.dubbo.remoting.zookeeper.EventType; import java.util.Map; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArraySet; -import static org.apache.dubbo.common.constants.CommonConstants.DOT_SEPARATOR; -import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR; +/** + * one path has one zookeeperDataListener + */ +public class CacheListener { -public class CacheListener implements DataListener { + private Map pathKeyListeners = new ConcurrentHashMap<>(); - private Map> keyListeners = new ConcurrentHashMap<>(); - private String rootPath; - - public CacheListener(String rootPath) { - this.rootPath = rootPath; + public CacheListener() { } - public void addListener(String key, ConfigurationListener configurationListener) { - Set listeners = keyListeners.computeIfAbsent(key, k -> new CopyOnWriteArraySet<>()); - listeners.add(configurationListener); + public ZookeeperDataListener addListener(String pathKey, ConfigurationListener configurationListener, String key, String group) { + ZookeeperDataListener zookeeperDataListener = pathKeyListeners.computeIfAbsent(pathKey, + _pathKey -> new ZookeeperDataListener(_pathKey, key, group)); + zookeeperDataListener.addListener(configurationListener); + return zookeeperDataListener; } - public void removeListener(String key, ConfigurationListener configurationListener) { - Set listeners = keyListeners.get(key); - if (listeners != null) { - listeners.remove(configurationListener); - } - } - - public Set getConfigurationListeners(String key) { - return keyListeners.get(key); - } - - /** - * This is used to convert a configuration nodePath into a key - * TODO doc - * - * @param path - * @return key (nodePath less the config root path) - */ - private String pathToKey(String path) { - if (StringUtils.isEmpty(path)) { - return path; - } - String groupKey = path.replace(rootPath + PATH_SEPARATOR, "").replaceAll(PATH_SEPARATOR, DOT_SEPARATOR); - return groupKey.substring(groupKey.indexOf(DOT_SEPARATOR) + 1); - } - - private String getGroup(String path) { - if (!StringUtils.isEmpty(path)) { - int beginIndex = path.indexOf(rootPath + PATH_SEPARATOR); - if (beginIndex > -1) { - String remain = path.substring((rootPath + PATH_SEPARATOR).length()); - int endIndex = remain.lastIndexOf(PATH_SEPARATOR); - if (endIndex > -1) { - return remain.substring(0, endIndex); - } + public ZookeeperDataListener removeListener(String pathKey, ConfigurationListener configurationListener) { + ZookeeperDataListener zookeeperDataListener = pathKeyListeners.get(pathKey); + if (zookeeperDataListener != null) { + zookeeperDataListener.removeListener(configurationListener); + if (CollectionUtils.isEmpty(zookeeperDataListener.getListeners())) { + pathKeyListeners.remove(pathKey); } } - return path; + return zookeeperDataListener; } + public ZookeeperDataListener getCachedListener(String pathKey) { + return pathKeyListeners.get(pathKey); + } - @Override - public void dataChanged(String path, Object value, EventType eventType) { - ConfigChangeType changeType; - if (EventType.NodeCreated.equals(eventType)) { - changeType = ConfigChangeType.ADDED; - } else if (value == null) { - changeType = ConfigChangeType.DELETED; - } else { - changeType = ConfigChangeType.MODIFIED; - } - String key = pathToKey(path); + public Map getPathKeyListeners() { + return pathKeyListeners; + } - ConfigChangedEvent configChangeEvent = new ConfigChangedEvent(key, getGroup(path), (String) value, changeType); - Set listeners = keyListeners.get(path); - if (CollectionUtils.isNotEmpty(listeners)) { - listeners.forEach(listener -> listener.process(configChangeEvent)); - } + public void clear() { + pathKeyListeners.clear(); } } diff --git a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDataListener.java b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDataListener.java new file mode 100644 index 0000000000..c489ca0e33 --- /dev/null +++ b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDataListener.java @@ -0,0 +1,76 @@ +/* + * 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.configcenter.support.zookeeper; + +import org.apache.dubbo.common.config.configcenter.ConfigChangeType; +import org.apache.dubbo.common.config.configcenter.ConfigChangedEvent; +import org.apache.dubbo.common.config.configcenter.ConfigurationListener; +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.remoting.zookeeper.DataListener; +import org.apache.dubbo.remoting.zookeeper.EventType; + +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; + +/** + * one path has multi configurationListeners + */ +public class ZookeeperDataListener implements DataListener { + private String path; + private String key; + private String group; + private Set listeners; + + public ZookeeperDataListener(String path, String key, String group) { + this.path = path; + this.key = key; + this.group = group; + this.listeners = new CopyOnWriteArraySet<>(); + } + + public void addListener(ConfigurationListener configurationListener) { + listeners.add(configurationListener); + } + + public void removeListener(ConfigurationListener configurationListener) { + listeners.remove(configurationListener); + } + + public Set getListeners() { + return listeners; + } + + @Override + public void dataChanged(String path, Object value, EventType eventType) { + if (!this.path.equals(path)) { + return; + } + ConfigChangeType changeType; + if (EventType.NodeCreated.equals(eventType)) { + changeType = ConfigChangeType.ADDED; + } else if (value == null) { + changeType = ConfigChangeType.DELETED; + } else { + changeType = ConfigChangeType.MODIFIED; + } + ConfigChangedEvent configChangeEvent = new ConfigChangedEvent(key, group, (String) value, changeType); + if (CollectionUtils.isNotEmpty(listeners)) { + listeners.forEach(listener -> listener.process(configChangeEvent)); + } + } + +} diff --git a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java index b335184e18..b427709dc8 100644 --- a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java +++ b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java @@ -29,15 +29,12 @@ import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter; import org.apache.zookeeper.data.Stat; import java.util.Collection; -import java.util.Set; +import java.util.Map; import java.util.concurrent.Executor; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -/** - * - */ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration { private Executor executor; @@ -51,7 +48,7 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration ZookeeperDynamicConfiguration(URL url, ZookeeperTransporter zookeeperTransporter) { super(url); - this.cacheListener = new CacheListener(rootPath); + this.cacheListener = new CacheListener(); final String threadName = this.getClass().getSimpleName(); this.executor = new ThreadPoolExecutor(DEFAULT_ZK_EXECUTOR_THREADS_NUM, DEFAULT_ZK_EXECUTOR_THREADS_NUM, @@ -78,6 +75,13 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration @Override protected void doClose() throws Exception { + // remove data listener + Map pathKeyListeners = cacheListener.getPathKeyListeners(); + for (Map.Entry entry : pathKeyListeners.entrySet()) { + zkClient.removeDataListener(entry.getKey(), entry.getValue()); + } + cacheListener.clear(); + // zkClient is shared in framework, should not close it here // zkClient.close(); // See: org.apache.dubbo.remoting.zookeeper.AbstractZookeeperTransporter#destroy() @@ -129,17 +133,21 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration } @Override - protected void doAddListener(String pathKey, ConfigurationListener listener) { - cacheListener.addListener(pathKey, listener); - zkClient.addDataListener(pathKey, cacheListener, executor); + protected void doAddListener(String pathKey, ConfigurationListener listener, String key, String group) { + ZookeeperDataListener cachedListener = cacheListener.getCachedListener(pathKey); + if (cachedListener != null) { + cachedListener.addListener(listener); + } else { + ZookeeperDataListener addedListener = cacheListener.addListener(pathKey, listener, key, group); + zkClient.addDataListener(pathKey, addedListener, executor); + } } @Override protected void doRemoveListener(String pathKey, ConfigurationListener listener) { - cacheListener.removeListener(pathKey, listener); - Set configurationListeners = cacheListener.getConfigurationListeners(pathKey); - if (CollectionUtils.isEmpty(configurationListeners)) { - zkClient.removeDataListener(pathKey, cacheListener); + ZookeeperDataListener zookeeperDataListener = cacheListener.removeListener(pathKey, listener); + if (CollectionUtils.isEmpty(zookeeperDataListener.getListeners())) { + zkClient.removeDataListener(pathKey, zookeeperDataListener); } } } diff --git a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationFactory.java b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationFactory.java index 309cf43faf..962c7bd028 100644 --- a/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationFactory.java +++ b/dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationFactory.java @@ -19,13 +19,9 @@ package org.apache.dubbo.configcenter.support.zookeeper; import org.apache.dubbo.common.URL; import org.apache.dubbo.common.config.configcenter.AbstractDynamicConfigurationFactory; import org.apache.dubbo.common.config.configcenter.DynamicConfiguration; -import org.apache.dubbo.common.extension.DisableInject; import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter; import org.apache.dubbo.rpc.model.ApplicationModel; -/** - * - */ public class ZookeeperDynamicConfigurationFactory extends AbstractDynamicConfigurationFactory { private ZookeeperTransporter zookeeperTransporter; @@ -37,11 +33,6 @@ public class ZookeeperDynamicConfigurationFactory extends AbstractDynamicConfigu this.zookeeperTransporter = ZookeeperTransporter.getExtension(applicationModel); } - @DisableInject - public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) { - this.zookeeperTransporter = zookeeperTransporter; - } - @Override protected DynamicConfiguration createDynamicConfiguration(URL url) { return new ZookeeperDynamicConfiguration(url, zookeeperTransporter);