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

Merge branch 'apache-3.0' into apache-3.1

This commit is contained in:
Albumen Kevin 2022-08-14 10:49:45 +08:00
parent 213944c03e b3f533d726
commit 56721482ca
10 changed files with 154 additions and 144 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 -> {

2
dubbo-common/src/main/java/org/apache/dubbo/common/status/support/MemoryStatusChecker.java
View File

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

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,
@ -87,6 +84,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()
@ -138,17 +142,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);

69
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java
View File

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

2
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/LazyConnectExchangeClient.java
View File

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

6
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClient.java
View File

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