Dubbo cloud native (#4927)

* Polish #4542 : [Enhancement] Adapt the Java standard Event/Listener mechanism * Polish #4541 : [Feature] Add local File System DynamicConfigurationFactory‘s extension * Polish #4541 : Bugfix * Polish #4541 : Optimization * Polish #4541 : Add the compatibility for PollingWatchService on the some platforms * Polish #4541 : Add delay publish without ThreadPoolExecutor * Polish #4541 : Refactor the extension name * Polish #4541 : Add remove ops * Polish #4541 : Add testable constructor * Polish #4541 : Add getConfigGroups method * Polish #4610 : [Refactor] Refactor the bootstrap module * Polish #4541 : Fix the nulling URL issue * Polish #4622 : [Refactor] Refactor ConfigManager * Polish #4622 : [Refactor] Refactor ConfigManager * Polish #4622 : Support multiple configcenters * Polish #4671 : ServiceNameMapping will not map the group, version and protocol * update referenceCount log (#4683) Add comments to support multiple shared connections * Polish /#4687 : Remove the duplicated test code in dubbo-config-spring (#4688) * #4685 修改代码if判断false问题 if (hasException == false)修改成if (!hasException) (#4695) * Fixed Service annotation method parameters are not in effect (#4598) * keep demo simple, and switch to use zookeeper as registry center (#4705) * keep demo simple, and switch to use zookeeper as registry center * remove comment * @reference auto-wires the instance of generic interface #4594 (#4677) * try to shorten maven output to make travis build pass (#4710) * use CountDownLatch to check zk registry if establish connection (#4589) * Minor change * Rename the extension name of WritableMetadataService * Polish #4759 : [Refactor] Change the signature of methods of MetadataService #4759 * Merge remote-tracking branch 'upstream/master' into dubbo-cloud-native # Conflicts: # dubbo-all/pom.xml # dubbo-bom/pom.xml # dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/AbstractInterfaceConfig.java # dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ApplicationConfig.java # dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ConfigCenterConfig.java # dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java # dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/RegistryConfig.java # dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java # dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/ReferenceConfigTest.java # dubbo-configcenter/dubbo-configcenter-api/src/main/java/org/apache/dubbo/configcenter/DynamicConfiguration.java # dubbo-configcenter/dubbo-configcenter-api/src/test/java/org/apache/dubbo/configcenter/mock/MockDynamicConfiguration.java # dubbo-configcenter/dubbo-configcenter-consul/src/main/java/org/apache/dubbo/configcenter/consul/ConsulDynamicConfiguration.java # dubbo-configcenter/dubbo-configcenter-etcd/src/test/java/org/apache/dubbo/configcenter/support/etcd/EtcdDynamicConfigurationTest.java # dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java # dubbo-configcenter/dubbo-configcenter-nacos/src/test/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfigurationTest.java # dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java # dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/definition/model/MethodDefinition.java # dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/report/identifier/MetadataIdentifier.java # dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/report/support/AbstractMetadataReport.java # dubbo-metadata/dubbo-metadata-api/src/test/java/org/apache/dubbo/metadata/report/identifier/MetadataIdentifierTest.java # dubbo-metadata/dubbo-metadata-definition-protobuf/src/main/java/org/apache/dubbo/metadata/definition/protobuf/ProtobufTypeBuilder.java # dubbo-metadata/dubbo-metadata-definition-protobuf/src/test/java/org/apache/dubbo/metadata/definition/protobuf/ProtobufTypeBuilderTest.java # dubbo-metadata/pom.xml # dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/AbstractConfiguratorListener.java # dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java # dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistryFactory.java # dubbo-rpc/dubbo-rpc-xml/src/main/java/org/apache/dubbo/xml/rpc/protocol/xmlrpc/XmlRpcProtocol.java * Polish #3984 : Add the implementation of Page<ServiceInstance> getInstances(String serviceName, int offset, int pageSize, boolean healthyOnly) * Code merge * Fix the cases * Merge remote-tracking branch 'upstream/cloud-native' into dubbo-cloud-native # Conflicts: # dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProviderBootstrap.java # dubbo-metadata/dubbo-metadata-definition-protobuf/pom.xml # dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/support/ServiceOrientedRegistryTest.java # dubbo-registry/dubbo-registry-consul/src/main/java/org/apache/dubbo/registry/consul/ConsulServiceDiscoveryFactory.java # dubbo-registry/dubbo-registry-etcd3/src/main/java/org/apache/dubbo/registry/etcd/EtcdServiceDiscovery.java * Refactor ConfigManager * Refactor ConfigManager * Resolve the issues on ConfigManager * Refactor and add test-cases for ConfigManager * Polish #4774 : [Feature] Dubbo Cloud Native - To Support in Spring * Polish #4808 : [Feature] Add the registered/unregistered event mechanism ShutdownHook * Polish #4807 : [Feature] Add the callback mechanism ShutdownHook #4807 * Polish #4813 : [Feature] add Prioritized implementation for ServiceInstanceCustomizer * Polish #4815 : [Feature] Add the ServiceLoader for Dubbo's services or components * Polish #4815 : [Feature] Add the ServiceLoader for Dubbo's services or components * Polish #4813 : [Feature] add Prioritized implementation for ServiceInstanceCustomizer * Polish #4807 : Add sort implementation * Refactor * Refactor * Polish #4845 : [Feature] Enhance the Event-Publishing feature to original ServiceDiscovery * Merge remote-tracking branch 'upstream/cloud-native' into dubbo-cloud-native # Conflicts: # dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceDiscoveryFactory.java # dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java * Merge remote-tracking branch 'upstream/cloud-native' into dubbo-cloud-native # Conflicts: # dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceDiscoveryFactory.java # dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java * Polish #4854 : [Feature] MetadataService supports the Dubbo protocol under auto-increased port * Polish #4857 : [Enhancement] Sync the Metadata storage type into ApplicationConfig * Polish #4868 : [Enhancement] Refactor ConfigChangeEvent * Polish #4868 : [Enhancement] Refactor ConfigChangeEvent * Polish #4873 : [Feature] Add a conditional EventListener into Event Module * Polish #4875 : [Feature] Refactor ServiceInstancesChangedListener * Remove the cycle dependencies * Remove the cycle dependencies * Polish #4903 : [Feature] Set source into the BeanDefinition of Dubbo Config * Polish #4902 : [Feature] Dubbo Cloud Native to Spring XML scenario * Polish #4713 : Initial the new module and dependencies * Polish #4690 : AnnotatedBeanDefinitionRegistryUtils#registerBeans can't remove the duplicated bean definitions * Polish #4690 : AnnotatedBeanDefinitionRegistryUtils#registerBeans can't remove the duplicated bean definitions * Polish #4690 : AnnotatedBeanDefinitionRegistryUtils#registerBeans can't remove the duplicated bean definitions * Polish #4910 : [Feature] To suppoort DubboLifecycleComponentApplicationListener in Spring XML scenario * Polish #4713 : Add Service discovery implementation for Eureka #4713 * Polish #4713 : Add Service registration and discovery implementation for Eureka * Polish #4713 : Add Service registration and discovery implementation for Eureka * Polish #4920 : [Refactor] Extract the common implementation for URLs' revision * Refactor * Polish #4925 : ServiceDiscovery limits only one ServiceInstancesChangedListener each service
apache · Aug 23, 2019 · d023a8f · d023a8f
1 parent df0d39c
commit d023a8f
Show file tree

Hide file tree

Showing 8 changed files with 82 additions and 21 deletions.
diff --git a/...y-api/src/main/java/org/apache/dubbo/registry/client/EventPublishingServiceDiscovery.java b/...y-api/src/main/java/org/apache/dubbo/registry/client/EventPublishingServiceDiscovery.java
@@ -220,6 +220,7 @@ public String toString() {
     @Override
     public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws NullPointerException, IllegalArgumentException {
         serviceDiscovery.addServiceInstancesChangedListener(listener);
+        eventDispatcher.addEventListener(listener);
     }
 
     @Override

diff --git a/...y/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java b/...y/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java
@@ -20,15 +20,19 @@
 import org.apache.dubbo.common.extension.SPI;
 import org.apache.dubbo.common.lang.Prioritized;
 import org.apache.dubbo.common.utils.Page;
+import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.event.EventDispatcher;
 import org.apache.dubbo.event.EventListener;
+import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
 import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
 
+import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Stream;
 
 import static java.util.Collections.unmodifiableList;
 import static java.util.Collections.unmodifiableMap;
@@ -206,6 +210,49 @@ default void addServiceInstancesChangedListener(ServiceInstancesChangedListener
         getDefaultExtension().addEventListener(listener);
     }
 
+    /**
+     * Dispatch the {@link ServiceInstancesChangedEvent}
+     *
+     * @param serviceName the name of service whose service instances have been changed
+     */
+    default void dispatchServiceInstancesChangedEvent(String serviceName) {
+        dispatchServiceInstancesChangedEvent(serviceName, getInstances(serviceName));
+    }
+
+    /**
+     * Dispatch the {@link ServiceInstancesChangedEvent}
+     *
+     * @param serviceName       the name of service whose service instances have been changed
+     * @param otherServiceNames the names of other services
+     */
+    default void dispatchServiceInstancesChangedEvent(String serviceName, String... otherServiceNames) {
+        dispatchServiceInstancesChangedEvent(serviceName, getInstances(serviceName));
+        if (otherServiceNames != null) {
+            Stream.of(otherServiceNames)
+                    .filter(StringUtils::isNotEmpty)
+                    .forEach(this::dispatchServiceInstancesChangedEvent);
+        }
+    }
+
+    /**
+     * Dispatch the {@link ServiceInstancesChangedEvent}
+     *
+     * @param serviceName      the name of service whose service instances have been changed
+     * @param serviceInstances the service instances have been changed
+     */
+    default void dispatchServiceInstancesChangedEvent(String serviceName, Collection<ServiceInstance> serviceInstances) {
+        dispatchServiceInstancesChangedEvent(new ServiceInstancesChangedEvent(serviceName, serviceInstances));
+    }
+
+    /**
+     * Dispatch the {@link ServiceInstancesChangedEvent}
+     *
+     * @param event the {@link ServiceInstancesChangedEvent}
+     */
+    default void dispatchServiceInstancesChangedEvent(ServiceInstancesChangedEvent event) {
+        getDefaultExtension().dispatch(event);
+    }
+
     // ==================================================================================== //
 
     /**

diff --git a/...registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java b/...registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
@@ -39,6 +39,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -104,6 +105,8 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
 
     private final WritableMetadataService writableMetadataService;
 
+    private final Set<String> listenedServices = new LinkedHashSet<>();
+
     public ServiceDiscoveryRegistry(URL registryURL) {
         super(registryURL);
         this.serviceDiscovery = createServiceDiscovery(registryURL);
@@ -283,8 +286,8 @@ protected void subscribeURLs(URL url, NotifyListener listener, String serviceNam
 
         subscribeURLs(url, listener, serviceName, serviceInstances);
 
-        // Add Listener
-        serviceDiscovery.addServiceInstancesChangedListener(new ServiceInstancesChangedListener(serviceName) {
+        // register ServiceInstancesChangedListener
+        registerServiceInstancesChangedListener(new ServiceInstancesChangedListener(serviceName) {
 
             @Override
             public void onEvent(ServiceInstancesChangedEvent event) {
@@ -293,6 +296,17 @@ public void onEvent(ServiceInstancesChangedEvent event) {
         });
     }
 
+    /**
+     * Register the {@link ServiceInstancesChangedListener} If absent
+     *
+     * @param listener the {@link ServiceInstancesChangedListener}
+     */
+    private void registerServiceInstancesChangedListener(ServiceInstancesChangedListener listener) {
+        if (listenedServices.add(listener.getServiceName())) {
+            serviceDiscovery.addServiceInstancesChangedListener(listener);
+        }
+    }
+
     protected void subscribeURLs(URL subscribedURL, NotifyListener listener, String serviceName,
                                  Collection<ServiceInstance> serviceInstances) {
 

diff --git a/...java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java b/...java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
@@ -59,4 +59,17 @@ public final String getServiceName() {
     public final boolean accept(ServiceInstancesChangedEvent event) {
         return Objects.equals(getServiceName(), event.getServiceName());
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof ServiceInstancesChangedListener)) return false;
+        ServiceInstancesChangedListener that = (ServiceInstancesChangedListener) o;
+        return Objects.equals(getServiceName(), that.getServiceName());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getClass(), getServiceName());
+    }
 }
