Move BeatReactor into HttpClientProxy (#3394)

client/src/main/java/com/alibaba/nacos/client/naming/NacosNamingService.java

@@ -58,14 +58,10 @@ public class NacosNamingService implements NamingService {
      */
     private String namespace;
     
-    private String cacheDir;
-    
     private String logName;
     
     private HostReactor hostReactor;
     
-    private BeatReactor beatReactor;
-    
     private EventDispatcher eventDispatcher;
     
     private NamingHttpClientProxy serverProxy;
@@ -87,49 +83,16 @@ public class NacosNamingService implements NamingService {
         this.namespace = InitUtils.initNamespaceForNaming(properties);
         InitUtils.initSerialization();
         InitUtils.initWebRootContext();
-        initCacheDir();
         initLogName(properties);
         
         ServerListManager serverListManager = new ServerListManager(properties);
         this.eventDispatcher = new EventDispatcher();
         this.serverProxy = new NamingHttpClientProxy(this.namespace, serverListManager, properties);
-        this.beatReactor = new BeatReactor(this.serverProxy, initClientBeatThreadCount(properties));
+        this.hostReactor = new HostReactor(this.eventDispatcher, this.serverProxy, this.namespace, properties);
-        this.hostReactor = new HostReactor(this.eventDispatcher, this.serverProxy, beatReactor, this.cacheDir,
-                isLoadCacheAtStart(properties), initPollingThreadCount(properties));
         this.grpcClientProxy = new NamingGrpcClientProxy(namespace, hostReactor);
         grpcClientProxy.start(serverListManager);
     }
     
-    private int initClientBeatThreadCount(Properties properties) {
-        if (properties == null) {
-            return UtilAndComs.DEFAULT_CLIENT_BEAT_THREAD_COUNT;
-        }
-        
-        return ConvertUtils.toInt(properties.getProperty(PropertyKeyConst.NAMING_CLIENT_BEAT_THREAD_COUNT),
-                UtilAndComs.DEFAULT_CLIENT_BEAT_THREAD_COUNT);
-    }
-    
-    private int initPollingThreadCount(Properties properties) {
-        if (properties == null) {
-            
-            return UtilAndComs.DEFAULT_POLLING_THREAD_COUNT;
-        }
-        
-        return ConvertUtils.toInt(properties.getProperty(PropertyKeyConst.NAMING_POLLING_THREAD_COUNT),
-                UtilAndComs.DEFAULT_POLLING_THREAD_COUNT);
-    }
-    
-    private boolean isLoadCacheAtStart(Properties properties) {
-        boolean loadCacheAtStart = false;
-        if (properties != null && StringUtils
-                .isNotEmpty(properties.getProperty(PropertyKeyConst.NAMING_LOAD_CACHE_AT_START))) {
-            loadCacheAtStart = ConvertUtils
-                    .toBoolean(properties.getProperty(PropertyKeyConst.NAMING_LOAD_CACHE_AT_START));
-        }
-        
-        return loadCacheAtStart;
-    }
-    
     private void initLogName(Properties properties) {
         logName = System.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME);
         if (StringUtils.isEmpty(logName)) {
@@ -143,13 +106,6 @@ public class NacosNamingService implements NamingService {
         }
     }
     
-    private void initCacheDir() {
-        cacheDir = System.getProperty("com.alibaba.nacos.naming.cache.dir");
-        if (StringUtils.isEmpty(cacheDir)) {
-            cacheDir = System.getProperty("user.home") + "/nacos/naming/" + namespace;
-        }
-    }
-    
     @Override
     public void registerInstance(String serviceName, String ip, int port) throws NacosException {
         registerInstance(serviceName, ip, port, Constants.DEFAULT_CLUSTER_NAME);
@@ -186,10 +142,6 @@ public class NacosNamingService implements NamingService {
     @Override
     public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
         //        String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
-        //        if (instance.isEphemeral()) {
-        //            BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
-        //            beatReactor.addBeatInfo(groupedServiceName, beatInfo);
-        //        }
         //        serverProxy.registerService(groupedServiceName, groupName, instance);
         grpcClientProxy.registerService(serviceName, groupName, instance);
     }
@@ -227,10 +179,6 @@ public class NacosNamingService implements NamingService {
     
     @Override
     public void deregisterInstance(String serviceName, String groupName, Instance instance) throws NacosException {
-        //        if (instance.isEphemeral()) {
-        //            beatReactor.removeBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), instance.getIp(),
-        //                    instance.getPort());
-        //        }
         //        serverProxy.deregisterService(NamingUtils.getGroupedName(serviceName, groupName), instance);
         grpcClientProxy.deregisterService(serviceName, groupName, instance);
     }
@@ -279,8 +227,7 @@ public class NacosNamingService implements NamingService {
         
         ServiceInfo serviceInfo;
         if (subscribe) {
-            serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
+            serviceInfo = hostReactor.getServiceInfo(serviceName, groupName, StringUtils.join(clusters, ","));
-                    StringUtils.join(clusters, ","));
         } else {
             //            serviceInfo = hostReactor
             //                    .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),
@@ -341,12 +288,10 @@ public class NacosNamingService implements NamingService {
         
         ServiceInfo serviceInfo;
         if (subscribe) {
-            serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
+            serviceInfo = hostReactor.getServiceInfo(serviceName, groupName, StringUtils.join(clusters, ","));
-                    StringUtils.join(clusters, ","));
         } else {
             serviceInfo = hostReactor
-                    .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),
+                    .getServiceInfoDirectlyFromServer(serviceName, groupName, StringUtils.join(clusters, ","));
-                            StringUtils.join(clusters, ","));
         }
         return selectInstances(serviceInfo, healthy);
     }
@@ -411,13 +356,11 @@ public class NacosNamingService implements NamingService {
             boolean subscribe) throws NacosException {
         
         if (subscribe) {
-            return Balancer.RandomByWeight.selectHost(hostReactor
+            return Balancer.RandomByWeight
-                    .getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
+                    .selectHost(hostReactor.getServiceInfo(serviceName, groupName, StringUtils.join(clusters, ",")));
-                            StringUtils.join(clusters, ",")));
         } else {
             return Balancer.RandomByWeight.selectHost(hostReactor
-                    .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),
+                    .getServiceInfoDirectlyFromServer(serviceName, groupName, StringUtils.join(clusters, ",")));
-                            StringUtils.join(clusters, ",")));
         }
     }
     
