Multiple registry (#4066)

fixes #3932, #3599, #3084

dubbo-all/pom.xml

@@ -282,6 +282,13 @@
             <scope>compile</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-registry-multiple</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-monitor-api</artifactId>
@@ -596,6 +603,7 @@
                                     <include>org.apache.dubbo:dubbo-registry-etcd3</include>
                                     <include>org.apache.dubbo:dubbo-registry-nacos</include>
                                     <include>org.apache.dubbo:dubbo-registry-sofa</include>
+                                    <include>org.apache.dubbo:dubbo-registry-multiple</include>
                                     <include>org.apache.dubbo:dubbo-monitor-api</include>
                                     <include>org.apache.dubbo:dubbo-monitor-default</include>
                                     <include>org.apache.dubbo:dubbo-config-api</include>

dubbo-registry/dubbo-registry-multiple/pom.xml

@@ -0,0 +1,60 @@
+<!--
+  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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.dubbo</groupId>
+        <artifactId>dubbo-registry</artifactId>
+        <version>${revision}</version>
+    </parent>
+    <artifactId>dubbo-registry-multiple</artifactId>
+    <packaging>jar</packaging>
+    <name>${project.artifactId}</name>
+    <description>The multiple registry module of dubbo project</description>
+    <properties>
+        <skip_maven_deploy>false</skip_maven_deploy>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-registry-api</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-registry-zookeeper</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-registry-redis</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.github.kstyrc</groupId>
+            <artifactId>embedded-redis</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistry.java

@@ -0,0 +1,314 @@
+/*
+ * 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.multiple;
+
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.registry.NotifyListener;
+import org.apache.dubbo.registry.Registry;
+import org.apache.dubbo.registry.RegistryFactory;
+import org.apache.dubbo.registry.support.AbstractRegistry;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.dubbo.common.constants.RegistryConstants.EMPTY_PROTOCOL;
+
+/**
+ * MultipleRegistry
+ */
+public class MultipleRegistry extends AbstractRegistry {
+
+    public static final String REGISTRY_FOR_SERVICE = "service-registry";
+    public static final String REGISTRY_FOR_REFERENCE = "reference-registry";
+
+    protected RegistryFactory registryFactory = ExtensionLoader.getExtensionLoader(RegistryFactory.class).getAdaptiveExtension();
+    private final Map<String, Registry> serviceRegistries = new ConcurrentHashMap<>(4);
+    private final Map<String, Registry> referenceRegistries = new ConcurrentHashMap<String, Registry>(4);
+    private final Map<NotifyListener, MultipleNotifyListenerWrapper> multipleNotifyListenerMap = new ConcurrentHashMap<NotifyListener, MultipleNotifyListenerWrapper>(32);
+    protected List<String> origServiceRegistryURLs;
+    protected List<String> origReferenceRegistryURLs;
+    protected List<String> effectServiceRegistryURLs;
+    protected List<String> effectReferenceRegistryURLs;
+    private URL registryUrl;
+    private String applicationName;
+
+
+    public MultipleRegistry(URL url) {
+        super(url);
+        this.registryUrl = url;
+        this.applicationName = url.getParameter(CommonConstants.APPLICATION_KEY);
+        init();
+        checkApplicationName(this.applicationName);
+        // This urls contain parameter and it donot inherit from the parameter of url in MultipleRegistry
+        origServiceRegistryURLs = url.getParameter(REGISTRY_FOR_SERVICE, new ArrayList<String>());
+        origReferenceRegistryURLs = url.getParameter(REGISTRY_FOR_REFERENCE, new ArrayList<String>());
+        effectServiceRegistryURLs = this.filterServiceRegistry(origServiceRegistryURLs);
+        effectReferenceRegistryURLs = this.filterReferenceRegistry(origReferenceRegistryURLs);
+
+        boolean defaultRegistry = url.getParameter(CommonConstants.DEFAULT_KEY, true);
+        if (defaultRegistry && effectServiceRegistryURLs.isEmpty() && effectReferenceRegistryURLs.isEmpty()) {
+            throw new IllegalArgumentException("Illegal registry url. You need to configure parameter " +
+                    REGISTRY_FOR_SERVICE + " or " + REGISTRY_FOR_REFERENCE);
+        }
+        Set<String> allURLs = new HashSet<String>(effectServiceRegistryURLs);
+        allURLs.addAll(effectReferenceRegistryURLs);
+        Map<String, Registry> tmpMap = new HashMap<String, Registry>(4);
+        for (String tmpUrl : allURLs) {
+            tmpMap.put(tmpUrl, registryFactory.getRegistry(URL.valueOf(tmpUrl)));
+        }
+        for (String serviceRegistyURL : effectServiceRegistryURLs) {
+            serviceRegistries.put(serviceRegistyURL, tmpMap.get(serviceRegistyURL));
+        }
+        for (String referenceReigstyURL : effectReferenceRegistryURLs) {
+            referenceRegistries.put(referenceReigstyURL, tmpMap.get(referenceReigstyURL));
+        }
+    }
+
+
+    @Override
+    public URL getUrl() {
+        return registryUrl;
+    }
+
+    @Override
+    public boolean isAvailable() {
+        boolean available = serviceRegistries.isEmpty() ? true : false;
+        for (Registry serviceRegistry : serviceRegistries.values()) {
+            if (serviceRegistry.isAvailable()) {
+                available = true;
+            }
+        }
+        if (!available) {
+            return false;
+        }
+
+        available = referenceRegistries.isEmpty() ? true : false;
+        for (Registry referenceRegistry : referenceRegistries.values()) {
+            if (referenceRegistry.isAvailable()) {
+                available = true;
+            }
+        }
+        if (!available) {
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public void destroy() {
+        Set<Registry> registries = new HashSet<Registry>(serviceRegistries.values());
+        registries.addAll(referenceRegistries.values());
+        for (Registry registry : registries) {
+            registry.destroy();
+        }
+    }
+
+    @Override
+    public void register(URL url) {
+        super.register(url);
+        for (Registry registry : serviceRegistries.values()) {
+            registry.register(url);
+        }
+    }
+
+    @Override
+    public void unregister(URL url) {
+        super.unregister(url);
+        for (Registry registry : serviceRegistries.values()) {
+            registry.unregister(url);
+        }
+    }
+
+    @Override
+    public void subscribe(URL url, NotifyListener listener) {
+        MultipleNotifyListenerWrapper multipleNotifyListenerWrapper = new MultipleNotifyListenerWrapper(listener);
+        multipleNotifyListenerMap.put(listener, multipleNotifyListenerWrapper);
+        for (Registry registry : referenceRegistries.values()) {
+            SingleNotifyListener singleNotifyListener = new SingleNotifyListener(multipleNotifyListenerWrapper, registry);
+            multipleNotifyListenerWrapper.putRegistryMap(registry.getUrl(), singleNotifyListener);
+            registry.subscribe(url, singleNotifyListener);
+        }
+        super.subscribe(url, multipleNotifyListenerWrapper);
+    }
+
+    @Override
+    public void unsubscribe(URL url, NotifyListener listener) {
+        MultipleNotifyListenerWrapper notifyListener = multipleNotifyListenerMap.remove(listener);
+        for (Registry registry : referenceRegistries.values()) {
+            SingleNotifyListener singleNotifyListener = notifyListener.registryMap.get(registry.getUrl());
+            registry.unsubscribe(url, singleNotifyListener);
+        }
+
+        if (notifyListener != null) {
+            super.unsubscribe(url, notifyListener);
+            notifyListener.destroy();
+        }
+    }
+
+    @Override
+    public List<URL> lookup(URL url) {
+        List<URL> urls = new ArrayList<URL>();
+        for (Registry registry : referenceRegistries.values()) {
+            List<URL> tmpUrls = registry.lookup(url);
+            if (!CollectionUtils.isEmpty(tmpUrls)) {
+                urls.addAll(tmpUrls);
+            }
+        }
+        return urls;
+    }
+
+    protected void init() {
+    }
+
+    protected List<String> filterServiceRegistry(List<String> serviceRegistryURLs) {
+        return serviceRegistryURLs;
+    }
+
+    protected List<String> filterReferenceRegistry(List<String> referenceRegistryURLs) {
+        return referenceRegistryURLs;
+    }
+
+
+    protected void checkApplicationName(String applicationName) {
+    }
+
+    protected String getApplicationName() {
+        return applicationName;
+    }
+
+    public Map<String, Registry> getServiceRegistries() {
+        return serviceRegistries;
+    }
+
+    public Map<String, Registry> getReferenceRegistries() {
+        return referenceRegistries;
+    }
+
+    public List<String> getOrigServiceRegistryURLs() {
+        return origServiceRegistryURLs;
+    }
+
+    public List<String> getOrigReferenceRegistryURLs() {
+        return origReferenceRegistryURLs;
+    }
+
+    public List<String> getEffectServiceRegistryURLs() {
+        return effectServiceRegistryURLs;
+    }
+
+    public List<String> getEffectReferenceRegistryURLs() {
+        return effectReferenceRegistryURLs;
+    }
+
+    static protected class MultipleNotifyListenerWrapper implements NotifyListener {
+
+        Map<URL, SingleNotifyListener> registryMap = new ConcurrentHashMap<URL, SingleNotifyListener>(4);
+        NotifyListener sourceNotifyListener;
+
+        public MultipleNotifyListenerWrapper(NotifyListener sourceNotifyListener) {
+            this.sourceNotifyListener = sourceNotifyListener;
+        }
+
+        public void putRegistryMap(URL registryURL, SingleNotifyListener singleNotifyListener) {
+            this.registryMap.put(registryURL, singleNotifyListener);
+        }
+
+        public void destroy() {
+            for (SingleNotifyListener singleNotifyListener : registryMap.values()) {
+                if (singleNotifyListener != null) {
+                    singleNotifyListener.destroy();
+                }
+            }
+            registryMap.clear();
+            sourceNotifyListener = null;
+        }
+
+        public synchronized void notifySourceListener() {
+            List<URL> notifyURLs = new ArrayList<URL>();
+            URL emptyURL = null;
+            for (SingleNotifyListener singleNotifyListener : registryMap.values()) {
+                List<URL> tmpUrls = singleNotifyListener.getUrlList();
+                if (CollectionUtils.isEmpty(tmpUrls)) {
+                    continue;
+                }
+                // empty protocol
+                if (tmpUrls.size() == 1
+                        && tmpUrls.get(0) != null
+                        && EMPTY_PROTOCOL.equals(tmpUrls.get(0).getProtocol())) {
+                    // if only one empty
+                    if (emptyURL == null) {
+                        emptyURL = tmpUrls.get(0);
+                    }
+                    continue;
+                }
+                notifyURLs.addAll(tmpUrls);
+            }
+            // if no notify URL, add empty protocol URL
+            if (emptyURL != null && notifyURLs.isEmpty()) {
+                notifyURLs.add(emptyURL);
+            }
+            this.notify(notifyURLs);
+        }
+
+        @Override
+        public void notify(List<URL> urls) {
+            sourceNotifyListener.notify(urls);
+        }
+
+        public Map<URL, SingleNotifyListener> getRegistryMap() {
+            return registryMap;
+        }
+    }
+
+    static protected class SingleNotifyListener implements NotifyListener {
+
+        MultipleNotifyListenerWrapper multipleNotifyListenerWrapper;
+        Registry registry;
+        volatile List<URL> urlList;
+
+        public SingleNotifyListener(MultipleNotifyListenerWrapper multipleNotifyListenerWrapper, Registry registry) {
+            this.registry = registry;
+            this.multipleNotifyListenerWrapper = multipleNotifyListenerWrapper;
+        }
+
+        @Override
+        public synchronized void notify(List<URL> urls) {
+            this.urlList = urls;
+            if (multipleNotifyListenerWrapper != null) {
+                this.multipleNotifyListenerWrapper.notifySourceListener();
+            }
+        }
+
+        public void destroy() {
+            this.multipleNotifyListenerWrapper = null;
+            this.registry = null;
+        }
+
+        public List<URL> getUrlList() {
+            return urlList;
+        }
+    }
+}

dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleRegistryFactory.java

@@ -0,0 +1,33 @@
+/*
+ * 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.multiple;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.registry.Registry;
+import org.apache.dubbo.registry.support.AbstractRegistryFactory;
+
+/**
+ * MultipleRegistryFactory
+ */
+public class MultipleRegistryFactory extends AbstractRegistryFactory {
+
+    @Override
+    protected Registry createRegistry(URL url) {
+        return new MultipleRegistry(url);
+    }
+
+}

dubbo-registry/dubbo-registry-multiple/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory

@@ -0,0 +1 @@
+multiple=org.apache.dubbo.registry.multiple.MultipleRegistryFactory

dubbo-registry/dubbo-registry-multiple/src/test/java/org/apache/dubbo/registry/multiple/MultipleRegistry2S2RTest.java

@@ -0,0 +1,206 @@
+/*
+ * 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.multiple;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.registry.NotifyListener;
+import org.apache.dubbo.registry.Registry;
+import org.apache.dubbo.registry.redis.RedisRegistry;
+import org.apache.dubbo.registry.zookeeper.ZookeeperRegistry;
+import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
+import org.apache.dubbo.remoting.zookeeper.curator.CuratorZookeeperClient;
+
+import org.apache.curator.test.TestingServer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import redis.embedded.RedisServer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 2019-04-30
+ */
+public class MultipleRegistry2S2RTest {
+
+    private static final String SERVICE_NAME = "org.apache.dubbo.registry.MultipleService2S2R";
+    private static final String SERVICE2_NAME = "org.apache.dubbo.registry.MultipleService2S2R2";
+
+    private static TestingServer zkServer;
+    private static RedisServer redisServer;
+    static int zkServerPort;
+    static int redisServerPort;
+
+    private static String zookeeperRegistryURLStr;
+    private static String redisRegistryURLStr;
+
+    private static MultipleRegistry multipleRegistry;
+    // for test content
+    private static ZookeeperClient zookeeperClient;
+
+    private static ZookeeperRegistry zookeeperRegistry;
+    private static RedisRegistry redisRegistry;
+
+
+    @BeforeAll
+    public static void setUp() throws Exception {
+        zkServerPort = NetUtils.getAvailablePort();
+        zkServer = new TestingServer(zkServerPort, true);
+        zookeeperRegistryURLStr = "zookeeper://127.0.0.1:" + zkServerPort;
+
+        redisServerPort = NetUtils.getAvailablePort();
+        redisServer = new RedisServer(redisServerPort);
+        redisServer.start();
+        redisRegistryURLStr = "redis://127.0.0.1:" + redisServerPort;
+
+
+        URL url = URL.valueOf("multiple://127.0.0.1?application=vic&" +
+                MultipleRegistry.REGISTRY_FOR_SERVICE + "=" + zookeeperRegistryURLStr + "," + redisRegistryURLStr + "&"
+                + MultipleRegistry.REGISTRY_FOR_REFERENCE + "=" + zookeeperRegistryURLStr + "," + redisRegistryURLStr);
+        multipleRegistry = (MultipleRegistry) new MultipleRegistryFactory().createRegistry(url);
+
+        // for test validation
+        zookeeperClient = new CuratorZookeeperClient(URL.valueOf(zookeeperRegistryURLStr));
+        zookeeperRegistry = MultipleRegistryTestUtil.getZookeeperRegistry(multipleRegistry.getServiceRegistries().values());
+        redisRegistry = MultipleRegistryTestUtil.getRedisRegistry(multipleRegistry.getServiceRegistries().values());
+    }
+
+    @AfterAll
+    public static void tearDown() throws Exception {
+        zkServer.stop();
+        redisServer.stop();
+    }
+
+    @Test
+    public void testParamConfig() {
+
+        Assertions.assertTrue(multipleRegistry.origReferenceRegistryURLs.size() == 2);
+        Assertions.assertTrue(multipleRegistry.origReferenceRegistryURLs.contains(zookeeperRegistryURLStr));
+        Assertions.assertTrue(multipleRegistry.origReferenceRegistryURLs.contains(redisRegistryURLStr));
+
+        Assertions.assertTrue(multipleRegistry.origServiceRegistryURLs.size() == 2);
+        Assertions.assertTrue(multipleRegistry.origServiceRegistryURLs.contains(zookeeperRegistryURLStr));
+        Assertions.assertTrue(multipleRegistry.origServiceRegistryURLs.contains(redisRegistryURLStr));
+
+        Assertions.assertTrue(multipleRegistry.effectReferenceRegistryURLs.size() == 2);
+        Assertions.assertTrue(multipleRegistry.effectReferenceRegistryURLs.contains(zookeeperRegistryURLStr));
+        Assertions.assertTrue(multipleRegistry.effectReferenceRegistryURLs.contains(redisRegistryURLStr));
+
+        Assertions.assertTrue(multipleRegistry.effectServiceRegistryURLs.size() == 2);
+        Assertions.assertTrue(multipleRegistry.effectServiceRegistryURLs.contains(zookeeperRegistryURLStr));
+        Assertions.assertTrue(multipleRegistry.effectServiceRegistryURLs.contains(redisRegistryURLStr));
+
+        Assertions.assertTrue(multipleRegistry.getServiceRegistries().containsKey(zookeeperRegistryURLStr));
+        Assertions.assertTrue(multipleRegistry.getServiceRegistries().containsKey(redisRegistryURLStr));
+        Assertions.assertTrue(multipleRegistry.getServiceRegistries().values().size() == 2);
+//        java.util.Iterator<Registry> registryIterable = multipleRegistry.getServiceRegistries().values().iterator();
+//        Registry firstRegistry = registryIterable.next();
+//        Registry secondRegistry = registryIterable.next();
+        Assertions.assertNotNull(MultipleRegistryTestUtil.getZookeeperRegistry(multipleRegistry.getServiceRegistries().values()));
+        Assertions.assertNotNull(MultipleRegistryTestUtil.getRedisRegistry(multipleRegistry.getServiceRegistries().values()));
+        Assertions.assertNotNull(MultipleRegistryTestUtil.getZookeeperRegistry(multipleRegistry.getReferenceRegistries().values()));
+        Assertions.assertNotNull(MultipleRegistryTestUtil.getRedisRegistry(multipleRegistry.getReferenceRegistries().values()));
+
+        Assertions.assertEquals(MultipleRegistryTestUtil.getZookeeperRegistry(multipleRegistry.getServiceRegistries().values()),
+                MultipleRegistryTestUtil.getZookeeperRegistry(multipleRegistry.getReferenceRegistries().values()));
+
+        Assertions.assertEquals(MultipleRegistryTestUtil.getRedisRegistry(multipleRegistry.getServiceRegistries().values()),
+                MultipleRegistryTestUtil.getRedisRegistry(multipleRegistry.getReferenceRegistries().values()));
+
+        Assertions.assertEquals(multipleRegistry.getApplicationName(), "vic");
+
+        Assertions.assertTrue(multipleRegistry.isAvailable());
+    }
+
+    @Test
+    public void testRegistryAndUnRegistry() throws InterruptedException {
+        URL serviceUrl = URL.valueOf("http2://multiple/" + SERVICE_NAME + "?notify=false&methods=test1,test2&category=providers");
+//        URL serviceUrl2 = URL.valueOf("http2://multiple2/" + SERVICE_NAME + "?notify=false&methods=test1,test2&category=providers");
+        multipleRegistry.register(serviceUrl);
+
+        String path = "/dubbo/" + SERVICE_NAME + "/providers";
+        List<String> providerList = zookeeperClient.getChildren(path);
+        Assertions.assertTrue(!providerList.isEmpty());
+        System.out.println(providerList.get(0));
+
+        Assertions.assertNotNull(MultipleRegistryTestUtil.getRedisHashContent(redisServerPort, path, serviceUrl.toFullString()));
+
+        final List<URL> list = new ArrayList<URL>();
+        multipleRegistry.subscribe(serviceUrl, new NotifyListener() {
+            @Override
+            public void notify(List<URL> urls) {
+                System.out.println("invoke notify: " + urls);
+                list.clear();
+                list.addAll(urls);
+            }
+        });
+        Thread.sleep(1500);
+        Assertions.assertTrue(list.size() == 2);
+
+        multipleRegistry.unregister(serviceUrl);
+        Thread.sleep(1500);
+        Assertions.assertTrue(list.size() == 1);
+        List<URL> urls = MultipleRegistryTestUtil.getProviderURLsFromNotifyURLS(list);
+        Assertions.assertTrue(list.size() == 1);
+        Assertions.assertTrue("empty".equals(list.get(0).getProtocol()));
+    }
+
+    @Test
+    public void testSubscription() throws InterruptedException {
+        URL serviceUrl = URL.valueOf("http2://multiple/" + SERVICE2_NAME + "?notify=false&methods=test1,test2&category=providers");
+//        URL serviceUrl2 = URL.valueOf("http2://multiple2/" + SERVICE_NAME + "?notify=false&methods=test1,test2&category=providers");
+        multipleRegistry.register(serviceUrl);
+
+        String path = "/dubbo/" + SERVICE2_NAME + "/providers";
+        List<String> providerList = zookeeperClient.getChildren(path);
+        Assertions.assertTrue(!providerList.isEmpty());
+        System.out.println(providerList.get(0));
+
+        Assertions.assertNotNull(MultipleRegistryTestUtil.getRedisHashContent(redisServerPort, path, serviceUrl.toFullString()));
+
+        final List<URL> list = new ArrayList<URL>();
+        multipleRegistry.subscribe(serviceUrl, new NotifyListener() {
+            @Override
+            public void notify(List<URL> urls) {
+                System.out.println("invoke notify: " + urls);
+                list.clear();
+                list.addAll(urls);
+            }
+        });
+        Thread.sleep(1500);
+        Assertions.assertTrue(list.size() == 2);
+
+        List<Registry> serviceRegistries = new ArrayList<Registry>(multipleRegistry.getServiceRegistries().values());
+        serviceRegistries.get(0).unregister(serviceUrl);
+        Thread.sleep(1500);
+        Assertions.assertTrue(list.size() == 1);
+        List<URL> urls = MultipleRegistryTestUtil.getProviderURLsFromNotifyURLS(list);
+        Assertions.assertTrue(list.size() == 1);
+        Assertions.assertTrue(!"empty".equals(list.get(0).getProtocol()));
+
+        serviceRegistries.get(1).unregister(serviceUrl);
+        Thread.sleep(1500);
+        Assertions.assertTrue(list.size() == 1);
+        urls = MultipleRegistryTestUtil.getProviderURLsFromNotifyURLS(list);
+        Assertions.assertTrue(list.size() == 1);
+        Assertions.assertTrue("empty".equals(list.get(0).getProtocol()));
+    }
+
+}

dubbo-registry/dubbo-registry-multiple/src/test/java/org/apache/dubbo/registry/multiple/MultipleRegistryTestUtil.java

@@ -0,0 +1,138 @@
+/*
+ * 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.multiple;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.common.utils.UrlUtils;
+import org.apache.dubbo.registry.Registry;
+import org.apache.dubbo.registry.redis.RedisRegistry;
+import org.apache.dubbo.registry.zookeeper.ZookeeperRegistry;
+import org.apache.dubbo.rpc.RpcException;
+
+import redis.clients.jedis.Jedis;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static org.apache.dubbo.common.constants.RegistryConstants.APP_DYNAMIC_CONFIGURATORS_CATEGORY;
+import static org.apache.dubbo.common.constants.RegistryConstants.CATEGORY_KEY;
+import static org.apache.dubbo.common.constants.RegistryConstants.COMPATIBLE_CONFIG_KEY;
+import static org.apache.dubbo.common.constants.RegistryConstants.CONFIGURATORS_CATEGORY;
+import static org.apache.dubbo.common.constants.RegistryConstants.DEFAULT_CATEGORY;
+import static org.apache.dubbo.common.constants.RegistryConstants.DYNAMIC_CONFIGURATORS_CATEGORY;
+import static org.apache.dubbo.common.constants.RegistryConstants.PROVIDERS_CATEGORY;
+import static org.apache.dubbo.common.constants.RegistryConstants.ROUTERS_CATEGORY;
+import static org.apache.dubbo.common.constants.RegistryConstants.ROUTE_PROTOCOL;
+
+/**
+ * 2019-05-13
+ */
+public class MultipleRegistryTestUtil {
+    public static ZookeeperRegistry getZookeeperRegistry(Collection<Registry> registryCollection) {
+        for (Registry registry : registryCollection) {
+            if (registry instanceof ZookeeperRegistry) {
+                return (ZookeeperRegistry) registry;
+            }
+        }
+        return null;
+    }
+
+    public static RedisRegistry getRedisRegistry(Collection<Registry> registryCollection) {
+        for (Registry registry : registryCollection) {
+            if (registry instanceof RedisRegistry) {
+                return (RedisRegistry) registry;
+            }
+        }
+        return null;
+    }
+
+    public static String getRedisContent(int port, String key) {
+        Jedis jedis = null;
+        try {
+            jedis = new Jedis("127.0.0.1", port);
+            return jedis.get(key);
+        } catch (Throwable e) {
+            throw new RpcException("Failed to put to redis . cause: " + e.getMessage(), e);
+        } finally {
+            if (jedis != null) {
+                jedis.close();
+            }
+        }
+    }
+
+    public static String getRedisHashContent(int port, String key, String field) {
+        Jedis jedis = null;
+        try {
+            jedis = new Jedis("127.0.0.1", port);
+            return jedis.hget(key, field);
+        } catch (Throwable e) {
+            throw new RpcException("Failed to put to redis . cause: " + e.getMessage(), e);
+        } finally {
+            if (jedis != null) {
+                jedis.close();
+            }
+        }
+    }
+
+    /**
+     * copy from @org.apache.dubbo.registry.integration.RegistryDirectory#notify(java.util.List)
+     *
+     * @param urls
+     * @return
+     */
+    public static List<URL> getProviderURLsFromNotifyURLS(List<URL> urls) {
+        Map<String, List<URL>> categoryUrls = urls.stream()
+                .filter(Objects::nonNull)
+                .filter(MultipleRegistryTestUtil::isValidCategory)
+                .filter(MultipleRegistryTestUtil::isNotCompatibleFor26x)
+                .collect(Collectors.groupingBy(url -> {
+                    if (UrlUtils.isConfigurator(url)) {
+                        return CONFIGURATORS_CATEGORY;
+                    } else if (UrlUtils.isRoute(url)) {
+                        return ROUTERS_CATEGORY;
+                    } else if (UrlUtils.isProvider(url)) {
+                        return PROVIDERS_CATEGORY;
+                    }
+                    return "";
+                }));
+
+        // providers
+        List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());
+        return providerURLs;
+
+    }
+
+    private static boolean isValidCategory(URL url) {
+        String category = url.getParameter(CATEGORY_KEY, DEFAULT_CATEGORY);
+        if ((ROUTERS_CATEGORY.equals(category) || ROUTE_PROTOCOL.equals(url.getProtocol())) ||
+                PROVIDERS_CATEGORY.equals(category) ||
+                CONFIGURATORS_CATEGORY.equals(category) || DYNAMIC_CONFIGURATORS_CATEGORY.equals(category) ||
+                APP_DYNAMIC_CONFIGURATORS_CATEGORY.equals(category)) {
+            return true;
+        }
+        return false;
+    }
+
+    private static boolean isNotCompatibleFor26x(URL url) {
+        return StringUtils.isEmpty(url.getParameter(COMPATIBLE_CONFIG_KEY));
+    }
+}

dubbo-registry/pom.xml

@@ -37,6 +37,7 @@
         <module>dubbo-registry-consul</module>
         <module>dubbo-registry-etcd3</module>
         <module>dubbo-registry-nacos</module>
+        <module>dubbo-registry-multiple</module>
         <module>dubbo-registry-sofa</module>
     </modules>
 </project>