From d66349f982e6668873f0edcec5d0bc8c6c28b97f Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Thu, 5 Jan 2023 20:45:59 +0800 Subject: [PATCH 1/5] Fix app level configuration not work (#11215) --- .../apache/dubbo/registry/integration/RegistryProtocol.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java index d7ce4f1ba4..ff1e66d176 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java @@ -230,8 +230,8 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { // subscription information to cover. final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl); final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker); - Map overrideListeners = getProviderConfigurationListener(providerUrl).getOverrideListeners(); - overrideListeners.put(registryUrl, overrideSubscribeListener); + Map overrideListeners = getProviderConfigurationListener(overrideSubscribeUrl).getOverrideListeners(); + overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener); providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener); //export invoker @@ -898,7 +898,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { try { if (subscribeUrl != null) { Map overrideListeners = getProviderConfigurationListener(subscribeUrl).getOverrideListeners(); - NotifyListener listener = overrideListeners.remove(registerUrl); + NotifyListener listener = overrideListeners.remove(subscribeUrl); if (listener != null) { if (!registry.isServiceDiscovery()) { registry.unsubscribe(subscribeUrl, listener); From f16febe70dbd0b99cf764b7a7ef4a9bfbb9be6d7 Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 6 Jan 2023 15:10:02 +0800 Subject: [PATCH 2/5] Update NOTICE --- NOTICE | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/NOTICE b/NOTICE index b1fa2a521f..a060218b41 100644 --- a/NOTICE +++ b/NOTICE @@ -1,5 +1,5 @@ Apache Dubbo -Copyright 2018-2022 The Apache Software Foundation +Copyright 2018-2023 The Apache Software Foundation This product includes software developed at The Apache Software Foundation (http://www.apache.org/). From 4a1aa856ab5ac2a4e80a5646c5b2d78a59d66116 Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 6 Jan 2023 16:36:59 +0800 Subject: [PATCH 3/5] Store old revisions (#11241) --- .../common/constants/CommonConstants.java | 11 ++- .../client/AbstractServiceDiscovery.java | 68 +++++++++++++++ .../registry/client/ServiceDiscovery.java | 10 ++- .../metadata/MetadataServiceDelegation.java | 28 +++--- .../client/ServiceDiscoveryCacheTest.java | 85 +++++++++++++++++++ .../client/support/MockServiceDiscovery.java | 5 -- .../multiple/MultipleServiceDiscovery.java | 21 +++-- 7 files changed, 196 insertions(+), 32 deletions(-) create mode 100644 dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryCacheTest.java diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java index 3bbd55cc86..621a341204 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java @@ -17,13 +17,13 @@ package org.apache.dubbo.common.constants; -import org.apache.dubbo.common.URL; - import java.net.NetworkInterface; import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.regex.Pattern; +import org.apache.dubbo.common.URL; + public interface CommonConstants { String DUBBO = "dubbo"; @@ -361,6 +361,13 @@ public interface CommonConstants { */ 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 diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java index e4960f532f..77f983a743 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java @@ -19,6 +19,10 @@ package org.apache.dubbo.registry.client; import java.util.List; import java.util.Optional; 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.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.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.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.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_LOAD_METADATA; 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 volatile ServiceInstance serviceInstance; protected volatile MetadataInfo metadataInfo; + protected final ConcurrentHashMap metadataInfos = new ConcurrentHashMap<>(); + protected final ScheduledFuture refreshCacheFuture; protected MetadataReport metadataReport; protected String metadataType; protected final MetaCacheManager metaCacheManager; @@ -88,6 +99,30 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery { this.metaCacheManager = new MetaCacheManager(localCacheEnabled, getCacheNameSuffix(), applicationModel.getFrameworkModel().getBeanFactory() .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 oldestRevision = new AtomicReference<>(); + AtomicReference 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; } + @Override + public MetadataInfo getLocalMetadata(String revision) { + MetadataInfoStat metadataInfoStat = metadataInfos.get(revision); + if (metadataInfoStat != null) { + return metadataInfoStat.getMetadataInfo(); + } else { + return null; + } + } + @Override public MetadataInfo getRemoteMetadata(String revision, List instances) { MetadataInfo metadata = metaCacheManager.get(revision); @@ -217,6 +262,7 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery { public final void destroy() throws Exception { isDestroy = true; metaCacheManager.destroy(); + refreshCacheFuture.cancel(true); doDestroy(); } @@ -291,12 +337,17 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery { } protected void reportMetadata(MetadataInfo metadataInfo) { + if (metadataInfo == null) { + return; + } if (metadataReport != null) { SubscriberMetadataIdentifier identifier = new SubscriberMetadataIdentifier(serviceName, metadataInfo.getRevision()); if ((DEFAULT_METADATA_STORAGE_TYPE.equals(metadataType) && metadataReport.shouldReportMetadata()) || REMOTE_METADATA_STORAGE_TYPE.equals(metadataType)) { metadataReport.publishAppMetadata(identifier, metadataInfo); } } + MetadataInfo clonedMetadataInfo = metadataInfo.clone(); + metadataInfos.put(metadataInfo.getRevision(), new MetadataInfoStat(clonedMetadataInfo)); } protected void unReportMetadata(MetadataInfo metadataInfo) { @@ -328,4 +379,21 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery { } 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; + } + } } diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java index 3a4a3625f5..29960b39e9 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java @@ -16,15 +16,15 @@ */ 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.lang.Prioritized; import org.apache.dubbo.metadata.MetadataInfo; import org.apache.dubbo.registry.RegistryService; 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; /** @@ -69,6 +69,10 @@ public interface ServiceDiscovery extends RegistryService, Prioritized { MetadataInfo getLocalMetadata(); + default MetadataInfo getLocalMetadata(String revision) { + return getLocalMetadata(); + } + MetadataInfo getRemoteMetadata(String revision); MetadataInfo getRemoteMetadata(String revision, List instances); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataServiceDelegation.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataServiceDelegation.java index 3701a93abd..b9440f02c5 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataServiceDelegation.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataServiceDelegation.java @@ -16,18 +16,6 @@ */ 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.Comparator; import java.util.HashMap; @@ -40,6 +28,18 @@ import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; 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.unmodifiableSortedSet; import static org.apache.dubbo.common.URL.buildKey; @@ -182,8 +182,8 @@ public class MetadataServiceDelegation implements MetadataService, Disposable { } for (ServiceDiscovery sd : registryManager.getServiceDiscoveries()) { - MetadataInfo metadataInfo = sd.getLocalMetadata(); - if (revision.equals(metadataInfo.getRevision())) { + MetadataInfo metadataInfo = sd.getLocalMetadata(revision); + if (metadataInfo != null && revision.equals(metadataInfo.getRevision())) { return metadataInfo; } } diff --git a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryCacheTest.java b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryCacheTest.java new file mode 100644 index 0000000000..9b759b8ef4 --- /dev/null +++ b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryCacheTest.java @@ -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 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(); + } +} diff --git a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/support/MockServiceDiscovery.java b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/support/MockServiceDiscovery.java index fc20255582..46e7e04264 100644 --- a/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/support/MockServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/support/MockServiceDiscovery.java @@ -40,11 +40,6 @@ public class MockServiceDiscovery extends AbstractServiceDiscovery { } - @Override - public void doUpdate(ServiceInstance oldServiceInstance, ServiceInstance newServiceInstance) throws RuntimeException { - - } - @Override public void doUnregister(ServiceInstance serviceInstance) { diff --git a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java index dfa52ea95e..6d02ad195b 100644 --- a/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java @@ -16,6 +16,14 @@ */ 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.constants.CommonConstants; 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.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 static final String REGISTRY_PREFIX_KEY = "child."; 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."); } + @Override + public MetadataInfo getLocalMetadata(String revision) { + throw new UnsupportedOperationException("Multiple registry implementation does not support getLocalMetadata() method."); + } + @Override public MetadataInfo getRemoteMetadata(String revision) { throw new UnsupportedOperationException("Multiple registry implementation does not support getMetadata() method."); From 36c9edc8ed4aa0a3a1fd0f970dcf7656bb802f23 Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 6 Jan 2023 16:46:05 +0800 Subject: [PATCH 4/5] Fix Multi Registry Override (#11245) --- .../integration/RegistryProtocol.java | 57 ++++++++++++------- 1 file changed, 37 insertions(+), 20 deletions(-) diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java index ff1e66d176..b038d9b34b 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java @@ -16,6 +16,16 @@ */ 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.config.ConfigurationUtils; 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.url.component.ServiceConfigURL; 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.StringUtils; 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.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.CLUSTER_KEY; import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN; @@ -193,8 +195,8 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { return 9090; } - public Map getOverrideListeners() { - Map map = new HashMap<>(); + public Map> getOverrideListeners() { + Map> map = new HashMap<>(); List applicationModels = frameworkModel.getApplicationModels(); if (applicationModels.size() == 1) { return applicationModels.get(0).getBeanFactory().getBean(ProviderConfigurationListener.class).getOverrideListeners(); @@ -230,8 +232,9 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { // subscription information to cover. final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl); final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker); - Map overrideListeners = getProviderConfigurationListener(overrideSubscribeUrl).getOverrideListeners(); - overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener); + Map> overrideListeners = getProviderConfigurationListener(overrideSubscribeUrl).getOverrideListeners(); + overrideListeners.computeIfAbsent(overrideSubscribeUrl, k -> new ConcurrentHashSet<>()) + .add(overrideSubscribeListener); providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener); //export invoker @@ -253,6 +256,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { exporter.setRegisterUrl(registeredProviderUrl); exporter.setSubscribeUrl(overrideSubscribeUrl); + exporter.setNotifyListener(overrideSubscribeListener); if (!registry.isServiceDiscovery()) { // Deprecated! Subscribe to override rules in 2.6.x or before. @@ -818,7 +822,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { private class ProviderConfigurationListener extends AbstractConfiguratorListener { - private final Map overrideListeners = new ConcurrentHashMap<>(); + private final Map> overrideListeners = new ConcurrentHashMap<>(); public ProviderConfigurationListener(ModuleModel moduleModel) { super(moduleModel); @@ -840,10 +844,14 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { @Override protected void notifyOverrides() { - overrideListeners.values().forEach(listener -> ((OverrideListener) listener).doOverrideIfNecessary()); + overrideListeners.values().forEach(listeners -> { + for (NotifyListener listener : listeners) { + ((OverrideListener) listener).doOverrideIfNecessary(); + } + }); } - public Map getOverrideListeners() { + public Map> getOverrideListeners() { return overrideListeners; } } @@ -863,6 +871,8 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { private URL subscribeUrl; private URL registerUrl; + private NotifyListener notifyListener; + public ExporterChangeableWrapper(Exporter exporter, Invoker originInvoker) { this.exporter = exporter; this.originInvoker = originInvoker; @@ -897,11 +907,11 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { } try { if (subscribeUrl != null) { - Map overrideListeners = getProviderConfigurationListener(subscribeUrl).getOverrideListeners(); - NotifyListener listener = overrideListeners.remove(subscribeUrl); - if (listener != null) { + Map> overrideListeners = getProviderConfigurationListener(subscribeUrl).getOverrideListeners(); + Set listeners = overrideListeners.get(subscribeUrl); + if (listeners.remove(notifyListener)) { if (!registry.isServiceDiscovery()) { - registry.unsubscribe(subscribeUrl, listener); + registry.unsubscribe(subscribeUrl, notifyListener); } ApplicationModel applicationModel = getApplicationModel(registerUrl.getScopeModel()); if (applicationModel.getModelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) { @@ -914,6 +924,9 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { } } } + if (listeners.isEmpty()) { + overrideListeners.remove(subscribeUrl); + } } } catch (Throwable t) { logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", t.getMessage(), t); @@ -941,6 +954,10 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { this.registerUrl = registerUrl; } + public void setNotifyListener(NotifyListener notifyListener) { + this.notifyListener = notifyListener; + } + public URL getRegisterUrl() { return registerUrl; } From c2eef7b5ce7e0c2e6c09ceadd361ca75c43da0e4 Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 6 Jan 2023 16:47:30 +0800 Subject: [PATCH 5/5] Fix zk resubscribe failed (#11247) --- .../ServiceInstancesChangedListener.java | 38 +++++++++---------- .../zookeeper/ZookeeperServiceDiscovery.java | 3 ++ 2 files changed, 22 insertions(+), 19 deletions(-) diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java index 6196331538..79c9819337 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java @@ -16,6 +16,22 @@ */ 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.URL; 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.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.LoggerCodeConstants.REGISTRY_FAILED_REFRESH_ADDRESS; 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.EMPTY_PROTOCOL; import static org.apache.dubbo.common.constants.RegistryConstants.ENABLE_EMPTY_PROTECTION_KEY; import static org.apache.dubbo.metadata.RevisionResolver.EMPTY_REVISION; @@ -466,12 +466,12 @@ public class ServiceInstancesChangedListener { return false; } ServiceInstancesChangedListener that = (ServiceInstancesChangedListener) o; - return Objects.equals(getServiceNames(), that.getServiceNames()); + return Objects.equals(getServiceNames(), that.getServiceNames()) && Objects.equals(listeners, that.listeners); } @Override public int hashCode() { - return Objects.hash(getClass(), getServiceNames()); + return Objects.hash(getClass(), getServiceNames(), listeners); } protected class AddressRefreshRetryTask implements Runnable { diff --git a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java index 0ee7cb5cdc..a4897bfc88 100644 --- a/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java @@ -156,6 +156,9 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery { @Override public void removeServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws IllegalArgumentException { + if (!instanceListeners.remove(listener)) { + return; + } listener.getServiceNames().forEach(serviceName -> { ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.get(serviceName); if (watcher != null) {