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

Enhance service discovery update interval (#11223) 

* Enhance service discovery update interval

* Update DefaultApplicationDeployer.java
This commit is contained in:
Albumen Kevin 2023-02-15 07:17:13 +08:00 committed by GitHub
parent e1baaae71b
commit 551ff4a374
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 83 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
dubbo-common/src/main/java/org/apache/dubbo/common/deploy/ApplicationDeployer.java
View File

@ -88,4 +88,15 @@ public interface ApplicationDeployer extends Deployer<ApplicationModel> {
* module state changed callbacks
*/
void notifyModuleChanged(ModuleModel moduleModel, DeployState state);
/**
* Increase the count of service update threads.
* NOTE: should call ${@link ApplicationDeployer#decreaseServiceRefreshCount()} after update finished
*/
void increaseServiceRefreshCount();
/**
* Decrease the count of service update threads
*/
void decreaseServiceRefreshCount();
}

30
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
View File

@ -75,6 +75,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@ -816,6 +817,13 @@ public class DefaultApplicationDeployer extends AbstractDeployer<ApplicationMode
private volatile boolean registered;
private final AtomicInteger instanceRefreshScheduleTimes = new AtomicInteger(0);
/**
* Indicate that how many threads are updating service
*/
private final AtomicInteger serviceRefreshState = new AtomicInteger(0);
private void registerServiceInstance() {
try {
registered = true;
@ -831,6 +839,18 @@ public class DefaultApplicationDeployer extends AbstractDeployer<ApplicationMode
if (applicationModel.isDestroyed()) {
return;
}
// refresh for 30 times (default for 30s) when deployer is not started, prevent submit too many revision
if (instanceRefreshScheduleTimes.incrementAndGet() % 30 != 0 && !isStarted()) {
return;
}
// refresh for 5 times (default for 5s) when services are being updated by other threads, prevent submit too many revision
// note: should not always wait here
if (serviceRefreshState.get() != 0 && instanceRefreshScheduleTimes.get() % 5 != 0) {
return;
}
try {
if (!applicationModel.isDestroyed() && registered) {
ServiceInstanceMetadataUtils.refreshMetadataAndInstance(applicationModel);
@ -844,6 +864,16 @@ public class DefaultApplicationDeployer extends AbstractDeployer<ApplicationMode
}
}
@Override
public void increaseServiceRefreshCount() {
serviceRefreshState.incrementAndGet();
}
@Override
public void decreaseServiceRefreshCount() {
serviceRefreshState.decrementAndGet();
}
private void unregisterServiceInstance() {
if (registered) {
ServiceInstanceMetadataUtils.unregisterMetadataAndInstance(applicationModel);

2
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataConstants.java
View File

@ -24,7 +24,7 @@ public class MetadataConstants {
public static final String SERVICE_META_DATA_STORE_TAG = ".smd";
public static final String CONSUMER_META_DATA_STORE_TAG = ".cmd";
public static final String METADATA_PUBLISH_DELAY_KEY = "dubbo.application.metadata.publish.delay";
public static final int DEFAULT_METADATA_PUBLISH_DELAY = 30000;
public static final int DEFAULT_METADATA_PUBLISH_DELAY = 1000;
public static final String METADATA_PROXY_TIMEOUT_KEY = "dubbo.application.metadata.proxy.delay";
public static final int DEFAULT_METADATA_TIMEOUT_VALUE = 5000;
public static String REPORT_CONSUMER_URL_KEY = "report-consumer-definition";

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

@ -29,6 +29,7 @@ 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;
import org.apache.dubbo.common.deploy.ApplicationDeployer;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
@ -122,8 +123,8 @@ import static org.apache.dubbo.remoting.Constants.CHECK_KEY;
import static org.apache.dubbo.remoting.Constants.CODEC_KEY;
import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY;
import static org.apache.dubbo.remoting.Constants.EXCHANGER_KEY;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static org.apache.dubbo.remoting.Constants.PREFER_SERIALIZATION_KEY;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static org.apache.dubbo.rpc.Constants.DEPRECATED_KEY;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
import static org.apache.dubbo.rpc.Constants.INTERFACES;
@ -210,7 +211,13 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
}
private void register(Registry registry, URL registeredProviderUrl) {
registry.register(registeredProviderUrl);
ApplicationDeployer deployer = registeredProviderUrl.getOrDefaultApplicationModel().getDeployer();
try {
deployer.increaseServiceRefreshCount();
registry.register(registeredProviderUrl);
} finally {
deployer.decreaseServiceRefreshCount();
}
}
private void registerStatedUrl(URL registryUrl, URL registeredProviderUrl, boolean registered) {
@ -736,7 +743,14 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
this.configurators = Configurator.toConfigurators(classifyUrls(matchedUrls, UrlUtils::isConfigurator))
.orElse(configurators);
doOverrideIfNecessary();
ApplicationDeployer deployer = subscribeUrl.getOrDefaultApplicationModel().getDeployer();
try {
deployer.increaseServiceRefreshCount();
doOverrideIfNecessary();
} finally {
deployer.decreaseServiceRefreshCount();
}
}
public synchronized void doOverrideIfNecessary() {
@ -802,10 +816,13 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
private URL providerUrl;
private OverrideListener notifyListener;
private final ModuleModel moduleModel;
public ServiceConfigurationListener(ModuleModel moduleModel, URL providerUrl, OverrideListener notifyListener) {
super(moduleModel);
this.providerUrl = providerUrl;
this.notifyListener = notifyListener;
this.moduleModel = moduleModel;
if (moduleModel.getModelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) {
this.initWith(DynamicConfiguration.getRuleKey(providerUrl) + CONFIGURATORS_SUFFIX);
}
@ -817,7 +834,13 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
@Override
protected void notifyOverrides() {
notifyListener.doOverrideIfNecessary();
ApplicationDeployer deployer = this.moduleModel.getApplicationModel().getDeployer();
try {
deployer.increaseServiceRefreshCount();
notifyListener.doOverrideIfNecessary();
} finally {
deployer.decreaseServiceRefreshCount();
}
}
}
@ -825,8 +848,11 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
private final Map<URL, Set<NotifyListener>> overrideListeners = new ConcurrentHashMap<>();
private final ModuleModel moduleModel;
public ProviderConfigurationListener(ModuleModel moduleModel) {
super(moduleModel);
this.moduleModel = moduleModel;
if (moduleModel.getModelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_CONFIGURATION_LISTEN, true)) {
this.initWith(moduleModel.getApplicationModel().getApplicationName() + CONFIGURATORS_SUFFIX);
}
@ -845,11 +871,17 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
@Override
protected void notifyOverrides() {
overrideListeners.values().forEach(listeners -> {
for (NotifyListener listener : listeners) {
((OverrideListener) listener).doOverrideIfNecessary();
}
});
ApplicationDeployer deployer = this.moduleModel.getApplicationModel().getDeployer();
try {
deployer.increaseServiceRefreshCount();
overrideListeners.values().forEach(listeners -> {
for (NotifyListener listener : listeners) {
((OverrideListener) listener).doOverrideIfNecessary();
}
});
} finally {
deployer.decreaseServiceRefreshCount();
}
}
public Map<URL, Set<NotifyListener>> getOverrideListeners() {