[Dubbo-6720] fix bug same interface unexport and export fail. also support hotload service

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java

@@ -1,116 +1,148 @@
-/*
- * 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.rpc.protocol;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.logger.Logger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.ConcurrentHashSet;
-import org.apache.dubbo.remoting.Constants;
-import org.apache.dubbo.rpc.Exporter;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.Protocol;
-import org.apache.dubbo.rpc.ProtocolServer;
-import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.support.ProtocolUtils;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
-import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
-import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
-
-/**
- * abstract ProtocolSupport.
- */
-public abstract class AbstractProtocol implements Protocol {
-
-    protected final Logger logger = LoggerFactory.getLogger(getClass());
-
-    protected final Map<String, Exporter<?>> exporterMap = new ConcurrentHashMap<String, Exporter<?>>();
-
-    /**
-     * <host:port, ProtocolServer>
-     */
-    protected final Map<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();
-
-    //TODO SoftReference
-    protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();
-
-    protected static String serviceKey(URL url) {
-        int port = url.getParameter(Constants.BIND_PORT_KEY, url.getPort());
-        return serviceKey(port, url.getPath(), url.getParameter(VERSION_KEY), url.getParameter(GROUP_KEY));
-    }
-
-    protected static String serviceKey(int port, String serviceName, String serviceVersion, String serviceGroup) {
-        return ProtocolUtils.serviceKey(port, serviceName, serviceVersion, serviceGroup);
-    }
-
-    public List<ProtocolServer> getServers() {
-        return Collections.unmodifiableList(new ArrayList<>(serverMap.values()));
-    }
-
-    @Override
-    public void destroy() {
-        for (Invoker<?> invoker : invokers) {
-            if (invoker != null) {
-                invokers.remove(invoker);
-                try {
-                    if (logger.isInfoEnabled()) {
-                        logger.info("Destroy reference: " + invoker.getUrl());
-                    }
-                    invoker.destroy();
-                } catch (Throwable t) {
-                    logger.warn(t.getMessage(), t);
-                }
-            }
-        }
-        for (String key : new ArrayList<String>(exporterMap.keySet())) {
-            Exporter<?> exporter = exporterMap.remove(key);
-            if (exporter != null) {
-                try {
-                    if (logger.isInfoEnabled()) {
-                        logger.info("Unexport service: " + exporter.getInvoker().getUrl());
-                    }
-                    exporter.unexport();
-                } catch (Throwable t) {
-                    logger.warn(t.getMessage(), t);
-                }
-            }
-        }
-    }
-
-    @Override
-    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
-        return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
-    }
-
-    protected abstract <T> Invoker<T> protocolBindingRefer(Class<T> type, URL url) throws RpcException;
-
-    public Map<String, Exporter<?>> getExporterMap() {
-        return exporterMap;
-    }
-
-    public Collection<Exporter<?>> getExporters() {
-        return Collections.unmodifiableCollection(exporterMap.values());
-    }
-}
+/*
+ * 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.rpc.protocol;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.utils.ConcurrentHashSet;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.rpc.Exporter;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Protocol;
+import org.apache.dubbo.rpc.ProtocolServer;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.support.ProtocolUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+
+/**
+ * abstract ProtocolSupport.
+ */
+public abstract class AbstractProtocol implements Protocol {
+
+    protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+    protected final Map<String, Exporter<?>> exporterMap = new ConcurrentHashMap<String, Exporter<?>>();
+
+    protected final DelegateExporterMap delegateExporterMap = new DelegateExporterMap() {
+        @Override
+        public boolean isEmpty() {
+            return CollectionUtils.isEmptyMap(exporterMap);
+        }
+
+        @Override
+        public Exporter<?> getExport(String key) {
+            return exporterMap.get(key);
+        }
+
+        @Override
+        public void addExportMap(String key, Exporter<?> exporter) {
+            exporterMap.put(key, exporter);
+        }
+
+        @Override
+        public void removeExportMap(String key, Exporter<?> exporter) {
+            Exporter<?> findExporter = exporterMap.get(key);
+            if(findExporter == exporter){
+                exporterMap.remove(key);
+            }
+        }
+
+        @Override
+        public Collection<Exporter<?>> getExporters() {
+            return Collections.unmodifiableCollection(exporterMap.values());
+        }
+    };
+
+    /**
+     * <host:port, ProtocolServer>
+     */
+    protected final Map<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();
+
+    //TODO SoftReference
+    protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();
+
+    protected static String serviceKey(URL url) {
+        int port = url.getParameter(Constants.BIND_PORT_KEY, url.getPort());
+        return serviceKey(port, url.getPath(), url.getParameter(VERSION_KEY), url.getParameter(GROUP_KEY));
+    }
+
+    protected static String serviceKey(int port, String serviceName, String serviceVersion, String serviceGroup) {
+        return ProtocolUtils.serviceKey(port, serviceName, serviceVersion, serviceGroup);
+    }
+
+    @Override
+    public List<ProtocolServer> getServers() {
+        return Collections.unmodifiableList(new ArrayList<>(serverMap.values()));
+    }
+
+    @Override
+    public void destroy() {
+        for (Invoker<?> invoker : invokers) {
+            if (invoker != null) {
+                invokers.remove(invoker);
+                try {
+                    if (logger.isInfoEnabled()) {
+                        logger.info("Destroy reference: " + invoker.getUrl());
+                    }
+                    invoker.destroy();
+                } catch (Throwable t) {
+                    logger.warn(t.getMessage(), t);
+                }
+            }
+        }
+        for (String key : new ArrayList<String>(exporterMap.keySet())) {
+            Exporter<?> exporter = exporterMap.remove(key);
+            if (exporter != null) {
+                try {
+                    if (logger.isInfoEnabled()) {
+                        logger.info("Unexport service: " + exporter.getInvoker().getUrl());
+                    }
+                    exporter.unexport();
+                } catch (Throwable t) {
+                    logger.warn(t.getMessage(), t);
+                }
+            }
+        }
+    }
+
+    @Override
+    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
+        return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
+    }
+
+    protected abstract <T> Invoker<T> protocolBindingRefer(Class<T> type, URL url) throws RpcException;
+
+    public Map<String, Exporter<?>> getExporterMap() {
+        return exporterMap;
+    }
+
+    public Collection<Exporter<?>> getExporters() {
+        return Collections.unmodifiableCollection(exporterMap.values());
+    }
+}

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProxyProtocol.java

