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