diff --git a/...bbo-registry-etcd3/src/main/java/org/apache/dubbo/registry/etcd/EtcdServiceDiscovery.java b/...bbo-registry-etcd3/src/main/java/org/apache/dubbo/registry/etcd/EtcdServiceDiscovery.java
@@ -161,7 +161,6 @@ public Set<String> getServices() {
     @Override
     public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws NullPointerException, IllegalArgumentException {
         registerServiceWatcher(listener.getServiceName());
-        dispatcher.addEventListener(listener);
     }
 
     @Override
@@ -190,7 +189,7 @@ protected void registerServiceWatcher(String serviceName) {
                         .orElseGet(() -> {
                             ChildListener watchListener, prev;
                             prev = childListenerMap.putIfAbsent(serviceName, watchListener = (parentPath, currentChildren) ->
-                                    dispatcher.dispatch(new ServiceInstancesChangedEvent(serviceName, getInstances(serviceName))));
+                                    dispatchServiceInstancesChangedEvent(serviceName));
                             return prev != null ? prev : watchListener;
                         });
 

diff --git a/...o-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosServiceDiscovery.java b/...o-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosServiceDiscovery.java
@@ -21,7 +21,6 @@
 import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.registry.client.ServiceDiscovery;
 import org.apache.dubbo.registry.client.ServiceInstance;
-import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
 import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
 import org.apache.dubbo.registry.nacos.util.NacosNamingServiceUtils;
 
@@ -125,6 +124,6 @@ private void handleEvent(NamingEvent event, ServiceInstancesChangedListener list
                 .stream()
                 .map(NacosNamingServiceUtils::toServiceInstance)
                 .collect(Collectors.toList());
-        listener.onEvent(new ServiceInstancesChangedEvent(serviceName, serviceInstances));
+        dispatchServiceInstancesChangedEvent(serviceName, serviceInstances);
     }
 }
