This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new 3b7934f512 Fix registry repeat export same service (#12578)
3b7934f512 is described below
commit 3b7934f5122833685acc6cd9bcff208f92ee9a6b
Author: Albumen Kevin <[email protected]>
AuthorDate: Wed Jun 21 11:29:54 2023 +0800
Fix registry repeat export same service (#12578)
* Fix registry repeat export same service
* fix uts
* Fix export
* Fix export
---
...egistryCenterExportProviderIntegrationTest.java | 2 +-
...egistryCenterExportProviderIntegrationTest.java | 2 +-
.../registry/RegistryScopeModelInitializer.java | 4 +-
.../registry/integration/ExporterFactory.java | 42 +++++++++++++++
.../integration/ReferenceCountExporter.java | 62 ++++++++++++++++++++++
.../registry/integration/RegistryProtocol.java | 13 +++--
6 files changed, 118 insertions(+), 7 deletions(-)
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<String, ReferenceCountExporter<?>> exporters = new
ConcurrentHashMap<>();
+
+ protected ReferenceCountExporter<?> createExporter(String providerKey,
Callable<Exporter<?>> 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<T> implements Exporter<T> {
+ private final Exporter<T> exporter;
+ private final String providerKey;
+ private final ExporterFactory exporterFactory;
+ private final AtomicInteger count = new AtomicInteger(0);
+
+ public ReferenceCountExporter(Exporter<T> exporter, String providerKey,
ExporterFactory exporterFactory) {
+ this.exporter = exporter;
+ this.providerKey = providerKey;
+ this.exporterFactory = exporterFactory;
+ }
+
+ @Override
+ public Invoker<T> 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<URL, ReExportTask> 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 <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T>
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<T>)
bounds.computeIfAbsent(providerUrlKey, _k -> new ConcurrentHashMap<>())
- .computeIfAbsent(registryUrlKey, s ->{
- Invoker<?> invokerDelegate = new
InvokerDelegate<>(originInvoker, providerUrl);
- return new ExporterChangeableWrapper<>((Exporter<T>)
protocol.export(invokerDelegate), originInvoker);
+ .computeIfAbsent(registryUrlKey, s -> {
+ return new ExporterChangeableWrapper<>(
+ (ReferenceCountExporter<T>) 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<T> exporter, Invoker<T>
originInvoker) {
+ public ExporterChangeableWrapper(ReferenceCountExporter<T> exporter,
Invoker<T> originInvoker) {
this.exporter = exporter;
+ exporter.increaseCount();
this.originInvoker = originInvoker;
FrameworkExecutorRepository frameworkExecutorRepository =
originInvoker.getUrl().getOrDefaultFrameworkModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class);
