From 551ff4a374cdcf90fb80c2ef517f62485c3ed9d3 Mon Sep 17 00:00:00 2001
From: Albumen Kevin <jhq0812@gmail.com>
Date: Wed, 15 Feb 2023 07:17:13 +0800
Subject: [PATCH] Enhance service discovery update interval (#11223)

* Enhance service discovery update interval

* Update DefaultApplicationDeployer.java
---
 .../common/deploy/ApplicationDeployer.java    | 11 ++++
 .../deploy/DefaultApplicationDeployer.java    | 30 +++++++++++
 .../dubbo/metadata/MetadataConstants.java     |  2 +-
 .../integration/RegistryProtocol.java         | 50 +++++++++++++++----
 4 files changed, 83 insertions(+), 10 deletions(-)

diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/deploy/ApplicationDeployer.java b/dubbo-common/src/main/java/org/apache/dubbo/common/deploy/ApplicationDeployer.java
index 611f47c715..7cdd356872 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/deploy/ApplicationDeployer.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/deploy/ApplicationDeployer.java
@@ -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();
 }
diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
index 5d82ac7da2..f4f64b1167 100644
--- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
+++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
@@ -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);
diff --git a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataConstants.java b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataConstants.java
index 39ff229708..c6bda320c6 100644
--- a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataConstants.java
+++ b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataConstants.java
@@ -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";
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 cac908b1fe..c19cae36d3 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
@@ -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() {