fix:优化dubbo接口级注册发现的内存使用 (#40)

polarismesh · Jan 19, 2024 · 1789f14 · 1789f14
1 parent 16f8499
commit 1789f14
Show file tree

Hide file tree

Showing 3 changed files with 74 additions and 21 deletions.
diff --git a/...bo-registry-polaris/src/main/java/com/tencent/polaris/dubbo/registry/PolarisRegistry.java b/...bo-registry-polaris/src/main/java/com/tencent/polaris/dubbo/registry/PolarisRegistry.java
@@ -19,6 +19,7 @@
 import com.tencent.polaris.api.exception.PolarisException;
 import com.tencent.polaris.api.listener.ServiceListener;
 import com.tencent.polaris.api.pojo.Instance;
+import com.tencent.polaris.api.pojo.ServiceChangeEvent;
 import com.tencent.polaris.api.utils.StringUtils;
 import com.tencent.polaris.common.registry.BaseBootConfigHandler;
 import com.tencent.polaris.common.registry.BootConfigHandler;
@@ -59,8 +60,9 @@ public class PolarisRegistry extends FailbackRegistry {
 
     private final AtomicBoolean destroyed = new AtomicBoolean(false);
 
-    private final Map<NotifyListener, ServiceListener> serviceListeners = new ConcurrentHashMap<>();
+    private final Map<URL, Set<NotifyListener>> dubboListeners = new ConcurrentHashMap<>();
 
+    private final Map<URL, ServiceListener> serviceListeners = new ConcurrentHashMap<>();
     private final PolarisOperator polarisOperator;
 
     public PolarisRegistry(URL url) {
@@ -128,15 +130,12 @@ public void doSubscribe(URL url, NotifyListener listener) {
         Instance[] instances = polarisOperator.getAvailableInstances(service, true);
         onInstances(url, listener, instances);
         LOGGER.info("[POLARIS] submit watch task for service {}", service);
-        serviceListeners.computeIfAbsent(listener, notifyListener -> {
-            ServiceListener serviceListener = event -> {
-                try {
-                    Instance[] curInstances = polarisOperator.getAvailableInstances(service, true);
-                    onInstances(url, listener, curInstances);
-                } catch (PolarisException e) {
-                    LOGGER.error("[POLARIS] fail to fetch instances for service {}: {}", service, e.toString());
-                }
-            };
+
+        dubboListeners.computeIfAbsent(url, s -> new ConcurrentHashSet<>());
+        dubboListeners.get(url).add(listener);
+
+        serviceListeners.computeIfAbsent(url, dubboUrl -> {
+            ServiceListener serviceListener = new DubboServiceListener(url, this);
             polarisOperator.watchService(service, serviceListener);
             return serviceListener;
         });
@@ -208,4 +207,32 @@ public void doUnsubscribe(URL url, NotifyListener listener) {
     public boolean isAvailable() {
         return true;
     }
+
+    private static class DubboServiceListener implements ServiceListener {
+
+        private final URL url;
+
+        private final String service;
+
+        private final PolarisRegistry registry;
+
+        private DubboServiceListener(URL url, PolarisRegistry registry) {
+            this.url = url;
+            this.service = url.getServiceInterface();
+            this.registry = registry;
+        }
+
+        @Override
+        public void onEvent(ServiceChangeEvent serviceChangeEvent) {
+            try {
+                Set<NotifyListener> listeners = registry.dubboListeners.getOrDefault(url, Collections.emptySet());
+                Instance[] curInstances = registry.polarisOperator.getAvailableInstances(service, true);
+                for (NotifyListener listener : listeners) {
+                    registry.onInstances(url, listener, curInstances);
+                }
+            } catch (PolarisException e) {
+                LOGGER.error("[POLARIS] fail to fetch instances for service {}: {}", service, e.toString());
+            }
+        }
+    }
 }
diff --git a/...x-registry-polaris/src/main/java/com/tencent/polaris/dubbox/registry/PolarisRegistry.java b/...x-registry-polaris/src/main/java/com/tencent/polaris/dubbox/registry/PolarisRegistry.java
@@ -28,6 +28,7 @@
 import com.tencent.polaris.api.exception.PolarisException;
 import com.tencent.polaris.api.listener.ServiceListener;
 import com.tencent.polaris.api.pojo.Instance;
+import com.tencent.polaris.api.pojo.ServiceChangeEvent;
 import com.tencent.polaris.api.utils.StringUtils;
 import com.tencent.polaris.common.registry.Consts;
 import com.tencent.polaris.common.registry.ConvertUtils;
@@ -60,7 +61,9 @@ public class PolarisRegistry extends FailbackRegistry {
 
     private final AtomicBoolean destroyed = new AtomicBoolean(false);
 
-    private final Map<NotifyListener, ServiceListener> serviceListeners = new ConcurrentHashMap<>();
+    private final Map<URL, Set<NotifyListener>> dubboListeners = new ConcurrentHashMap<>();
+
+    private final Map<URL, ServiceListener> serviceListeners = new ConcurrentHashMap<>();
 
     private final PolarisOperator polarisOperator;
 
@@ -125,15 +128,11 @@ public void doSubscribe(URL url, NotifyListener listener) {
         onInstances(url, listener, instances);
         LOGGER.info("[POLARIS] submit watch task for service {}", service);
 
-        serviceListeners.computeIfAbsent(listener, notifyListener -> {
-            ServiceListener serviceListener = event -> {
-                try {
-                    Instance[] curInstances = polarisOperator.getAvailableInstances(service, true);
-                    onInstances(url, listener, curInstances);
-                } catch (PolarisException e) {
-                    LOGGER.error("[POLARIS] fail to fetch instances for service {}: {}", service, e.toString());
-                }
-            };
+        dubboListeners.computeIfAbsent(url, s -> new ConcurrentHashSet<>());
+        dubboListeners.get(url).add(listener);
+
+        serviceListeners.computeIfAbsent(url, dubboUrl -> {
+            ServiceListener serviceListener = new DubboServiceListener(url, this);
             polarisOperator.watchService(service, serviceListener);
             return serviceListener;
         });
@@ -202,4 +201,32 @@ public void doUnsubscribe(URL url, NotifyListener listener) {
     public boolean isAvailable() {
         return true;
     }
+
+    private static class DubboServiceListener implements ServiceListener {
+
+        private final URL url;
+
+        private final String service;
+
+        private final PolarisRegistry registry;
+
+        private DubboServiceListener(URL url, PolarisRegistry registry) {
+            this.url = url;
+            this.service = url.getServiceInterface();
+            this.registry = registry;
+        }
+
+        @Override
+        public void onEvent(ServiceChangeEvent serviceChangeEvent) {
+            try {
+                Set<NotifyListener> listeners = registry.dubboListeners.getOrDefault(url, Collections.emptySet());
+                Instance[] curInstances = registry.polarisOperator.getAvailableInstances(service, true);
+                for (NotifyListener listener : listeners) {
+                    registry.onInstances(url, listener, curInstances);
+                }
+            } catch (PolarisException e) {
+                LOGGER.error("[POLARIS] fail to fetch instances for service {}: {}", service, e.toString());
+            }
+        }
+    }
 }
diff --git a/dubbox/pom.xml b/dubbox/pom.xml
@@ -11,7 +11,6 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>dubbox</artifactId>
-
     <packaging>pom</packaging>
 
     <properties>