@@ -77,7 +77,7 @@ public void setProxyFactory(ProxyFactory proxyFactory) {
     @SuppressWarnings("unchecked")
     public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
         final String uri = serviceKey(invoker.getUrl());
-        Exporter<T> exporter = (Exporter<T>) exporterMap.get(uri);
+        Exporter<T> exporter = (Exporter<T>) delegateExporterMap.getExport(uri);
         if (exporter != null) {
             // When modifying the configuration through override, you need to re-expose the newly modified service.
             if (Objects.equals(exporter.getInvoker().getUrl(), invoker.getUrl())) {
@@ -88,7 +88,7 @@ public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
         exporter = new AbstractExporter<T>(invoker) {
             @Override
             public void afterUnExport() {
-                exporterMap.remove(uri);
+                delegateExporterMap.removeExportMap(uri, this);
                 if (runnable != null) {
                     try {
                         runnable.run();
@@ -98,7 +98,7 @@ public void afterUnExport() {
                 }
             }
         };
-        exporterMap.put(uri, exporter);
+        delegateExporterMap.addExportMap(uri, exporter);
         return exporter;
     }
 
@@ -277,5 +277,4 @@ public boolean isClosed() {
         }
     }
 
-
 }

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/DelegateExporterMap.java

@@ -0,0 +1,58 @@
+/*
+ * 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.rpc.protocol;
+
+import org.apache.dubbo.rpc.Exporter;
+import java.util.Collection;
+
+/**
+ * delegate exportermap oper
+ */
+public interface DelegateExporterMap {
+
+    /**
+     * check is empty map
+     * @return
+     */
+    boolean isEmpty();
+
+    /**
+     * get export
+     * @param key
+     * @return
+     */
+    Exporter<?> getExport(String key);
+
+    /**
+     * add
+     * @param key
+     * @param exporter
+     */
+    void addExportMap(String key, Exporter<?> exporter);
+
+    /**
+     * delete
+     * @param key
+     */
+    void removeExportMap(String key, Exporter<?> exporter);
+
+    /**
+     * get all exports
+     * @return
+     */
+    Collection<Exporter<?>> getExporters();
+}

dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboExporter.java

@@ -16,11 +16,9 @@
  */
 package org.apache.dubbo.rpc.protocol.dubbo;
 
-import org.apache.dubbo.rpc.Exporter;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.protocol.AbstractExporter;
-
-import java.util.Map;
+import org.apache.dubbo.rpc.protocol.DelegateExporterMap;
 
 /**
  * DubboExporter
@@ -29,17 +27,16 @@ public class DubboExporter<T> extends AbstractExporter<T> {
 
     private final String key;
 
-    private final Map<String, Exporter<?>> exporterMap;
+    private final DelegateExporterMap delegateExporterMap;
 
-    public DubboExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap) {
+    public DubboExporter(Invoker<T> invoker, String key, DelegateExporterMap delegateExporterMap) {
         super(invoker);
         this.key = key;
-        this.exporterMap = exporterMap;
+        this.delegateExporterMap = delegateExporterMap;
     }
 
     @Override
     public void afterUnExport() {
-        exporterMap.remove(key);
+        delegateExporterMap.removeExportMap(key, this);
     }
-
 }