[servicecomb-java-chassis] 12/19: [SCB-1691] Each ServiceRegistry uses an isolated EventBus

[ya...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yaohaishi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git

commit c79d5e122b2d526fe4227ecbfb5061f5d174859e
Author: yhs0092 <yh...@163.com>
AuthorDate: Tue Feb 11 21:16:33 2020 +0800

    [SCB-1691] Each ServiceRegistry uses an isolated EventBus
---
 .../servicecomb/serviceregistry/RegistryUtils.java | 38 +++++++++++++++++++---
 .../registry/ServiceRegistryFactory.java           |  8 +++++
 2 files changed, 42 insertions(+), 4 deletions(-)

diff --git a/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/RegistryUtils.java b/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/RegistryUtils.java
index 49f28e2..3f99eae 100644
--- a/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/RegistryUtils.java
+++ b/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/RegistryUtils.java
@@ -25,6 +25,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.regex.Matcher;
@@ -52,10 +53,13 @@ import org.apache.servicecomb.serviceregistry.registry.cache.MicroserviceCache;
 import org.apache.servicecomb.serviceregistry.registry.cache.MicroserviceCache.MicroserviceCacheStatus;
 import org.apache.servicecomb.serviceregistry.registry.cache.MicroserviceCacheKey;
 import org.apache.servicecomb.serviceregistry.swagger.SwaggerLoader;
+import org.apache.servicecomb.serviceregistry.task.MicroserviceInstanceRegisterTask;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.util.StringUtils;
 
 import com.google.common.base.Charsets;
+import com.google.common.eventbus.Subscribe;
 import com.google.common.hash.Hashing;
 import com.netflix.config.DynamicPropertyFactory;
 
@@ -115,14 +119,13 @@ public final class RegistryUtils {
 
   private static void initializeServiceRegistries(MicroserviceDefinition microserviceDefinition) {
     serviceRegistry =
-        ServiceRegistryFactory
-            .create(EventManager.eventBus, ServiceRegistryConfig.INSTANCE, microserviceDefinition);
+        ServiceRegistryFactory.create(ServiceRegistryConfig.INSTANCE, microserviceDefinition);
     EXTRA_SERVICE_REGISTRY_CONFIGS.forEach((k, v) -> {
-      ServiceRegistry serviceRegistry = ServiceRegistryFactory
-          .create(EventManager.getEventBus(), v, microserviceDefinition);
+      ServiceRegistry serviceRegistry = ServiceRegistryFactory.create(v, microserviceDefinition);
       addExtraServiceRegistry(serviceRegistry);
     });
     executeOnEachServiceRegistry(ServiceRegistry::init);
+    executeOnEachServiceRegistry(AfterServiceInstanceRegistryHandler::new);
   }
 
   public static void run() {
@@ -411,4 +414,31 @@ public final class RegistryUtils {
           "Illegal registry name, the format should be " + ServiceRegistry.REGISTRY_NAME_FORMAT);
     }
   }
+
+  public static class AfterServiceInstanceRegistryHandler {
+    private static AtomicInteger instanceRegisterCounter = new AtomicInteger(EXTRA_SERVICE_REGISTRIES.size() + 1);
+
+    private ServiceRegistry serviceRegistry;
+
+    AfterServiceInstanceRegistryHandler(ServiceRegistry serviceRegistry) {
+      this.serviceRegistry = serviceRegistry;
+      serviceRegistry.getEventBus().register(this);
+    }
+
+    @Subscribe
+    public void afterRegistryInstance(MicroserviceInstanceRegisterTask microserviceInstanceRegisterTask) {
+      LOGGER.info("receive MicroserviceInstanceRegisterTask event of [{}]", serviceRegistry.getName());
+      if (StringUtils.isEmpty(serviceRegistry.getMicroserviceInstance().getInstanceId())) {
+        return;
+      }
+
+      LOGGER.info("ServiceRegistry[{}] has completed instance registry", serviceRegistry.getName());
+      EventManager.unregister(this);
+
+      if (instanceRegisterCounter.decrementAndGet() > 0) {
+        return;
+      }
+      EventManager.getEventBus().post(microserviceInstanceRegisterTask);
+    }
+  }
 }
diff --git a/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/registry/ServiceRegistryFactory.java b/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/registry/ServiceRegistryFactory.java
index 0dab855..a672bbf 100644
--- a/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/registry/ServiceRegistryFactory.java
+++ b/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/registry/ServiceRegistryFactory.java
@@ -52,8 +52,16 @@ public final class ServiceRegistryFactory {
     return new LocalServiceRegistry(eventBus, serviceRegistryConfig, microserviceDefinition).localFile(localFile);
   }
 
+  public static ServiceRegistry create(ServiceRegistryConfig serviceRegistryConfig,
+      MicroserviceDefinition microserviceDefinition) {
+    return create(null, serviceRegistryConfig, microserviceDefinition);
+  }
+
   public static ServiceRegistry create(EventBus eventBus, ServiceRegistryConfig serviceRegistryConfig,
       MicroserviceDefinition microserviceDefinition) {
+    if (null == eventBus) {
+      eventBus = new SimpleEventBus();
+    }
     String localModeFile = System.getProperty(LocalServiceRegistryClientImpl.LOCAL_REGISTRY_FILE_KEY);
     if (!StringUtils.isEmpty(localModeFile)) {
       LOGGER.info(