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

Fix registry repeat export same service (#12578) 

* Fix registry repeat export same service

* fix uts

* Fix export

* Fix export
This commit is contained in:
Albumen Kevin 2023-06-21 11:29:54 +08:00 committed by GitHub
parent 25c2b3808a
commit 3b7934f512
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 118 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/multiple/exportprovider/MultipleRegistryCenterExportProviderIntegrationTest.java
View File

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

2
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/integration/single/exportprovider/SingleRegistryCenterExportProviderIntegrationTest.java
View File

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

4
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/RegistryScopeModelInitializer.java
View File

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

42
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ExporterFactory.java Normal file
View File

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

62
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/ReferenceCountExporter.java Normal file
View File

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

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

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