diff --git a/...ookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java b/...ookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java
@@ -148,7 +148,6 @@ public Page<ServiceInstance> getInstances(String serviceName, int offset, int pa
     public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener)
             throws NullPointerException, IllegalArgumentException {
         registerServiceWatcher(listener.getServiceName());
-        dispatcher.addEventListener(listener);
     }
 
     private void doInServiceRegistry(ThrowableConsumer<org.apache.curator.x.discovery.ServiceDiscovery> consumer) {
@@ -164,7 +163,7 @@ private <R> R doInServiceDiscovery(ThrowableFunction<org.apache.curator.x.discov
     protected void registerServiceWatcher(String serviceName) {
         String path = buildServicePath(serviceName);
         CuratorWatcher watcher = watcherCaches.computeIfAbsent(path, key ->
-                new ZookeeperServiceDiscoveryChangeWatcher(this, serviceName, dispatcher));
+                new ZookeeperServiceDiscoveryChangeWatcher(this, serviceName));
         try {
             curatorFramework.getChildren().usingWatcher(watcher).forPath(path);
         } catch (KeeperException.NoNodeException e) {

diff --git a/...main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscoveryChangeWatcher.java b/...main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscoveryChangeWatcher.java
@@ -16,17 +16,13 @@
  */
 package org.apache.dubbo.registry.zookeeper;
 
-import org.apache.dubbo.event.EventDispatcher;
 import org.apache.dubbo.registry.client.ServiceDiscovery;
-import org.apache.dubbo.registry.client.ServiceInstance;
 import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
 
 import org.apache.curator.framework.api.CuratorWatcher;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 
-import java.util.Collection;
-
 import static org.apache.zookeeper.Watcher.Event.EventType.NodeChildrenChanged;
 import static org.apache.zookeeper.Watcher.Event.EventType.NodeDataChanged;
 
@@ -43,13 +39,10 @@ public class ZookeeperServiceDiscoveryChangeWatcher implements CuratorWatcher {
 
     private final String serviceName;
 
-    private final EventDispatcher dispatcher;
-
     public ZookeeperServiceDiscoveryChangeWatcher(ZookeeperServiceDiscovery zookeeperServiceDiscovery,
-                                                  String serviceName, EventDispatcher dispatcher) {
+                                                  String serviceName) {
         this.zookeeperServiceDiscovery = zookeeperServiceDiscovery;
         this.serviceName = serviceName;
-        this.dispatcher = dispatcher;
     }
 
     @Override
@@ -58,11 +51,7 @@ public void process(WatchedEvent event) throws Exception {
         Watcher.Event.EventType eventType = event.getType();
 
         if (NodeChildrenChanged.equals(eventType) || NodeDataChanged.equals(eventType)) {
-            dispatcher.dispatch(new ServiceInstancesChangedEvent(serviceName, getServiceInstances(serviceName)));
+            zookeeperServiceDiscovery.dispatchServiceInstancesChangedEvent(serviceName);
         }
     }
-
-    private Collection<ServiceInstance> getServiceInstances(String serviceName) {
-        return zookeeperServiceDiscovery.getInstances(serviceName);
-    }
 }