From 3b7934f5122833685acc6cd9bcff208f92ee9a6b Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Wed, 21 Jun 2023 11:29:54 +0800 Subject: [PATCH] Fix registry repeat export same service (#12578) * Fix registry repeat export same service * fix uts * Fix export * Fix export --- ...ryCenterExportProviderIntegrationTest.java | 2 +- ...ryCenterExportProviderIntegrationTest.java | 2 +- .../RegistryScopeModelInitializer.java | 4 +- .../registry/integration/ExporterFactory.java | 42 +++++++++++++ .../integration/ReferenceCountExporter.java | 62 +++++++++++++++++++ .../integration/RegistryProtocol.java | 13 ++-- 6 files changed, 118 insertions(+), 7 deletions(-) create mode 100644 dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ExporterFactory.java create mode 100644 dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ReferenceCountExporter.java diff --git a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/multiple/exportprovider/MultipleRegistryCenterExportProviderIntegrationTest.java b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/multiple/exportprovider/MultipleRegistryCenterExportProviderIntegrationTest.java index 445ddb6dfc..d2f9b7ee2d 100644 --- a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/multiple/exportprovider/MultipleRegistryCenterExportProviderIntegrationTest.java +++ b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/multiple/exportprovider/MultipleRegistryCenterExportProviderIntegrationTest.java @@ -188,7 +188,7 @@ class MultipleRegistryCenterExportProviderIntegrationTest implements Integration // 1. InjvmExporter // 2. DubboExporter with service-discovery-registry protocol // 3. DubboExporter with registry protocol - Assertions.assertEquals(exporterListener.getExportedExporters().size(), 7); + Assertions.assertEquals(exporterListener.getExportedExporters().size(), 4); // The exported exporter contains MultipleRegistryCenterExportProviderFilter Assertions.assertTrue(exporterListener.getFilters().contains(filter)); diff --git a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/single/exportprovider/SingleRegistryCenterExportProviderIntegrationTest.java b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/single/exportprovider/SingleRegistryCenterExportProviderIntegrationTest.java index 3ab4bf7b53..41de19efcb 100644 --- a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/single/exportprovider/SingleRegistryCenterExportProviderIntegrationTest.java +++ b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/single/exportprovider/SingleRegistryCenterExportProviderIntegrationTest.java @@ -195,7 +195,7 @@ class SingleRegistryCenterExportProviderIntegrationTest implements IntegrationTe // 1. InjvmExporter // 2. DubboExporter with service-discovery-registry protocol // 3. DubboExporter with registry protocol - Assertions.assertEquals(exporterListener.getExportedExporters().size(), 5); + Assertions.assertEquals(exporterListener.getExportedExporters().size(), 4); // The exported exporter contains SingleRegistryCenterExportProviderFilter Assertions.assertTrue(exporterListener.getFilters().contains(filter)); // The consumer can be notified and get provider's metadata through metadata mapping info. diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/RegistryScopeModelInitializer.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/RegistryScopeModelInitializer.java index 6be4e29947..40c788985c 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/RegistryScopeModelInitializer.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/RegistryScopeModelInitializer.java @@ -17,6 +17,7 @@ package org.apache.dubbo.registry; import org.apache.dubbo.common.beans.factory.ScopeBeanFactory; +import org.apache.dubbo.registry.integration.ExporterFactory; import org.apache.dubbo.registry.support.RegistryManager; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.FrameworkModel; @@ -26,7 +27,8 @@ import org.apache.dubbo.rpc.model.ScopeModelInitializer; public class RegistryScopeModelInitializer implements ScopeModelInitializer { @Override public void initializeFrameworkModel(FrameworkModel frameworkModel) { - + ScopeBeanFactory beanFactory = frameworkModel.getBeanFactory(); + beanFactory.registerBean(ExporterFactory.class); } @Override diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ExporterFactory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ExporterFactory.java new file mode 100644 index 0000000000..5eebc1bcaa --- /dev/null +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ExporterFactory.java @@ -0,0 +1,42 @@ +/* + * 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.integration; + +import org.apache.dubbo.rpc.Exporter; + +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; + +public class ExporterFactory { + private final Map> exporters = new ConcurrentHashMap<>(); + + protected ReferenceCountExporter createExporter(String providerKey, Callable> exporterProducer) { + return exporters.computeIfAbsent(providerKey, + key -> { + try { + return new ReferenceCountExporter<>(exporterProducer.call(), key, this); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + protected void remove(String key, ReferenceCountExporter exporter) { + exporters.remove(key, exporter); + } +} diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ReferenceCountExporter.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ReferenceCountExporter.java new file mode 100644 index 0000000000..bcdce7172f --- /dev/null +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ReferenceCountExporter.java @@ -0,0 +1,62 @@ +/* + * 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.integration; + +import org.apache.dubbo.rpc.Exporter; +import org.apache.dubbo.rpc.Invoker; + +import java.util.concurrent.atomic.AtomicInteger; + +public class ReferenceCountExporter implements Exporter { + private final Exporter exporter; + private final String providerKey; + private final ExporterFactory exporterFactory; + private final AtomicInteger count = new AtomicInteger(0); + + public ReferenceCountExporter(Exporter exporter, String providerKey, ExporterFactory exporterFactory) { + this.exporter = exporter; + this.providerKey = providerKey; + this.exporterFactory = exporterFactory; + } + + @Override + public Invoker getInvoker() { + return exporter.getInvoker(); + } + + public void increaseCount() { + count.incrementAndGet(); + } + + @Override + public void unexport() { + if (count.decrementAndGet() == 0) { + exporter.unexport(); + } + exporterFactory.remove(providerKey, this); + } + + @Override + public void register() { + + } + + @Override + public void unregister() { + + } +} 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 a9e3b908c7..b69376e784 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 @@ -171,6 +171,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { private ConcurrentMap reExportFailedTasks = new ConcurrentHashMap<>(); private HashedWheelTimer retryTimer = new HashedWheelTimer(new NamedThreadFactory("DubboReexportTimer", true), DEFAULT_REGISTRY_RETRY_PERIOD, TimeUnit.MILLISECONDS, 128); private FrameworkModel frameworkModel; + private ExporterFactory exporterFactory; //Filter the parameters that do not need to be output in url(Starting with .) private static String[] getFilteredKeys(URL url) { @@ -190,6 +191,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { @Override public void setFrameworkModel(FrameworkModel frameworkModel) { this.frameworkModel = frameworkModel; + this.exporterFactory = frameworkModel.getBeanFactory().getBean(ExporterFactory.class); } public void setProtocol(Protocol protocol) { @@ -312,11 +314,13 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { private ExporterChangeableWrapper doLocalExport(final Invoker originInvoker, URL providerUrl) { String providerUrlKey = getProviderUrlKey(originInvoker); String registryUrlKey = getRegistryUrlKey(originInvoker); + Invoker invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl); + ReferenceCountExporter exporter = exporterFactory.createExporter(providerUrlKey, () -> protocol.export(invokerDelegate)); return (ExporterChangeableWrapper) bounds.computeIfAbsent(providerUrlKey, _k -> new ConcurrentHashMap<>()) - .computeIfAbsent(registryUrlKey, s ->{ - Invoker invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl); - return new ExporterChangeableWrapper<>((Exporter) protocol.export(invokerDelegate), originInvoker); + .computeIfAbsent(registryUrlKey, s -> { + return new ExporterChangeableWrapper<>( + (ReferenceCountExporter) exporter, originInvoker); }); } @@ -953,8 +957,9 @@ public class RegistryProtocol implements Protocol, ScopeModelAware { private NotifyListener notifyListener; private final AtomicBoolean registered = new AtomicBoolean(false); - public ExporterChangeableWrapper(Exporter exporter, Invoker originInvoker) { + public ExporterChangeableWrapper(ReferenceCountExporter exporter, Invoker originInvoker) { this.exporter = exporter; + exporter.increaseCount(); this.originInvoker = originInvoker; FrameworkExecutorRepository frameworkExecutorRepository = originInvoker.getUrl().getOrDefaultFrameworkModel().getBeanFactory() .getBean(FrameworkExecutorRepository.class);