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

Merge branch 'apache-3.1' into apache-3.2 

# Conflicts:
#	dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
This commit is contained in:
Albumen Kevin 2023-01-06 16:48:39 +08:00
parent 4188b884b2 c2eef7b5ce
commit 0419f25ac2
11 changed files with 256 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
NOTICE
View File

@ -1,5 +1,5 @@
Apache Dubbo Apache Dubbo
Copyright 2018-2022 The Apache Software Foundation Copyright 2018-2023 The Apache Software Foundation
This product includes software developed at This product includes software developed at
The Apache Software Foundation (http://www.apache.org/). The Apache Software Foundation (http://www.apache.org/).

11
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -17,13 +17,13 @@
package org.apache.dubbo.common.constants; package org.apache.dubbo.common.constants;
import org.apache.dubbo.common.URL;
import java.net.NetworkInterface; import java.net.NetworkInterface;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import org.apache.dubbo.common.URL;
public interface CommonConstants { public interface CommonConstants {
String DUBBO = "dubbo"; String DUBBO = "dubbo";
@ -367,6 +367,13 @@ public interface CommonConstants {
*/ */
String REGISTRY_LOCAL_FILE_CACHE_ENABLED = "file.cache"; String REGISTRY_LOCAL_FILE_CACHE_ENABLED = "file.cache";
String METADATA_INFO_CACHE_EXPIRE_KEY = "metadata-info-cache.expire";
int DEFAULT_METADATA_INFO_CACHE_EXPIRE = 10 * 60 * 1000;
String METADATA_INFO_CACHE_SIZE_KEY = "metadata-info-cache.size";
int DEFAULT_METADATA_INFO_CACHE_SIZE = 16;
/** /**
* The limit of callback service instances for one interface on every client * The limit of callback service instances for one interface on every client

68
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java
View File

@ -19,6 +19,10 @@ package org.apache.dubbo.registry.client;
import java.util.List; import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.dubbo.common.URL; import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
@ -37,9 +41,14 @@ import org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils;
import org.apache.dubbo.registry.client.metadata.store.MetaCacheManager; import org.apache.dubbo.registry.client.metadata.store.MetaCacheManager;
import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.ApplicationModel;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_METADATA_INFO_CACHE_EXPIRE;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_METADATA_INFO_CACHE_SIZE;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_METADATA_STORAGE_TYPE; import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_METADATA_STORAGE_TYPE;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_INFO_CACHE_EXPIRE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_INFO_CACHE_SIZE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_LOCAL_FILE_CACHE_ENABLED; import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_LOCAL_FILE_CACHE_ENABLED;
import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_METADATA_STORAGE_TYPE; import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_METADATA_STORAGE_TYPE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_FETCH_INSTANCE; import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_FETCH_INSTANCE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_LOAD_METADATA; import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_LOAD_METADATA;
import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_CLUSTER_KEY; import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_CLUSTER_KEY;
@ -59,6 +68,8 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
protected final String serviceName; protected final String serviceName;
protected volatile ServiceInstance serviceInstance; protected volatile ServiceInstance serviceInstance;
protected volatile MetadataInfo metadataInfo; protected volatile MetadataInfo metadataInfo;
protected final ConcurrentHashMap<String, MetadataInfoStat> metadataInfos = new ConcurrentHashMap<>();
protected final ScheduledFuture<?> refreshCacheFuture;
protected MetadataReport metadataReport; protected MetadataReport metadataReport;
protected String metadataType; protected String metadataType;
protected final MetaCacheManager metaCacheManager; protected final MetaCacheManager metaCacheManager;
@ -88,6 +99,30 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
this.metaCacheManager = new MetaCacheManager(localCacheEnabled, getCacheNameSuffix(), this.metaCacheManager = new MetaCacheManager(localCacheEnabled, getCacheNameSuffix(),
applicationModel.getFrameworkModel().getBeanFactory() applicationModel.getFrameworkModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class).getCacheRefreshingScheduledExecutor()); .getBean(FrameworkExecutorRepository.class).getCacheRefreshingScheduledExecutor());
int metadataInfoCacheExpireTime = registryURL.getParameter(METADATA_INFO_CACHE_EXPIRE_KEY, DEFAULT_METADATA_INFO_CACHE_EXPIRE);
int metadataInfoCacheSize = registryURL.getParameter(METADATA_INFO_CACHE_SIZE_KEY, DEFAULT_METADATA_INFO_CACHE_SIZE);
this.refreshCacheFuture = applicationModel.getFrameworkModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor()
.scheduleAtFixedRate(() -> {
try {
while (metadataInfos.size() > metadataInfoCacheSize) {
AtomicReference<String> oldestRevision = new AtomicReference<>();
AtomicReference<MetadataInfoStat> oldestStat = new AtomicReference<>();
metadataInfos.forEach((k, v) -> {
if (System.currentTimeMillis() - v.getUpdateTime() > metadataInfoCacheExpireTime &&
(oldestStat.get() == null || oldestStat.get().getUpdateTime() > v.getUpdateTime())) {
oldestRevision.set(k);
oldestStat.set(v);
}
});
if (oldestStat.get() != null) {
metadataInfos.remove(oldestRevision.get(), oldestStat.get());
}
}
} catch (Throwable t) {
logger.error(INTERNAL_ERROR, "", "", "Error occurred when clean up metadata info cache.", t);
}
}, metadataInfoCacheExpireTime / 2, metadataInfoCacheExpireTime / 2, TimeUnit.MILLISECONDS);
} }
@ -163,6 +198,16 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
return this.metadataInfo; return this.metadataInfo;
} }
@Override
public MetadataInfo getLocalMetadata(String revision) {
MetadataInfoStat metadataInfoStat = metadataInfos.get(revision);
if (metadataInfoStat != null) {
return metadataInfoStat.getMetadataInfo();
} else {
return null;
}
}
@Override @Override
public MetadataInfo getRemoteMetadata(String revision, List<ServiceInstance> instances) { public MetadataInfo getRemoteMetadata(String revision, List<ServiceInstance> instances) {
MetadataInfo metadata = metaCacheManager.get(revision); MetadataInfo metadata = metaCacheManager.get(revision);
@ -217,6 +262,7 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
public final void destroy() throws Exception { public final void destroy() throws Exception {
isDestroy = true; isDestroy = true;
metaCacheManager.destroy(); metaCacheManager.destroy();
refreshCacheFuture.cancel(true);
doDestroy(); doDestroy();
} }
@ -291,12 +337,17 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
} }
protected void reportMetadata(MetadataInfo metadataInfo) { protected void reportMetadata(MetadataInfo metadataInfo) {
if (metadataInfo == null) {
return;
}
if (metadataReport != null) { if (metadataReport != null) {
SubscriberMetadataIdentifier identifier = new SubscriberMetadataIdentifier(serviceName, metadataInfo.getRevision()); SubscriberMetadataIdentifier identifier = new SubscriberMetadataIdentifier(serviceName, metadataInfo.getRevision());
if ((DEFAULT_METADATA_STORAGE_TYPE.equals(metadataType) && metadataReport.shouldReportMetadata()) || REMOTE_METADATA_STORAGE_TYPE.equals(metadataType)) { if ((DEFAULT_METADATA_STORAGE_TYPE.equals(metadataType) && metadataReport.shouldReportMetadata()) || REMOTE_METADATA_STORAGE_TYPE.equals(metadataType)) {
metadataReport.publishAppMetadata(identifier, metadataInfo); metadataReport.publishAppMetadata(identifier, metadataInfo);
} }
} }
MetadataInfo clonedMetadataInfo = metadataInfo.clone();
metadataInfos.put(metadataInfo.getRevision(), new MetadataInfoStat(clonedMetadataInfo));
} }
protected void unReportMetadata(MetadataInfo metadataInfo) { protected void unReportMetadata(MetadataInfo metadataInfo) {
@ -328,4 +379,21 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
} }
return stringBuilder.toString(); return stringBuilder.toString();
} }
private static class MetadataInfoStat {
private final MetadataInfo metadataInfo;
private final long updateTime = System.currentTimeMillis();
public MetadataInfoStat(MetadataInfo metadataInfo) {
this.metadataInfo = metadataInfo;
}
public MetadataInfo getMetadataInfo() {
return metadataInfo;
}
public long getUpdateTime() {
return updateTime;
}
}
} }

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java
View File

@ -16,15 +16,15 @@
*/ */
package org.apache.dubbo.registry.client; package org.apache.dubbo.registry.client;
import java.util.List;
import java.util.Set;
import org.apache.dubbo.common.URL; import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.lang.Prioritized; import org.apache.dubbo.common.lang.Prioritized;
import org.apache.dubbo.metadata.MetadataInfo; import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.RegistryService; import org.apache.dubbo.registry.RegistryService;
import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener; import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
import java.util.List;
import java.util.Set;
import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_DELAY_NOTIFICATION_KEY; import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_DELAY_NOTIFICATION_KEY;
/** /**
@ -69,6 +69,10 @@ public interface ServiceDiscovery extends RegistryService, Prioritized {
MetadataInfo getLocalMetadata(); MetadataInfo getLocalMetadata();
default MetadataInfo getLocalMetadata(String revision) {
return getLocalMetadata();
}
MetadataInfo getRemoteMetadata(String revision); MetadataInfo getRemoteMetadata(String revision);
MetadataInfo getRemoteMetadata(String revision, List<ServiceInstance> instances); MetadataInfo getRemoteMetadata(String revision, List<ServiceInstance> instances);

38
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
View File

@ -16,6 +16,22 @@
*/ */
package org.apache.dubbo.registry.client.event.listener; package org.apache.dubbo.registry.client.event.listener;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.dubbo.common.ProtocolServiceKey; import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL; import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.URLBuilder; import org.apache.dubbo.common.URLBuilder;
@ -38,25 +54,9 @@ import org.apache.dubbo.registry.client.metadata.ServiceInstanceNotificationCust
import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ScopeModelUtil; import org.apache.dubbo.rpc.model.ScopeModelUtil;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY; import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_REFRESH_ADDRESS;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR; import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_REFRESH_ADDRESS;
import static org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_ENABLE_EMPTY_PROTECTION; import static org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_ENABLE_EMPTY_PROTECTION;
import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL; import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL;
import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY;
@ -458,12 +458,12 @@ public class ServiceInstancesChangedListener {
return false; return false;
} }
ServiceInstancesChangedListener that = (ServiceInstancesChangedListener) o; ServiceInstancesChangedListener that = (ServiceInstancesChangedListener) o;
return Objects.equals(getServiceNames(), that.getServiceNames()); return Objects.equals(getServiceNames(), that.getServiceNames()) && Objects.equals(listeners, that.listeners);
} }
@Override @Override
public int hashCode() { public int hashCode() {
return Objects.hash(getClass(), getServiceNames()); return Objects.hash(getClass(), getServiceNames(), listeners);
} }
protected class AddressRefreshRetryTask implements Runnable { protected class AddressRefreshRetryTask implements Runnable {

28
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataServiceDelegation.java
View File

@ -16,18 +16,6 @@
*/ */
package org.apache.dubbo.registry.client.metadata; package org.apache.dubbo.registry.client.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.Disposable;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.InstanceMetadataChangedListener;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.support.RegistryManager;
import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Comparator; import java.util.Comparator;
import java.util.HashMap; import java.util.HashMap;
@ -40,6 +28,18 @@ import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.Disposable;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.InstanceMetadataChangedListener;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.support.RegistryManager;
import org.apache.dubbo.rpc.model.ApplicationModel;
import static java.util.Collections.emptySortedSet; import static java.util.Collections.emptySortedSet;
import static java.util.Collections.unmodifiableSortedSet; import static java.util.Collections.unmodifiableSortedSet;
import static org.apache.dubbo.common.URL.buildKey; import static org.apache.dubbo.common.URL.buildKey;
@ -182,8 +182,8 @@ public class MetadataServiceDelegation implements MetadataService, Disposable {
} }
for (ServiceDiscovery sd : registryManager.getServiceDiscoveries()) { for (ServiceDiscovery sd : registryManager.getServiceDiscoveries()) {
MetadataInfo metadataInfo = sd.getLocalMetadata(); MetadataInfo metadataInfo = sd.getLocalMetadata(revision);
if (revision.equals(metadataInfo.getRevision())) { if (metadataInfo != null && revision.equals(metadataInfo.getRevision())) {
return metadataInfo; return metadataInfo;
} }
} }

57
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java
View File

@ -16,6 +16,16 @@
*/ */
package org.apache.dubbo.registry.integration; package org.apache.dubbo.registry.integration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.dubbo.common.URL; import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.ConfigurationUtils; import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration; import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
@ -25,6 +35,7 @@ import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.timer.HashedWheelTimer; import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.common.url.component.ServiceConfigURL; import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.utils.CollectionUtils; import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.NamedThreadFactory; import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.StringUtils; import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils; import org.apache.dubbo.common.utils.UrlUtils;
@ -59,15 +70,6 @@ import org.apache.dubbo.rpc.model.ScopeModelUtil;
import org.apache.dubbo.rpc.protocol.InvokerWrapper; import org.apache.dubbo.rpc.protocol.InvokerWrapper;
import org.apache.dubbo.rpc.support.ProtocolUtils; import org.apache.dubbo.rpc.support.ProtocolUtils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY; import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY; import static org.apache.dubbo.common.constants.CommonConstants.CLUSTER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN; import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
@ -194,8 +196,8 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
return 9090; return 9090;
} }
public Map<URL, NotifyListener> getOverrideListeners() { public Map<URL, Set<NotifyListener>> getOverrideListeners() {
Map<URL, NotifyListener> map = new HashMap<>(); Map<URL, Set<NotifyListener>> map = new HashMap<>();
List<ApplicationModel> applicationModels = frameworkModel.getApplicationModels(); List<ApplicationModel> applicationModels = frameworkModel.getApplicationModels();
if (applicationModels.size() == 1) { if (applicationModels.size() == 1) {
return applicationModels.get(0).getBeanFactory().getBean(ProviderConfigurationListener.class).getOverrideListeners(); return applicationModels.get(0).getBeanFactory().getBean(ProviderConfigurationListener.class).getOverrideListeners();
@ -231,8 +233,9 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
// subscription information to cover. // subscription information to cover.
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl); final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker); final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
Map<URL, NotifyListener> overrideListeners = getProviderConfigurationListener(providerUrl).getOverrideListeners(); Map<URL, Set<NotifyListener>> overrideListeners = getProviderConfigurationListener(overrideSubscribeUrl).getOverrideListeners();
overrideListeners.put(registryUrl, overrideSubscribeListener); overrideListeners.computeIfAbsent(overrideSubscribeUrl, k -> new ConcurrentHashSet<>())
.add(overrideSubscribeListener);
providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener); providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
//export invoker //export invoker
@ -254,6 +257,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
exporter.setRegisterUrl(registeredProviderUrl); exporter.setRegisterUrl(registeredProviderUrl);
exporter.setSubscribeUrl(overrideSubscribeUrl); exporter.setSubscribeUrl(overrideSubscribeUrl);
exporter.setNotifyListener(overrideSubscribeListener);
if (!registry.isServiceDiscovery()) { if (!registry.isServiceDiscovery()) {
// Deprecated! Subscribe to override rules in 2.6.x or before. // Deprecated! Subscribe to override rules in 2.6.x or before.
@ -819,7 +823,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
private class ProviderConfigurationListener extends AbstractConfiguratorListener { private class ProviderConfigurationListener extends AbstractConfiguratorListener {
private final Map<URL, NotifyListener> overrideListeners = new ConcurrentHashMap<>(); private final Map<URL, Set<NotifyListener>> overrideListeners = new ConcurrentHashMap<>();
public ProviderConfigurationListener(ModuleModel moduleModel) { public ProviderConfigurationListener(ModuleModel moduleModel) {
super(moduleModel); super(moduleModel);
@ -841,10 +845,14 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
@Override @Override
protected void notifyOverrides() { protected void notifyOverrides() {
overrideListeners.values().forEach(listener -> ((OverrideListener) listener).doOverrideIfNecessary()); overrideListeners.values().forEach(listeners -> {
for (NotifyListener listener : listeners) {
((OverrideListener) listener).doOverrideIfNecessary();
}
});
} }
public Map<URL, NotifyListener> getOverrideListeners() { public Map<URL, Set<NotifyListener>> getOverrideListeners() {
return overrideListeners; return overrideListeners;
} }
} }
@ -864,6 +872,8 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
private URL subscribeUrl; private URL subscribeUrl;
private URL registerUrl; private URL registerUrl;
private NotifyListener notifyListener;
public ExporterChangeableWrapper(Exporter<T> exporter, Invoker<T> originInvoker) { public ExporterChangeableWrapper(Exporter<T> exporter, Invoker<T> originInvoker) {
this.exporter = exporter; this.exporter = exporter;
this.originInvoker = originInvoker; this.originInvoker = originInvoker;
@ -898,11 +908,11 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
} }
try { try {
if (subscribeUrl != null) { if (subscribeUrl != null) {
Map<URL, NotifyListener> overrideListeners = getProviderConfigurationListener(subscribeUrl).getOverrideListeners(); Map<URL, Set<NotifyListener>> overrideListeners = getProviderConfigurationListener(subscribeUrl).getOverrideListeners();
NotifyListener listener = overrideListeners.remove(registerUrl); Set<NotifyListener> listeners = overrideListeners.get(subscribeUrl);
if (listener != null) { if (listeners.remove(notifyListener)) {
if (!registry.isServiceDiscovery()) { if (!registry.isServiceDiscovery()) {
registry.unsubscribe(subscribeUrl, listener); registry.unsubscribe(subscribeUrl, notifyListener);
} }
ApplicationModel applicationModel = getApplicationModel(registerUrl.getScopeModel()); ApplicationModel applicationModel = getApplicationModel(registerUrl.getScopeModel());
if (applicationModel.getModelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) { if (applicationModel.getModelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) {
@ -915,6 +925,9 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
} }
} }
} }
if (listeners.isEmpty()) {
overrideListeners.remove(subscribeUrl);
}
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", t.getMessage(), t); logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", t.getMessage(), t);
@ -942,6 +955,10 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
this.registerUrl = registerUrl; this.registerUrl = registerUrl;
} }
public void setNotifyListener(NotifyListener notifyListener) {
this.notifyListener = notifyListener;
}
public URL getRegisterUrl() { public URL getRegisterUrl() {
return registerUrl; return registerUrl;
} }

85
dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryCacheTest.java Normal file
View File

@ -0,0 +1,85 @@
/*
* 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.registry.client;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.client.support.MockServiceDiscovery;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_INFO_CACHE_EXPIRE_KEY;
import static org.awaitility.Awaitility.await;
class ServiceDiscoveryCacheTest {
@Test
void test() throws InterruptedException {
ApplicationModel applicationModel = FrameworkModel.defaultModel().newApplication();
applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("Test"));
URL registryUrl = URL.valueOf("mock://127.0.0.1:12345").addParameter(METADATA_INFO_CACHE_EXPIRE_KEY, 10);
MockServiceDiscovery mockServiceDiscovery = Mockito.spy(new MockServiceDiscovery(applicationModel, registryUrl));
mockServiceDiscovery.register(URL.valueOf("mock://127.0.0.1:12345")
.setServiceInterface("org.apache.dubbo.registry.service.DemoService"));
mockServiceDiscovery.register();
ServiceInstance localInstance = mockServiceDiscovery.getLocalInstance();
Assertions.assertEquals(localInstance.getServiceMetadata(), mockServiceDiscovery.getLocalMetadata(localInstance.getServiceMetadata().getRevision()));
List<MetadataInfo> instances = new LinkedList<>();
instances.add(localInstance.getServiceMetadata().clone());
for (int i = 0; i < 15; i++) {
Thread.sleep(1);
mockServiceDiscovery.register(URL.valueOf("mock://127.0.0.1:12345")
.setServiceInterface("org.apache.dubbo.registry.service.DemoService" + i));
mockServiceDiscovery.update();
instances.add(mockServiceDiscovery.getLocalInstance().getServiceMetadata().clone());
}
for (MetadataInfo instance : instances) {
Assertions.assertEquals(instance, mockServiceDiscovery.getLocalMetadata(instance.getRevision()));
}
for (int i = 0; i < 5; i++) {
Thread.sleep(1);
mockServiceDiscovery.register(URL.valueOf("mock://127.0.0.1:12345")
.setServiceInterface("org.apache.dubbo.registry.service.DemoService-new" + i));
mockServiceDiscovery.update();
instances.add(mockServiceDiscovery.getLocalInstance().getServiceMetadata().clone());
}
await().until(() -> Objects.isNull(mockServiceDiscovery.getLocalMetadata(instances.get(4).getRevision())));
for (int i = 0; i < 5; i++) {
Assertions.assertNull(mockServiceDiscovery.getLocalMetadata(instances.get(i).getRevision()));
}
applicationModel.destroy();
}
}

5
dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/support/MockServiceDiscovery.java
View File

@ -40,11 +40,6 @@ public class MockServiceDiscovery extends AbstractServiceDiscovery {
} }
@Override
public void doUpdate(ServiceInstance oldServiceInstance, ServiceInstance newServiceInstance) throws RuntimeException {
}
@Override @Override
public void doUnregister(ServiceInstance serviceInstance) { public void doUnregister(ServiceInstance serviceInstance) {

21
dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java
View File

@ -16,6 +16,14 @@
*/ */
package org.apache.dubbo.registry.multiple; package org.apache.dubbo.registry.multiple;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import org.apache.dubbo.common.URL; import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.metadata.MetadataInfo; import org.apache.dubbo.metadata.MetadataInfo;
@ -26,14 +34,6 @@ import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent; import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener; import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
public class MultipleServiceDiscovery implements ServiceDiscovery { public class MultipleServiceDiscovery implements ServiceDiscovery {
public static final String REGISTRY_PREFIX_KEY = "child."; public static final String REGISTRY_PREFIX_KEY = "child.";
private static final String REGISTRY_TYPE = "registry-type"; private static final String REGISTRY_TYPE = "registry-type";
@ -137,6 +137,11 @@ public class MultipleServiceDiscovery implements ServiceDiscovery {
throw new UnsupportedOperationException("Multiple registry implementation does not support getMetadata() method."); throw new UnsupportedOperationException("Multiple registry implementation does not support getMetadata() method.");
} }
@Override
public MetadataInfo getLocalMetadata(String revision) {
throw new UnsupportedOperationException("Multiple registry implementation does not support getLocalMetadata() method.");
}
@Override @Override
public MetadataInfo getRemoteMetadata(String revision) { public MetadataInfo getRemoteMetadata(String revision) {
throw new UnsupportedOperationException("Multiple registry implementation does not support getMetadata() method."); throw new UnsupportedOperationException("Multiple registry implementation does not support getMetadata() method.");

3
dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java
View File

@ -156,6 +156,9 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
@Override @Override
public void removeServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws IllegalArgumentException { public void removeServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws IllegalArgumentException {
if (!instanceListeners.remove(listener)) {
return;
}
listener.getServiceNames().forEach(serviceName -> { listener.getServiceNames().forEach(serviceName -> {
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.get(serviceName); ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.get(serviceName);
if (watcher != null) { if (watcher != null) {