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

Optimize the logic of zkDataListener (#10445)

This commit is contained in:
灼华 2022-08-12 15:35:18 +08:00 committed by GitHub
parent 01deaf9a0d
commit b3f533d726
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 125 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/TreePathDynamicConfiguration.java
View File

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

2
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/file/FileSystemDynamicConfiguration.java
View File

@ -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 -> {

96
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java
View File

@ -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<String, ZookeeperDataListener> pathKeyListeners = new ConcurrentHashMap<>();
private Map<String, Set<ConfigurationListener>> keyListeners = new ConcurrentHashMap<>();
private String rootPath;
public CacheListener(String rootPath) {
this.rootPath = rootPath;
public CacheListener() {
}
public void addListener(String key, ConfigurationListener configurationListener) {
Set<ConfigurationListener> 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<ConfigurationListener> listeners = keyListeners.get(key);
if (listeners != null) {
listeners.remove(configurationListener);
}
}
public Set<ConfigurationListener> 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<String, ZookeeperDataListener> getPathKeyListeners() {
return pathKeyListeners;
}
ConfigChangedEvent configChangeEvent = new ConfigChangedEvent(key, getGroup(path), (String) value, changeType);
Set<ConfigurationListener> listeners = keyListeners.get(path);
if (CollectionUtils.isNotEmpty(listeners)) {
listeners.forEach(listener -> listener.process(configChangeEvent));
}
public void clear() {
pathKeyListeners.clear();
}
}

76
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDataListener.java Normal file
View File

@ -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<ConfigurationListener> 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<ConfigurationListener> 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));
}
}
}

32
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java
View File

@ -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<String, ZookeeperDataListener> pathKeyListeners = cacheListener.getPathKeyListeners();
for (Map.Entry<String, ZookeeperDataListener> 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<ConfigurationListener> 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);
}
}
}

9
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationFactory.java
View File

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