@@ -505,12 +448,11 @@ public class NacosNamingService implements NamingService {
     }
     
     public BeatReactor getBeatReactor() {
-        return beatReactor;
+        return serverProxy.getBeatReactor();
     }
     
     @Override
     public void shutDown() throws NacosException {
-        beatReactor.shutdown();
         eventDispatcher.shutdown();
         hostReactor.shutdown();
         serverProxy.shutdown();

client/src/main/java/com/alibaba/nacos/client/naming/beat/BeatReactor.java

@@ -16,6 +16,7 @@
 
 package com.alibaba.nacos.client.naming.beat;
 
+import com.alibaba.nacos.api.PropertyKeyConst;
 import com.alibaba.nacos.api.common.Constants;
 import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.CommonParams;
@@ -26,11 +27,13 @@ import com.alibaba.nacos.client.monitor.MetricsMonitor;
 import com.alibaba.nacos.client.naming.remote.http.NamingHttpClientProxy;
 import com.alibaba.nacos.client.naming.utils.UtilAndComs;
 import com.alibaba.nacos.common.lifecycle.Closeable;
+import com.alibaba.nacos.common.utils.ConvertUtils;
 import com.alibaba.nacos.common.utils.JacksonUtils;
 import com.alibaba.nacos.common.utils.ThreadUtils;
 import com.fasterxml.jackson.databind.JsonNode;
 
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -55,11 +58,12 @@ public class BeatReactor implements Closeable {
     public final Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<String, BeatInfo>();
     
     public BeatReactor(NamingHttpClientProxy serverProxy) {
-        this(serverProxy, UtilAndComs.DEFAULT_CLIENT_BEAT_THREAD_COUNT);
+        this(serverProxy, null);
     }
     
-    public BeatReactor(NamingHttpClientProxy serverProxy, int threadCount) {
+    public BeatReactor(NamingHttpClientProxy serverProxy, Properties properties) {
         this.serverProxy = serverProxy;
+        int threadCount = initClientBeatThreadCount(properties);
         this.executorService = new ScheduledThreadPoolExecutor(threadCount, new ThreadFactory() {
             @Override
             public Thread newThread(Runnable r) {
@@ -71,6 +75,15 @@ public class BeatReactor implements Closeable {
         });
     }
     
+    private int initClientBeatThreadCount(Properties properties) {
+        if (properties == null) {
+            return UtilAndComs.DEFAULT_CLIENT_BEAT_THREAD_COUNT;
+        }
+        
+        return ConvertUtils.toInt(properties.getProperty(PropertyKeyConst.NAMING_CLIENT_BEAT_THREAD_COUNT),
+                UtilAndComs.DEFAULT_CLIENT_BEAT_THREAD_COUNT);
+    }
+    
     /**
      * Add beat information.
      *

client/src/main/java/com/alibaba/nacos/client/naming/core/HostReactor.java

@@ -16,17 +16,18 @@
 
 package com.alibaba.nacos.client.naming.core;
 
+import com.alibaba.nacos.api.PropertyKeyConst;
 import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.pojo.Instance;
 import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
+import com.alibaba.nacos.api.naming.utils.NamingUtils;
 import com.alibaba.nacos.client.monitor.MetricsMonitor;
 import com.alibaba.nacos.client.naming.backups.FailoverReactor;
-import com.alibaba.nacos.client.naming.beat.BeatInfo;
-import com.alibaba.nacos.client.naming.beat.BeatReactor;
 import com.alibaba.nacos.client.naming.cache.DiskCache;
 import com.alibaba.nacos.client.naming.remote.http.NamingHttpClientProxy;
 import com.alibaba.nacos.client.naming.utils.UtilAndComs;
 import com.alibaba.nacos.common.lifecycle.Closeable;
+import com.alibaba.nacos.common.utils.ConvertUtils;
 import com.alibaba.nacos.common.utils.JacksonUtils;
 import com.alibaba.nacos.common.utils.StringUtils;
 import com.alibaba.nacos.common.utils.ThreadUtils;
@@ -36,6 +37,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
@@ -67,25 +69,19 @@ public class HostReactor implements Closeable {
     
     private final EventDispatcher eventDispatcher;
     
-    private final BeatReactor beatReactor;
-    
     private final NamingHttpClientProxy serverProxy;
     
     private final FailoverReactor failoverReactor;
     
-    private final String cacheDir;
-    
     private final ScheduledExecutorService executor;
     
-    public HostReactor(EventDispatcher eventDispatcher, NamingHttpClientProxy serverProxy, BeatReactor beatReactor,
+    private String cacheDir;
-            String cacheDir) {
-        this(eventDispatcher, serverProxy, beatReactor, cacheDir, false, UtilAndComs.DEFAULT_POLLING_THREAD_COUNT);
-    }
     
-    public HostReactor(EventDispatcher eventDispatcher, NamingHttpClientProxy serverProxy, BeatReactor beatReactor,
+    public HostReactor(EventDispatcher eventDispatcher, NamingHttpClientProxy serverProxy, String namespace,
-            String cacheDir, boolean loadCacheAtStart, int pollingThreadCount) {
+            Properties properties) {
+        initCacheDir(namespace);
         // init executorService
-        this.executor = new ScheduledThreadPoolExecutor(pollingThreadCount, new ThreadFactory() {
+        this.executor = new ScheduledThreadPoolExecutor(initPollingThreadCount(properties), new ThreadFactory() {
             @Override
             public Thread newThread(Runnable r) {
                 Thread thread = new Thread(r);
@@ -95,10 +91,8 @@ public class HostReactor implements Closeable {
             }
         });
         this.eventDispatcher = eventDispatcher;
-        this.beatReactor = beatReactor;
         this.serverProxy = serverProxy;
-        this.cacheDir = cacheDir;
+        if (isLoadCacheAtStart(properties)) {
-        if (loadCacheAtStart) {
             this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(DiskCache.read(this.cacheDir));
         } else {
             this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(16);
@@ -109,6 +103,31 @@ public class HostReactor implements Closeable {
         this.pushReceiver = new PushReceiver(this);
     }
     
+    private void initCacheDir(String namespace) {
+        cacheDir = System.getProperty("com.alibaba.nacos.naming.cache.dir");
+        if (StringUtils.isEmpty(cacheDir)) {
+            cacheDir = System.getProperty("user.home") + "/nacos/naming/" + namespace;
+        }
+    }
+    
+    private int initPollingThreadCount(Properties properties) {
+        if (properties == null) {
+            return UtilAndComs.DEFAULT_POLLING_THREAD_COUNT;
+        }
+        return ConvertUtils.toInt(properties.getProperty(PropertyKeyConst.NAMING_POLLING_THREAD_COUNT),
+                UtilAndComs.DEFAULT_POLLING_THREAD_COUNT);
+    }
+    
+    private boolean isLoadCacheAtStart(Properties properties) {
+        boolean loadCacheAtStart = false;
+        if (properties != null && StringUtils
+                .isNotEmpty(properties.getProperty(PropertyKeyConst.NAMING_LOAD_CACHE_AT_START))) {
+            loadCacheAtStart = ConvertUtils
+                    .toBoolean(properties.getProperty(PropertyKeyConst.NAMING_LOAD_CACHE_AT_START));
+        }
+        return loadCacheAtStart;
+    }
+    
     public Map<String, ServiceInfo> getServiceInfoMap() {
         return serviceInfoMap;
     }
@@ -123,10 +142,10 @@ public class HostReactor implements Closeable {
      * @param json service json
      * @return service info
      */
-    public ServiceInfo processServiceJson(String json) {
+    public ServiceInfo processServiceInfo(String json) {
         ServiceInfo serviceInfo = JacksonUtils.toObj(json, ServiceInfo.class);
         serviceInfo.setJsonFromServer(json);
-        return processServiceJson(serviceInfo);
+        return processServiceInfo(serviceInfo);
     }
     
     /**
@@ -135,7 +154,7 @@ public class HostReactor implements Closeable {
      * @param serviceInfo new service info
      * @return service info
      */
-    public ServiceInfo processServiceJson(ServiceInfo serviceInfo) {
+    public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
         ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());
         if (serviceInfo.getHosts() == null || !serviceInfo.validate()) {
             //empty or error push, just ignore
@@ -210,7 +229,7 @@ public class HostReactor implements Closeable {
             
             if (modHosts.size() > 0) {
                 changed = true;
-                updateBeatInfo(modHosts);
+                serverProxy.updateBeatInfo(modHosts);
                 NAMING_LOGGER.info("modified ips(" + modHosts.size() + ") service: " + serviceInfo.getKey() + " -> "
                         + JacksonUtils.toJson(modHosts));
             }
@@ -243,52 +262,39 @@ public class HostReactor implements Closeable {
         return serviceInfo;
     }
     
-    private void updateBeatInfo(Set<Instance> modHosts) {
+    private ServiceInfo getServiceInfo0(String groupedServiceName, String clusters) {
-        for (Instance instance : modHosts) {
-            String key = beatReactor.buildKey(instance.getServiceName(), instance.getIp(), instance.getPort());
-            if (beatReactor.dom2Beat.containsKey(key) && instance.isEphemeral()) {
-                BeatInfo beatInfo = beatReactor.buildBeatInfo(instance);
-                beatReactor.addBeatInfo(instance.getServiceName(), beatInfo);
-            }
-        }
-    }
         
-    private ServiceInfo getServiceInfo0(String serviceName, String clusters) {
+        String key = ServiceInfo.getKey(groupedServiceName, clusters);
-        
-        String key = ServiceInfo.getKey(serviceName, clusters);
         
         return serviceInfoMap.get(key);
     }
     
-    public ServiceInfo getServiceInfoDirectlyFromServer(final String serviceName, final String clusters)
+    public ServiceInfo getServiceInfoDirectlyFromServer(final String serviceName, final String groupName,
-            throws NacosException {
+            final String clusters) throws NacosException {
-        String result = serverProxy.queryList(serviceName, clusters, 0, false);
+        return serverProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false);
-        if (StringUtils.isNotEmpty(result)) {
-            return JacksonUtils.toObj(result, ServiceInfo.class);
-        }
-        return null;
     }
     
-    public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
+    public ServiceInfo getServiceInfo(final String serviceName, final String groupName, final String clusters) {
         
         NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
-        String key = ServiceInfo.getKey(serviceName, clusters);
+        String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
+        String key = ServiceInfo.getKey(groupedServiceName, clusters);
         if (failoverReactor.isFailoverSwitch()) {
             return failoverReactor.getService(key);
         }
         
-        ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);
+        ServiceInfo serviceObj = getServiceInfo0(groupedServiceName, clusters);
         
         if (null == serviceObj) {
-            serviceObj = new ServiceInfo(serviceName, clusters);
+            serviceObj = new ServiceInfo(groupedServiceName, clusters);
             
             serviceInfoMap.put(serviceObj.getKey(), serviceObj);
             
-            updatingService(serviceName);
+            updatingService(groupedServiceName);
-            updateServiceNow(serviceName, clusters);
+            updateServiceNow(serviceName, groupName, clusters);
-            finishUpdating(serviceName);
+            finishUpdating(groupedServiceName);
             
-        } else if (updatingMap.containsKey(serviceName)) {
+        } else if (updatingMap.containsKey(groupedServiceName)) {
             
             if (UPDATE_HOLD_INTERVAL > 0) {
                 // hold a moment waiting for update finish
@@ -303,7 +309,7 @@ public class HostReactor implements Closeable {
             }
         }
         
-        scheduleUpdateIfAbsent(serviceName, clusters);
+        scheduleUpdateIfAbsent(serviceName, groupName, clusters);
         
         return serviceInfoMap.get(serviceObj.getKey());
     }
@@ -312,20 +318,22 @@ public class HostReactor implements Closeable {
      * Schedule update if absent.
      *
      * @param serviceName service name
+     * @param groupName   group name
      * @param clusters    clusters
      */
-    public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
+    public void scheduleUpdateIfAbsent(String serviceName, String groupName, String clusters) {
-        if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
+        String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters);
+        if (futureMap.get(serviceKey) != null) {
             return;
         }
         
         synchronized (futureMap) {
-            if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
+            if (futureMap.get(serviceKey) != null) {
                 return;
             }
             
-            ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters));
+            ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, groupName, clusters));
-            futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
+            futureMap.put(serviceKey, future);
         }
     }
     
@@ -333,16 +341,16 @@ public class HostReactor implements Closeable {
      * Update service now.
      *
      * @param serviceName service name
+     * @param groupName   group name
      * @param clusters    clusters
      */
-    public void updateServiceNow(String serviceName, String clusters) {
+    public void updateServiceNow(String serviceName, String groupName, String clusters) {
         ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
         try {
-            
+            ServiceInfo result = serverProxy
-            String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);
+                    .queryInstancesOfService(serviceName, groupName, clusters, pushReceiver.getUdpPort(), false);
-            
+            if (null != result) {
-            if (StringUtils.isNotEmpty(result)) {
+                processServiceInfo(result);
-                processServiceJson(result);
             }
         } catch (Exception e) {
             NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
@@ -359,11 +367,12 @@ public class HostReactor implements Closeable {
      * Refresh only.
      *
      * @param serviceName service name
+     * @param groupName   group name
      * @param clusters    cluster
      */
-    public void refreshOnly(String serviceName, String clusters) {
+    public void refreshOnly(String serviceName, String groupName, String clusters) {
         try {
-            serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);
+            serverProxy.queryInstancesOfService(serviceName, groupName, clusters, pushReceiver.getUdpPort(), false);
         } catch (Exception e) {
             NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
         }
@@ -391,13 +400,22 @@ public class HostReactor implements Closeable {
         
         long lastRefTime = Long.MAX_VALUE;
         
-        private final String clusters;
-        
         private final String serviceName;
         
-        public UpdateTask(String serviceName, String clusters) {
+        private final String groupName;
+        
+        private final String clusters;
+        
+        private final String groupedServiceName;
+        
+        private final String serviceKey;
+        
+        public UpdateTask(String serviceName, String groupName, String clusters) {
             this.serviceName = serviceName;
+            this.groupName = groupName;
             this.clusters = clusters;
+            this.groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
+            this.serviceKey = ServiceInfo.getKey(groupedServiceName, clusters);
         }
         
         @Override
@@ -405,36 +423,36 @@ public class HostReactor implements Closeable {
             long delayTime = -1;
             
             try {
-                ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
+                ServiceInfo serviceObj = serviceInfoMap.get(serviceKey);
                 
                 if (serviceObj == null) {
-                    updateServiceNow(serviceName, clusters);
+                    updateServiceNow(serviceName, groupName, clusters);
                     delayTime = DEFAULT_DELAY;
                     return;
                 }
                 
                 if (serviceObj.getLastRefTime() <= lastRefTime) {
-                    updateServiceNow(serviceName, clusters);
+                    updateServiceNow(serviceName, groupName, clusters);
-                    serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
+                    serviceObj = serviceInfoMap.get(serviceKey);
                 } else {
                     // if serviceName already updated by push, we should not override it
                     // since the push data may be different from pull through force push
-                    refreshOnly(serviceName, clusters);
+                    refreshOnly(serviceName, groupName, clusters);
                 }
                 
                 lastRefTime = serviceObj.getLastRefTime();
                 
-                if (!eventDispatcher.isSubscribed(serviceName, clusters) && !futureMap
+                if (!eventDispatcher.isSubscribed(groupedServiceName, clusters) && !futureMap.containsKey(serviceKey)) {
-                        .containsKey(ServiceInfo.getKey(serviceName, clusters))) {
                     // abort the update task
-                    NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);
+                    NAMING_LOGGER
+                            .info("update task is stopped, service:" + groupedServiceName + ", clusters:" + clusters);
                     return;
                 }
                 
                 delayTime = serviceObj.getCacheMillis();
                 
             } catch (Throwable e) {
-                NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);
+                NAMING_LOGGER.warn("[NA] failed to update serviceName: " + groupedServiceName, e);
             } finally {
                 if (delayTime > 0) {
                     executor.schedule(this, delayTime, TimeUnit.MILLISECONDS);

client/src/main/java/com/alibaba/nacos/client/naming/core/PushReceiver.java

@@ -86,7 +86,7 @@ public class PushReceiver implements Runnable, Closeable {
                 PushPacket pushPacket = JacksonUtils.toObj(json, PushPacket.class);
                 String ack;
                 if ("dom".equals(pushPacket.type) || "service".equals(pushPacket.type)) {
-                    hostReactor.processServiceJson(pushPacket.data);
+                    hostReactor.processServiceInfo(pushPacket.data);
                     
                     // send ack to server
                     ack = "{\"type\": \"push-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":"

client/src/main/java/com/alibaba/nacos/client/naming/remote/NamingClientProxy.java

@@ -24,6 +24,8 @@ import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
 import com.alibaba.nacos.api.selector.AbstractSelector;
 import com.alibaba.nacos.common.lifecycle.Closeable;
 
+import java.util.Set;
+
 /**
  * Naming Client Proxy.
  *
@@ -145,4 +147,11 @@ public interface NamingClientProxy extends Closeable {
      * @throws NacosException nacos exception
      */
     void unsubscribe(String serviceName, String clusters) throws NacosException;
+    
+    /**
+     * Update beat info.
+     *
+     * @param modifiedInstances modified instances
+     */
+    void updateBeatInfo(Set<Instance> modifiedInstances);
 }

client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingGrpcClientProxy.java

@@ -37,6 +37,8 @@ import com.alibaba.nacos.client.remote.RpcClient;
 import com.alibaba.nacos.client.remote.RpcClientFactory;
 import com.alibaba.nacos.client.remote.ServerListFactory;
 
+import java.util.Set;
+
 import static com.alibaba.nacos.client.utils.LogUtils.NAMING_LOGGER;
 
 /**
@@ -76,7 +78,6 @@ public class NamingGrpcClientProxy implements NamingClientProxy {
                 instance);
         InstanceRequest request = new InstanceRequest(namespaceId, serviceName, groupName,
                 NamingRemoteConstants.REGISTER_INSTANCE, instance);
-        Response response = rpcClient.request(request);
         requestToServer(request, Response.class);
     }
     
@@ -155,6 +156,10 @@ public class NamingGrpcClientProxy implements NamingClientProxy {
         requestToServer(request, SubscribeServiceResponse.class);
     }
     
+    @Override
+    public void updateBeatInfo(Set<Instance> modifiedInstances) {
+    }
+    
     private <T extends Response> T requestToServer(Request request, Class<T> responseClass) throws NacosException {
         try {
             Response response = rpcClient.request(request);

client/src/main/java/com/alibaba/nacos/client/naming/remote/gprc/NamingPushResponseHandler.java

@@ -37,6 +37,6 @@ public class NamingPushResponseHandler implements ServerPushResponseHandler<Noti
     @Override
     public void responseReply(Response response) {
         NotifySubscriberResponse notifyResponse = (NotifySubscriberResponse) response;
-        hostReactor.processServiceJson(notifyResponse.getServiceInfo());
+        hostReactor.processServiceInfo(notifyResponse.getServiceInfo());
     }
 }

client/src/main/java/com/alibaba/nacos/client/naming/remote/http/NamingHttpClientProxy.java

@@ -32,6 +32,7 @@ import com.alibaba.nacos.api.selector.SelectorType;
 import com.alibaba.nacos.client.config.impl.SpasAdapter;
 import com.alibaba.nacos.client.monitor.MetricsMonitor;
 import com.alibaba.nacos.client.naming.beat.BeatInfo;
+import com.alibaba.nacos.client.naming.beat.BeatReactor;
 import com.alibaba.nacos.client.naming.core.ServerListManager;
 import com.alibaba.nacos.client.naming.remote.NamingClientProxy;
 import com.alibaba.nacos.client.naming.utils.CollectionUtils;
@@ -59,6 +60,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -78,8 +80,6 @@ public class NamingHttpClientProxy implements NamingClientProxy {
     
     private static final int DEFAULT_SERVER_PORT = 8848;
     
-    private int serverPort = DEFAULT_SERVER_PORT;
-    
     private final String namespaceId;
     
     private final SecurityProxy securityProxy;
@@ -88,16 +88,21 @@ public class NamingHttpClientProxy implements NamingClientProxy {
     
     private final ServerListManager serverListManager;
     
-    private Properties properties;
+    private final BeatReactor beatReactor;
+    
+    private int serverPort = DEFAULT_SERVER_PORT;
     
     private ScheduledExecutorService executorService;
     
+    private Properties properties;
+    
     public NamingHttpClientProxy(String namespaceId, ServerListManager serverListManager, Properties properties) {
         this.serverListManager = serverListManager;
         this.securityProxy = new SecurityProxy(properties, nacosRestTemplate);
         this.properties = properties;
         this.setServerPort(DEFAULT_SERVER_PORT);
         this.namespaceId = namespaceId;
+        this.beatReactor = new BeatReactor(this, properties);
         this.initRefreshTask();
     }
     
@@ -128,7 +133,11 @@ public class NamingHttpClientProxy implements NamingClientProxy {
         
         NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", namespaceId, serviceName,
                 instance);
-        
+        String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
+        if (instance.isEphemeral()) {
+            BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
+            beatReactor.addBeatInfo(groupedServiceName, beatInfo);
+        }
         final Map<String, String> params = new HashMap<String, String>(16);
         params.put(CommonParams.NAMESPACE_ID, namespaceId);
         params.put(CommonParams.SERVICE_NAME, serviceName);
@@ -151,7 +160,10 @@ public class NamingHttpClientProxy implements NamingClientProxy {
         NAMING_LOGGER
                 .info("[DEREGISTER-SERVICE] {} deregistering service {} with instance: {}", namespaceId, serviceName,
                         instance);
-    
+        if (instance.isEphemeral()) {
+            beatReactor.removeBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), instance.getIp(),
+                    instance.getPort());
+        }
         final Map<String, String> params = new HashMap<String, String>(8);
         params.put(CommonParams.NAMESPACE_ID, namespaceId);
         params.put(CommonParams.SERVICE_NAME, NamingUtils.getGroupedName(serviceName, groupName));
@@ -188,7 +200,7 @@ public class NamingHttpClientProxy implements NamingClientProxy {
             throws NacosException {
         final Map<String, String> params = new HashMap<String, String>(8);
         params.put(CommonParams.NAMESPACE_ID, namespaceId);
-        params.put(CommonParams.SERVICE_NAME, serviceName);
+        params.put(CommonParams.SERVICE_NAME, NamingUtils.getGroupedName(serviceName, groupName));
         params.put("clusters", clusters);
         params.put("udpPort", String.valueOf(udpPort));
         params.put("clientIP", NetUtils.localIP());
@@ -259,30 +271,6 @@ public class NamingHttpClientProxy implements NamingClientProxy {
         reqApi(UtilAndComs.nacosUrlService, params, HttpMethod.PUT);
     }
     
-    /**
-     * Query instance list.
-     *
-     * @param serviceName service name
-     * @param clusters    clusters
-     * @param udpPort     udp port
-     * @param healthyOnly healthy only
-     * @return instance list
-     * @throws NacosException nacos exception
-     */
-    public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly)
-            throws NacosException {
-        
-        final Map<String, String> params = new HashMap<String, String>(8);
-        params.put(CommonParams.NAMESPACE_ID, namespaceId);
-        params.put(CommonParams.SERVICE_NAME, serviceName);
-        params.put("clusters", clusters);
-        params.put("udpPort", String.valueOf(udpPort));
-        params.put("clientIP", NetUtils.localIP());
-        params.put("healthyOnly", String.valueOf(healthyOnly));
-        
-        return reqApi(UtilAndComs.nacosUrlBase + "/instance/list", params, HttpMethod.GET);
-    }
-    
     /**
      * Send beat.
      *
@@ -376,6 +364,17 @@ public class NamingHttpClientProxy implements NamingClientProxy {
     
     }
     
+    @Override
+    public void updateBeatInfo(Set<Instance> modifiedInstances) {
+        for (Instance instance : modifiedInstances) {
+            String key = beatReactor.buildKey(instance.getServiceName(), instance.getIp(), instance.getPort());
+            if (beatReactor.dom2Beat.containsKey(key) && instance.isEphemeral()) {
+                BeatInfo beatInfo = beatReactor.buildBeatInfo(instance);
+                beatReactor.addBeatInfo(instance.getServiceName(), beatInfo);
+            }
+        }
+    }
+    
     public String reqApi(String api, Map<String, String> params, String method) throws NacosException {
         return reqApi(api, params, Collections.EMPTY_MAP, method);
     }
@@ -579,11 +578,16 @@ public class NamingHttpClientProxy implements NamingClientProxy {
         }
     }
     
+    public BeatReactor getBeatReactor() {
+        return this.beatReactor;
+    }
+    
     @Override
     public void shutdown() throws NacosException {
         String className = this.getClass().getName();
         NAMING_LOGGER.info("{} do shutdown begin", className);
         ThreadUtils.shutdownThreadPool(executorService, NAMING_LOGGER);
+        beatReactor.shutdown();
         NamingHttpClientManager.getInstance().shutdown();
         NAMING_LOGGER.info("{} do shutdown stop", className);
     }

client/src/test/java/com/alibaba/nacos/client/naming/core/HostReactorTest.java

@@ -20,8 +20,8 @@ import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.pojo.Instance;
 import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
 import com.alibaba.nacos.client.naming.beat.BeatInfo;
-import com.alibaba.nacos.client.naming.beat.BeatReactor;
 import com.alibaba.nacos.client.naming.remote.http.NamingHttpClientProxy;
+import com.alibaba.nacos.common.utils.JacksonUtils;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -39,8 +39,6 @@ import static org.mockito.Mockito.when;
 @RunWith(MockitoJUnitRunner.class)
 public class HostReactorTest {
     
-    private static final String CACHE_DIR = HostReactorTest.class.getResource("/").getPath() + "cache/";
-    
     @Mock
     private NamingHttpClientProxy namingHttpClientProxy;
     
@@ -49,11 +47,8 @@ public class HostReactorTest {
     
     private HostReactor hostReactor;
     
-    private BeatReactor beatReactor;
-    
     @Before
     public void setUp() throws Exception {
-        beatReactor = new BeatReactor(namingHttpClientProxy);
         BeatInfo beatInfo = new BeatInfo();
         beatInfo.setServiceName("testName");
         beatInfo.setIp("1.1.1.1");
@@ -63,23 +58,22 @@ public class HostReactorTest {
         beatInfo.setMetadata(new HashMap<String, String>());
         beatInfo.setScheduled(false);
         beatInfo.setPeriod(1000L);
-        beatReactor.addBeatInfo("testName", beatInfo);
+        hostReactor = new HostReactor(eventDispatcher, namingHttpClientProxy, "public", null);
-        hostReactor = new HostReactor(eventDispatcher, namingHttpClientProxy, beatReactor, CACHE_DIR);
     }
     
     @Test
     public void testProcessServiceJson() {
-        ServiceInfo actual = hostReactor.processServiceJson(EXAMPLE);
+        ServiceInfo actual = hostReactor.processServiceInfo(EXAMPLE);
         assertServiceInfo(actual);
-        hostReactor.processServiceJson(CHANGE_DATA_EXAMPLE);
+        ServiceInfo actual2 = hostReactor.processServiceInfo(CHANGE_DATA_EXAMPLE);
-        BeatInfo actualBeatInfo = beatReactor.dom2Beat.get(beatReactor.buildKey("testName", "1.1.1.1", 1234));
+        assertEquals(2.0, actual2.getHosts().get(0).getWeight(), 0.0);
-        assertEquals(2.0, actualBeatInfo.getWeight(), 0.0);
     }
     
     @Test
     public void testGetServiceInfoDirectlyFromServer() throws NacosException {
-        when(namingHttpClientProxy.queryList("testName", "testClusters", 0, false)).thenReturn(EXAMPLE);
+        when(namingHttpClientProxy.queryInstancesOfService("testName", "", "testClusters", 0, false))
-        ServiceInfo actual = hostReactor.getServiceInfoDirectlyFromServer("testName", "testClusters");
+                .thenReturn(JacksonUtils.toObj(EXAMPLE, ServiceInfo.class));
+        ServiceInfo actual = hostReactor.getServiceInfoDirectlyFromServer("testName", "", "testClusters");
         assertServiceInfo(actual);
     }