负载均衡bugfix；监听并发问题bugfix；direct memory泄漏bugfix (#4365)

* memory gc optimize ; string pool bugfix

* grpc executor

* grpc executor

* 负载均衡bugfix；监听并发问题bugfix；direct memory泄漏bugfix

* check style fix

api/src/main/java/com/alibaba/nacos/api/remote/DefaultRequestFuture.java

@@ -81,8 +81,10 @@ public class DefaultRequestFuture implements RequestFuture {
         this.requestCallBack = requestCallBack;
         this.requestId = requestId;
         this.connectionId = connectionId;
-        this.timeoutFuture = RpcScheduledExecutor.TIMEOUT_SHEDULER
+        if (requestCallBack != null) {
-                .schedule(new TimeoutHandler(), requestCallBack.getTimeout(), TimeUnit.MILLISECONDS);
+            this.timeoutFuture = RpcScheduledExecutor.TIMEOUT_SHEDULER
+                    .schedule(new TimeoutHandler(), requestCallBack.getTimeout(), TimeUnit.MILLISECONDS);
+        }
         this.timeoutInnerTrigger = timeoutInnerTrigger;
     }
     

api/src/main/java/com/alibaba/nacos/api/remote/Requester.java

@@ -83,4 +83,11 @@ public interface Requester {
      * close connection.
      */
     public void close();
+    
+    /**
+     * check this requester is busy.
+     *
+     * @return
+     */
+    public boolean isBusy();
 }

common/src/main/java/com/alibaba/nacos/common/remote/client/Connection.java

@@ -40,6 +40,11 @@ public abstract class Connection implements Requester {
         this.serverInfo = serverInfo;
     }
     
+    @Override
+    public boolean isBusy() {
+        return false;
+    }
+    
     /**
      * Getter method for property <tt>abandon</tt>.
      *

common/src/main/java/com/alibaba/nacos/common/remote/client/grpc/GrpcConnection.java

@@ -78,8 +78,7 @@ public class GrpcConnection extends Connection {
         try {
             grpcResponse = requestFuture.get(timeouts, TimeUnit.MILLISECONDS);
         } catch (Exception e) {
-            e.printStackTrace();
+            throw new NacosException(NacosException.SERVER_ERROR, e);
-            return null;
         }
         
         Response response = (Response) GrpcUtils.parse(grpcResponse).getBody();
@@ -101,13 +100,7 @@ public class GrpcConnection extends Connection {
             @Override
             public Response get() throws InterruptedException, ExecutionException {
                 Payload grpcResponse = null;
-                try {
+                grpcResponse = requestFuture.get();
-                    grpcResponse = requestFuture.get();
-                } catch (Exception e) {
-                    e.printStackTrace();
-                    return null;
-                }
-                
                 Response response = (Response) GrpcUtils.parse(grpcResponse).getBody();
                 return response;
             }
@@ -177,6 +170,10 @@ public class GrpcConnection extends Connection {
     
     @Override
     public void close() {
+        if (this.payloadStreamObserver != null) {
+            payloadStreamObserver.onCompleted();
+        }
+        
         if (this.channel != null && !channel.isShutdown()) {
             this.channel.shutdownNow();
         }

common/src/main/java/com/alibaba/nacos/common/remote/exception/ConnectionAlreadyClosedException.java

@@ -26,6 +26,10 @@ public class ConnectionAlreadyClosedException extends RemoteException {
     
     private static final int CONNECTION_ALREADY_CLOSED = 600;
     
+    public ConnectionAlreadyClosedException(String msg) {
+        super(CONNECTION_ALREADY_CLOSED);
+    }
+    
     public ConnectionAlreadyClosedException() {
         super(CONNECTION_ALREADY_CLOSED);
     }

common/src/main/java/com/alibaba/nacos/common/remote/exception/ConnectionBusyException.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 1999-2020 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.nacos.common.remote.exception;
+
+/**
+ * connection is busy exception.
+ *
+ * @author liuzunfei
+ * @version $Id: ConnectionBusyException.java, v 0.1 2020年11月30日 7:28 PM liuzunfei Exp $
+ */
+public class ConnectionBusyException extends RemoteException {
+    
+    private static final int CONNECTION_BUSY = 601;
+    
+    public ConnectionBusyException(String msg) {
+        super(CONNECTION_BUSY, msg);
+    }
+    
+    public ConnectionBusyException(Throwable throwable) {
+        super(CONNECTION_BUSY, throwable);
+    }
+}

common/src/main/java/com/alibaba/nacos/common/remote/exception/RemoteException.java

@@ -30,6 +30,10 @@ public class RemoteException extends NacosRuntimeException {
         super(errorCode);
     }
     
+    public RemoteException(int errorCode, String msg) {
+        super(errorCode, msg);
+    }
+    
     public RemoteException(int errorCode, Throwable throwable) {
         super(errorCode, throwable);
     }

config/src/main/java/com/alibaba/nacos/config/server/remote/ConfigChangeListenContext.java

@@ -17,11 +17,12 @@
 package com.alibaba.nacos.config.server.remote;
 
 import com.alibaba.nacos.common.utils.CollectionUtils;
-import com.alibaba.nacos.common.utils.MapUtil;
 import org.springframework.stereotype.Component;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -36,28 +37,27 @@ import java.util.concurrent.ConcurrentHashMap;
 public class ConfigChangeListenContext {
     
     /**
-     * groupKey-> connnection set.
+     * groupKey-> connection set.
      */
-    private final Map<String, Set<String>> groupKeyContext = new ConcurrentHashMap<String, Set<String>>(128);
+    private ConcurrentHashMap<String, HashSet<String>> groupKeyContext = new ConcurrentHashMap<String, HashSet<String>>();
     
     /**
-     * connectionId-> groupkey set.
+     * connectionId-> group key set.
      */
-    private final Map<String, HashMap<String, String>> connectionIdContext = new ConcurrentHashMap<String, HashMap<String, String>>(128);
+    private ConcurrentHashMap<String, HashMap<String, String>> connectionIdContext = new ConcurrentHashMap<String, HashMap<String, String>>();
     
     /**
-     * add listen .
+     * add listen.
      *
-     * @param listenKey    listenKey.
+     * @param groupKey     groupKey.
      * @param connectionId connectionId.
      */
-    public void addListen(String listenKey, String md5, String connectionId) {
+    public synchronized void addListen(String groupKey, String md5, String connectionId) {
-        
         // 1.add groupKeyContext
-        Set<String> listenClients = groupKeyContext.get(listenKey);
+        Set<String> listenClients = groupKeyContext.get(groupKey);
         if (listenClients == null) {
-            groupKeyContext.putIfAbsent(listenKey, new HashSet<String>());
+            groupKeyContext.putIfAbsent(groupKey, new HashSet<String>());
-            listenClients = groupKeyContext.get(listenKey);
+            listenClients = groupKeyContext.get(groupKey);
         }
         listenClients.add(connectionId);
         
@@ -67,44 +67,70 @@ public class ConfigChangeListenContext {
             connectionIdContext.putIfAbsent(connectionId, new HashMap<String, String>(16));
             groupKeys = connectionIdContext.get(connectionId);
         }
-        groupKeys.put(listenKey, md5);
+        groupKeys.put(groupKey, md5);
         
     }
     
     /**
      * remove listen context for connection id .
      *
-     * @param listenKey    listenKey.
+     * @param groupKey     groupKey.
      * @param connectionId connection id.
      */
-    public void removeListen(String listenKey, String connectionId) {
+    public synchronized void removeListen(String groupKey, String connectionId) {
         
         //1. remove groupKeyContext
-        Set<String> connectionIds = groupKeyContext.get(listenKey);
+        Set<String> connectionIds = groupKeyContext.get(groupKey);
         if (connectionIds != null) {
             connectionIds.remove(connectionId);
             if (connectionIds.isEmpty()) {
-                MapUtil.removeKey(groupKeyContext, listenKey, CollectionUtils::isEmpty);
+                groupKeyContext.remove(groupKey);
             }
         }
         
         //2.remove connectionIdContext
         HashMap<String, String> groupKeys = connectionIdContext.get(connectionId);
         if (groupKeys != null) {
-            groupKeys.remove(listenKey);
+            groupKeys.remove(groupKey);
         }
     }
     
-    public Set<String> getListeners(String listenKey) {
+    /**
-        return groupKeyContext.get(listenKey);
+     * get listeners of the group key.
+     *
+     * @param groupKey groupKey.
+     * @return the copy of listeners, may be return null.
+     */
+    public synchronized Set<String> getListeners(String groupKey) {
+        
+        HashSet<String> strings = groupKeyContext.get(groupKey);
+        if (CollectionUtils.isNotEmpty(strings)) {
+            Set<String> listenConnections = new HashSet<String>();
+            safeCopy(strings, listenConnections);
+            return listenConnections;
+        }
+        return null;
     }
     
     /**
-     * remove the context related to the connectionid.
+     * copy collections.
+     *
+     * @param src  may be modified concurrently
+     * @param dest dest collection
+     */
+    private void safeCopy(Collection src, Collection dest) {
+        Iterator iterator = src.iterator();
+        while (iterator.hasNext()) {
+            dest.add(iterator.next());
+        }
+    }
+    
+    /**
+     * remove the context related to the connection id.
      *
      * @param connectionId connectionId.
      */
-    public void clearContextForConnectionId(final String connectionId) {
+    public synchronized void clearContextForConnectionId(final String connectionId) {
         
         Map<String, String> listenKeys = getListenKeys(connectionId);
         
@@ -115,32 +141,35 @@ public class ConfigChangeListenContext {
                 if (CollectionUtils.isNotEmpty(connectionIds)) {
                     connectionIds.remove(connectionId);
                 } else {
-                    MapUtil.removeKey(groupKeyContext, groupKey.getKey(), CollectionUtils::isEmpty);
+                    groupKeyContext.remove(groupKey.getKey());
                 }
                 
             }
         }
-        MapUtil.removeKey(connectionIdContext, connectionId, MapUtil::isEmpty);
+        
+        connectionIdContext.remove(connectionId);
     }
     
     /**
-     * get listenkeys.
+     * get listen keys.
      *
-     * @param connectionId connetionid.
+     * @param connectionId connection id.
-     * @return
+     * @return listen group keys of the connection id, key:group key,value:md5
      */
-    public Map<String, String> getListenKeys(String connectionId) {
+    public synchronized Map<String, String> getListenKeys(String connectionId) {
-        return connectionIdContext.get(connectionId);
+        Map<String, String> copy = new HashMap<String, String>(connectionIdContext.get(connectionId));
+        return copy;
     }
     
     /**
-     * get listenkey.
+     * get md5.
      *
-     * @param connectionId connetionid.
+     * @param connectionId connection id.
-     * @return
+     * @return md5 of the listen group key.
      */
     public String getListenKeyMd5(String connectionId, String groupKey) {
         Map<String, String> groupKeyContexts = connectionIdContext.get(connectionId);
         return groupKeyContexts == null ? null : groupKeyContexts.get(groupKey);
     }
+    
 }

config/src/main/java/com/alibaba/nacos/config/server/remote/RpcConfigChangeNotifier.java

@@ -21,6 +21,7 @@ import com.alibaba.nacos.api.remote.response.AbstractPushCallBack;
 import com.alibaba.nacos.common.notify.Event;
 import com.alibaba.nacos.common.notify.NotifyCenter;
 import com.alibaba.nacos.common.notify.listener.Subscriber;
+import com.alibaba.nacos.common.remote.exception.ConnectionAlreadyClosedException;
 import com.alibaba.nacos.common.utils.CollectionUtils;
 import com.alibaba.nacos.config.server.model.event.LocalDataChangeEvent;
 import com.alibaba.nacos.config.server.utils.ConfigExecutor;
@@ -30,9 +31,9 @@ import com.alibaba.nacos.core.remote.Connection;
 import com.alibaba.nacos.core.remote.ConnectionManager;
 import com.alibaba.nacos.core.remote.RpcPushService;
 import com.alibaba.nacos.core.utils.Loggers;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
@@ -46,53 +47,50 @@ import java.util.concurrent.TimeUnit;
 @Component(value = "rpcConfigChangeNotifier")
 public class RpcConfigChangeNotifier extends Subscriber<LocalDataChangeEvent> {
     
-    final ConfigChangeListenContext configChangeListenContext;
+    public RpcConfigChangeNotifier() {
-    
-    private final RpcPushService rpcPushService;
-    
-    private final ConnectionManager connectionManager;
-    
-    public RpcConfigChangeNotifier(ConfigChangeListenContext configChangeListenContext, RpcPushService rpcPushService,
-            ConnectionManager connectionManager) {
         NotifyCenter.registerSubscriber(this);
-        this.configChangeListenContext = configChangeListenContext;
-        this.rpcPushService = rpcPushService;
-        this.connectionManager = connectionManager;
     }
     
+    @Autowired
+    ConfigChangeListenContext configChangeListenContext;
+    
+    @Autowired
+    private RpcPushService rpcPushService;
+    
+    @Autowired
+    private ConnectionManager connectionManager;
+    
     /**
      * adaptor to config module ,when server side config change ,invoke this method.
      *
      * @param groupKey     groupKey
-     * @param notifyRequest notifyRequest
+     * @param notifyRequet notifyRequet
      */
-    public void configDataChanged(String groupKey, final ConfigChangeNotifyRequest notifyRequest) {
+    public void configDataChanged(String groupKey, final ConfigChangeNotifyRequest notifyRequet) {
         
         Set<String> listeners = configChangeListenContext.getListeners(groupKey);
-        if (listeners == null || listeners.isEmpty()) {
+        if (CollectionUtils.isEmpty(listeners)) {
             return;
         }
-        Set<String> clients = new HashSet<>(listeners);
+        
         int notifyCount = 0;
-        if (!CollectionUtils.isEmpty(clients)) {
+        for (final String client : listeners) {
-            for (final String client : clients) {
+            Connection connection = connectionManager.getConnection(client);
-                Connection connection = connectionManager.getConnection(client);
+            if (connection == null) {
-                if (connection == null) {
+                continue;
+            }
+            
+            if (notifyRequet.isBeta()) {
+                List<String> betaIps = notifyRequet.getBetaIps();
+                if (betaIps != null && !betaIps.contains(connection.getMetaInfo().getClientIp())) {
                     continue;
                 }
-    
-                if (notifyRequest.isBeta()) {
-                    List<String> betaIps = notifyRequest.getBetaIps();
-                    if (betaIps != null && !betaIps.contains(connection.getMetaInfo().getClientIp())) {
-                        continue;
-                    }
-                }
-    
-                RpcPushTask rpcPushRetryTask = new RpcPushTask(notifyRequest, 50, client,
-                        connection.getMetaInfo().getClientIp(), connection.getMetaInfo().getConnectionId());
-                push(rpcPushRetryTask);
-                notifyCount++;
             }
+            
+            RpcPushTask rpcPushRetryTask = new RpcPushTask(notifyRequet, 50, client,
+                    connection.getMetaInfo().getClientIp(), connection.getMetaInfo().getConnectionId());
+            push(rpcPushRetryTask);
+            notifyCount++;
         }
         
         Loggers.REMOTE_PUSH.info("push [{}] clients ,groupKey=[{}]", notifyCount, groupKey);
@@ -104,10 +102,10 @@ public class RpcConfigChangeNotifier extends Subscriber<LocalDataChangeEvent> {
         boolean isBeta = event.isBeta;
         List<String> betaIps = event.betaIps;
         String[] strings = GroupKey.parseKey(groupKey);
-        String dataId = strings[0];
+        String dataid = strings[0];
         String group = strings[1];
         String tenant = strings.length > 2 ? strings[2] : "";
-        ConfigChangeNotifyRequest notifyRequest = ConfigChangeNotifyRequest.build(dataId, group, tenant);
+        ConfigChangeNotifyRequest notifyRequest = ConfigChangeNotifyRequest.build(dataid, group, tenant);
         notifyRequest.setBeta(isBeta);
         notifyRequest.setBetaIps(betaIps);
         if (PropertyUtil.isPushContent()) {
@@ -128,7 +126,7 @@ public class RpcConfigChangeNotifier extends Subscriber<LocalDataChangeEvent> {
     
     class RpcPushTask implements Runnable {
         
-        ConfigChangeNotifyRequest notifyRequest;
+        ConfigChangeNotifyRequest notifyRequet;
         
         int maxRetryTimes = -1;
         
@@ -144,9 +142,9 @@ public class RpcConfigChangeNotifier extends Subscriber<LocalDataChangeEvent> {
             this(notifyRequet, -1, clientId, clientIp, appName);
         }
         
-        public RpcPushTask(ConfigChangeNotifyRequest notifyRequest, int maxRetryTimes, String clientId, String clientIp,
+        public RpcPushTask(ConfigChangeNotifyRequest notifyRequet, int maxRetryTimes, String clientId, String clientIp,
                 String appName) {
-            this.notifyRequest = notifyRequest;
+            this.notifyRequet = notifyRequet;
             this.maxRetryTimes = maxRetryTimes;
             this.clientId = clientId;
             this.clientIp = clientIp;
@@ -159,7 +157,7 @@ public class RpcConfigChangeNotifier extends Subscriber<LocalDataChangeEvent> {
         
         @Override
         public void run() {
-            rpcPushService.pushWithCallback(clientId, notifyRequest, new AbstractPushCallBack(3000L) {
+            rpcPushService.pushWithCallback(clientId, notifyRequet, new AbstractPushCallBack(3000L) {
                 int retryTimes = tryTimes;
                 
                 @Override
@@ -169,6 +167,9 @@ public class RpcConfigChangeNotifier extends Subscriber<LocalDataChangeEvent> {
                 
                 @Override
                 public void onFail(Throwable e) {
+                    if (e instanceof ConnectionAlreadyClosedException) {
+                        Loggers.CORE.warn(e.getMessage());
+                    }
                     push(RpcPushTask.this);
                 }
                 
@@ -179,7 +180,7 @@ public class RpcConfigChangeNotifier extends Subscriber<LocalDataChangeEvent> {
     }
     
     private void push(RpcPushTask retryTask) {
-        ConfigChangeNotifyRequest notifyRequet = retryTask.notifyRequest;
+        ConfigChangeNotifyRequest notifyRequet = retryTask.notifyRequet;
         if (retryTask.isOverTimes()) {
             Loggers.CORE
                     .warn("push callback retry fail over times .dataId={},group={},tenant={},clientId={},will unregister client.",

config/src/main/java/com/alibaba/nacos/config/server/service/LongPollingService.java

@@ -399,7 +399,7 @@ public class LongPollingService {
                     try {
                         getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
                         
-                        // Delete subsciber's relations.
+                        // Delete subscriber's relations.
                         allSubs.remove(ClientLongPolling.this);
                         
                         if (isFixedPolling()) {

console/src/main/java/com/alibaba/nacos/console/controller/ServerLoaderController.java

@@ -35,6 +35,7 @@ import com.alibaba.nacos.core.cluster.remote.ClusterRpcClientProxy;
 import com.alibaba.nacos.core.remote.Connection;
 import com.alibaba.nacos.core.remote.ConnectionManager;
 import com.alibaba.nacos.core.remote.core.ServerLoaderInfoRequestHandler;
+import com.alibaba.nacos.core.remote.core.ServerReloaderRequestHandler;
 import com.alibaba.nacos.core.utils.RemoteUtils;
 import org.apache.commons.lang3.math.NumberUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -77,6 +78,9 @@ public class ServerLoaderController {
     @Autowired
     private ClusterRpcClientProxy clusterRpcClientProxy;
     
+    @Autowired
+    private ServerReloaderRequestHandler serverReloaderRequestHandler;
+    
     @Autowired
     private ServerLoaderInfoRequestHandler serverLoaderInfoRequestHandler;
     
@@ -184,15 +188,9 @@ public class ServerLoaderController {
             }
         }
         
-        List<ServerLoaderMetris> responseList = new LinkedList<ServerLoaderMetris>();
-        
         try {
-            ServerLoaderInfoResponse handle = serverLoaderInfoRequestHandler
+            serverReloaderRequestHandler.handle(serverLoaderInfoRequest, new RequestMeta());
-                    .handle(new ServerLoaderInfoRequest(), new RequestMeta());
+            
-            ServerLoaderMetris metris = new ServerLoaderMetris();
-            metris.setAddress(serverMemberManager.getSelf().getAddress());
-            metris.setMetric(handle.getLoaderMetrics());
-            responseList.add(metris);
         } catch (NacosException e) {
             e.printStackTrace();
         }

core/src/main/java/com/alibaba/nacos/core/remote/Connection.java

@@ -57,6 +57,11 @@ public abstract class Connection implements Requester {
         return metaInfo;
     }
     
+    @Override
+    public boolean isBusy() {
+        return false;
+    }
+    
     @Override
     public String toString() {
         return ToStringBuilder.reflectionToString(this);

core/src/main/java/com/alibaba/nacos/core/remote/ConnectionManager.java

@@ -80,14 +80,17 @@ public class ConnectionManager {
      * @param connectionId connectionId
      * @param connection   connection
      */
-    public void register(String connectionId, Connection connection) {
+    public synchronized void register(String connectionId, Connection connection) {
-        Connection connectionInner = connetions.put(connectionId, connection);
+        if (connection.isConnected()) {
-        if (connectionInner == null) {
+            Connection connectionInner = connetions.put(connectionId, connection);
-            clientConnectionEventListenerRegistry.notifyClientConnected(connection);
+            if (connectionInner == null) {
-            Loggers.REMOTE
+                clientConnectionEventListenerRegistry.notifyClientConnected(connection);
-                    .info("new connection registered successfully, connectionid = {},connection={} ", connectionId,
+                Loggers.REMOTE
-                            connection);
+                        .info("new connection registered successfully, connectionid = {},connection={} ", connectionId,
+                                connection);
+            }
         }
+        
     }
     
     /**
@@ -95,7 +98,7 @@ public class ConnectionManager {
      *
      * @param connectionId connectionId.
      */
-    public void unregister(String connectionId) {
+    public synchronized void unregister(String connectionId) {
         Connection remove = this.connetions.remove(connectionId);
         if (remove != null) {
             remove.close();
@@ -165,9 +168,6 @@ public class ConnectionManager {
             public void run() {
                 try {
                     
-                    Loggers.REMOTE.info("rpc ack size :{}", RpcAckCallbackSynchronizer.CALLBACK_CONTEXT.size());
-                    ;
-    
                     MetricsMonitor.getLongConnectionMonitor().set(connetions.size());
                     
                     long currentStamp = System.currentTimeMillis();
@@ -184,18 +184,18 @@ public class ConnectionManager {
                         }
                     }
                     
+                    ConnectResetRequest connectResetRequest = new ConnectResetRequest();
+                    if (StringUtils.isNotBlank(redirectAddress) && redirectAddress.contains(Constants.COLON)) {
+                        String[] split = redirectAddress.split(Constants.COLON);
+                        connectResetRequest.setServerIp(split[0]);
+                        connectResetRequest.setServerPort(split[1]);
+                    }
+                    
                     for (String expeledClientId : expelClient) {
                         try {
                             Connection connection = getConnection(expeledClientId);
                             if (connection != null) {
-    
+                                connection.asyncRequest(connectResetRequest, buildMeta(), null);
-                                ConnectResetRequest connectResetRequest = new ConnectResetRequest();
-                                if (StringUtils.isNotBlank(redirectAddress) && redirectAddress.contains(":")) {
-                                    String[] split = redirectAddress.split(Constants.COLON);
-                                    connectResetRequest.setServerIp(split[0]);
-                                    connectResetRequest.setServerPort(split[1]);
-                                }
-                                connection.request(connectResetRequest, buildMeta());
                                 Loggers.REMOTE
                                         .info("expel connection ,send switch server response connectionid = {},connectResetRequest={} ",
                                                 expeledClientId, connectResetRequest);
@@ -214,7 +214,7 @@ public class ConnectionManager {
                         redirectAddress = null;
                     }
                     
-                } catch (Exception e) {
+                } catch (Throwable e) {
                     Loggers.REMOTE.error("error occurs when heathy check... ", e);
                 }
             }

core/src/main/java/com/alibaba/nacos/core/remote/core/ServerReloaderRequestHandler.java

@@ -52,9 +52,7 @@ public class ServerReloaderRequestHandler extends RequestHandler<ServerReloadReq
         if (sdkCount <= reloadCount) {
             response.setMessage("ignore");
         } else {
-            if (reloadCount * (1 + RemoteUtils.LOADER_FACTOR) < sdkCount) {
+            reloadCount = (int) Math.max(reloadCount, sdkCount * (1 - RemoteUtils.LOADER_FACTOR));
-                reloadCount = (int) (sdkCount * (1 - RemoteUtils.LOADER_FACTOR));
-            }
             connectionManager.loadCount(reloadCount, null);
             response.setMessage("ok");
         }

core/src/main/java/com/alibaba/nacos/core/remote/grpc/BaseGrpcServer.java

@@ -17,6 +17,7 @@
 package com.alibaba.nacos.core.remote.grpc;
 
 import com.alibaba.nacos.api.grpc.auto.Payload;
+import com.alibaba.nacos.common.executor.ExecutorFactory;
 import com.alibaba.nacos.common.remote.ConnectionType;
 import com.alibaba.nacos.common.utils.ReflectUtils;
 import com.alibaba.nacos.common.utils.UuidUtils;
@@ -24,6 +25,7 @@ import com.alibaba.nacos.core.remote.BaseRpcServer;
 import com.alibaba.nacos.core.remote.ConnectionManager;
 import com.alibaba.nacos.core.remote.RequestHandlerRegistry;
 import com.alibaba.nacos.core.utils.Loggers;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.grpc.Attributes;
 import io.grpc.Context;
 import io.grpc.Contexts;
@@ -31,6 +33,7 @@ import io.grpc.Grpc;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.Server;
+import io.grpc.ServerBuilder;
 import io.grpc.ServerCall;
 import io.grpc.ServerCallHandler;
 import io.grpc.ServerInterceptor;
@@ -38,7 +41,6 @@ import io.grpc.ServerInterceptors;
 import io.grpc.ServerServiceDefinition;
 import io.grpc.ServerTransportFilter;
 import io.grpc.internal.ServerStream;
-import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
 import io.grpc.netty.shaded.io.netty.channel.Channel;
 import io.grpc.protobuf.ProtoUtils;
 import io.grpc.stub.ServerCalls;
@@ -46,6 +48,7 @@ import io.grpc.util.MutableHandlerRegistry;
 import org.springframework.beans.factory.annotation.Autowired;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.Executor;
 
 /**
  * Grpc implementation as  a rpc server.
@@ -55,6 +58,8 @@ import java.net.InetSocketAddress;
  */
 public abstract class BaseGrpcServer extends BaseRpcServer {
     
+    private static Executor grpcExecutor;
+    
     private Server server;
     
     private static final String REQUEST_BI_STREAM_SERVICE_NAME = "BiRequestStream";
@@ -106,8 +111,12 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
     
         addServices(handlerRegistry, serverInterceptor);
     
-        server = NettyServerBuilder.forPort(getServicePort()).fallbackHandlerRegistry(handlerRegistry)
+        grpcExecutor = ExecutorFactory.Managed
-                .addTransportFilter(new ServerTransportFilter() {
+                .newCustomerThreadExecutor("core", Runtime.getRuntime().availableProcessors(),
+                        Runtime.getRuntime().availableProcessors() * 4, 10000,
+                        new ThreadFactoryBuilder().setDaemon(true).setNameFormat("nacos-grpc-executor-%d").build());
+        server = ServerBuilder.forPort(getServicePort()).executor(grpcExecutor)
+                .fallbackHandlerRegistry(handlerRegistry).addTransportFilter(new ServerTransportFilter() {
                     @Override
                     public Attributes transportReady(Attributes transportAttrs) {
                         InetSocketAddress remoteAddress = (InetSocketAddress) transportAttrs
@@ -120,17 +129,17 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
                         Attributes attrWraper = transportAttrs.toBuilder()
                                 .set(TRANS_KEY_CONN_ID, UuidUtils.generateUuid()).set(TRANS_KEY_CLIENT_IP, remoteIp)
                                 .set(TRANS_KEY_CLIENT_PORT, remotePort).set(TRANS_KEY_LOCAL_PORT, localPort).build();
-                        String connectionid = attrWraper.get(TRANS_KEY_CONN_ID);
+                        String connectionId = attrWraper.get(TRANS_KEY_CONN_ID);
-                        Loggers.REMOTE.info(" connection transportReady,connectionid = {} ", connectionid);
+                        Loggers.REMOTE.info(" connection transportReady,connectionId = {} ", connectionId);
                         return attrWraper;
     
                     }
                 
                     @Override
                     public void transportTerminated(Attributes transportAttrs) {
-                        String connectionid = transportAttrs.get(TRANS_KEY_CONN_ID);
+                        String connectionId = transportAttrs.get(TRANS_KEY_CONN_ID);
-                        Loggers.REMOTE.info(" connection transportTerminated,connectionid = {} ", connectionid);
+                        Loggers.REMOTE.info(" connection transportTerminated,connectionId = {} ", connectionId);
-                        connectionManager.unregister(connectionid);
+                        connectionManager.unregister(connectionId);
                     }
                 }).build();
         server.start();

core/src/main/java/com/alibaba/nacos/core/remote/grpc/GrpcBiStreamRequestAcceptor.java

@@ -30,8 +30,8 @@ import com.alibaba.nacos.core.remote.Connection;
 import com.alibaba.nacos.core.remote.ConnectionManager;
 import com.alibaba.nacos.core.remote.ConnectionMetaInfo;
 import com.alibaba.nacos.core.remote.RpcAckCallbackSynchronizer;
-import com.alibaba.nacos.core.utils.Loggers;
 import com.alibaba.nacos.sys.utils.ApplicationUtils;
+import io.grpc.stub.ServerCallStreamObserver;
 import io.grpc.stub.StreamObserver;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -90,10 +90,8 @@ public class GrpcBiStreamRequestAcceptor extends BiRequestStreamGrpc.BiRequestSt
                         return;
                     }
                     
-                    //if connnection is already terminated,not register it.
+                    //if connection is already terminated,not register it.
-                    if (connection.isConnected()) {
+                    connectionManager.register(connectionId, connection);
-                        connectionManager.register(connectionId, connection);
-                    }
                     
                 } else if (plainRequest.getBody() instanceof Response) {
                     Response response = (Response) plainRequest.getBody();
@@ -104,8 +102,16 @@ public class GrpcBiStreamRequestAcceptor extends BiRequestStreamGrpc.BiRequestSt
             
             @Override
             public void onError(Throwable t) {
-                Loggers.REMOTE.error("grpc streamObserver error .", t);
+                if (responseObserver instanceof ServerCallStreamObserver) {
-                responseObserver.onCompleted();
+                    ServerCallStreamObserver serverCallStreamObserver = ((ServerCallStreamObserver) responseObserver);
+                    if (serverCallStreamObserver.isCancelled()) {
+                        //client close the stream.
+                        return;
+                    } else {
+                        serverCallStreamObserver.onCompleted();
+                    }
+                }
+                
             }
             
             @Override

core/src/main/java/com/alibaba/nacos/core/remote/grpc/GrpcConnection.java

@@ -27,6 +27,7 @@ import com.alibaba.nacos.api.remote.response.Response;
 import com.alibaba.nacos.api.utils.NetUtils;
 import com.alibaba.nacos.common.remote.client.grpc.GrpcUtils;
 import com.alibaba.nacos.common.remote.exception.ConnectionAlreadyClosedException;
+import com.alibaba.nacos.common.remote.exception.ConnectionBusyException;
 import com.alibaba.nacos.common.utils.VersionUtils;
 import com.alibaba.nacos.core.remote.Connection;
 import com.alibaba.nacos.core.remote.ConnectionMetaInfo;
@@ -34,6 +35,7 @@ import com.alibaba.nacos.core.remote.RpcAckCallbackSynchronizer;
 import com.alibaba.nacos.core.utils.Loggers;
 import io.grpc.StatusRuntimeException;
 import io.grpc.netty.shaded.io.netty.channel.Channel;
+import io.grpc.stub.ServerCallStreamObserver;
 import io.grpc.stub.StreamObserver;
 
 import java.util.Map;
@@ -58,7 +60,13 @@ public class GrpcConnection extends Connection {
     
     private void sendRequestNoAck(Request request, RequestMeta meta) throws NacosException {
         try {
-            streamObserver.onNext(GrpcUtils.convert(request, wrapMeta(meta)));
+            //StreamObserver#onNext() is not thread-safe,synchronized is required to avoid direct memory leak.
+            synchronized (streamObserver) {
+                if (this.isBusy()) {
+                    throw new ConnectionBusyException(this.getMetaInfo().getConnectionId() + ",connection busy.");
+                }
+                streamObserver.onNext(GrpcUtils.convert(request, wrapMeta(meta)));
+            }
         } catch (Exception e) {
             if (e instanceof StatusRuntimeException) {
                 throw new ConnectionAlreadyClosedException(e);
@@ -89,14 +97,15 @@ public class GrpcConnection extends Connection {
         String requestId = String.valueOf(PushAckIdGenerator.getNextId());
         request.setRequestId(requestId);
         sendRequestNoAck(request, meta);
+        
         DefaultRequestFuture defaultPushFuture = new DefaultRequestFuture(getMetaInfo().getConnectionId(), requestId,
-                callBack,
+                callBack, new DefaultRequestFuture.TimeoutInnerTrigger() {
-                new DefaultRequestFuture.TimeoutInnerTrigger() {
+            @Override
-                    @Override
+            public void triggerOnTimeout() {
-                    public void triggerOnTimeout() {
+                RpcAckCallbackSynchronizer.clearFuture(getMetaInfo().getConnectionId(), requestId);
-                        RpcAckCallbackSynchronizer.clearFuture(getMetaInfo().getConnectionId(), requestId);
+            }
-                    }
+        });
-                });
+        
         RpcAckCallbackSynchronizer.syncCallback(getMetaInfo().getConnectionId(), requestId, defaultPushFuture);
         return defaultPushFuture;
     }
@@ -137,14 +146,35 @@ public class GrpcConnection extends Connection {
     @Override
     public void close() {
         try {
-            streamObserver.onCompleted();
+            if (isConnected()) {
+                closeBiStream();
+                channel.close();
+            }
+            
         } catch (Exception e) {
             Loggers.REMOTE.debug(String.format("[%s] connection close exception  : %s", "grpc", e.getMessage()));
         }
     }
     
+    private void closeBiStream() {
+        if (streamObserver instanceof ServerCallStreamObserver) {
+            ServerCallStreamObserver serverCallStreamObserver = ((ServerCallStreamObserver) streamObserver);
+            if (!serverCallStreamObserver.isCancelled()) {
+                serverCallStreamObserver.onCompleted();
+            }
+        }
+    }
+    
+    @Override
+    public boolean isBusy() {
+        if (streamObserver instanceof ServerCallStreamObserver) {
+            return !((ServerCallStreamObserver) streamObserver).isReady();
+        }
+        return false;
+    }
+    
     @Override
     public boolean isConnected() {
-        return channel.isActive();
+        return channel != null && channel.isOpen() && channel.isActive();
     }
 }

pom.xml

@@ -128,6 +128,7 @@
         <!-- dependency version -->
         <spring-boot-dependencies.version>2.1.17.RELEASE</spring-boot-dependencies.version>
         <servlet-api.version>3.0</servlet-api.version>
+        <commons-lang.version>2.6</commons-lang.version>
         <commons-lang3.version>3.4</commons-lang3.version>
         <commons-io.version>2.2</commons-io.version>
         <commons-collections.version>3.2.2</commons-collections.version>
@@ -140,6 +141,7 @@
         <httpcore.version>4.4.1</httpcore.version>
         <httpclient.version>4.5</httpclient.version>
         <httpasyncclient.version>4.1.3</httpasyncclient.version>
+        <async-http-client.version>1.7.17</async-http-client.version>
         <mysql-connector-java.version>8.0.16</mysql-connector-java.version>
         <derby.version>10.14.2.0</derby.version>
         <cglib-nodep.version>2.1</cglib-nodep.version>
@@ -303,7 +305,7 @@
                     <encoding>UTF-8</encoding>
                     <consoleOutput>true</consoleOutput>
                     <failsOnError>true</failsOnError>
-                    <excludes>**/istio/model/**,**/nacos/test/**,**/api/grpc/auto/**,**/consistency/entity/**</excludes>
+                    <excludes>**/istio/model/**,**/nacos/test/**,**/api/grpc/auto/**</excludes>
                 </configuration>
                 <executions>
                     <execution>
@@ -338,7 +340,6 @@
                         <exclude>**/consistency/entity/**</exclude>
                         <exclude>**/*.txt</exclude>
                         <exclude>**/*.factories</exclude>
-                        <exclude>/console-ui/**</exclude>
                     </excludes>
                 </configuration>
                 <executions>
@@ -734,6 +735,12 @@
                 <version>${hessian.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>commons-lang</groupId>
+                <artifactId>commons-lang</artifactId>
+                <version>${commons-lang.version}</version>
+            </dependency>
+
             <!-- Apache commons -->
             <dependency>
                 <groupId>org.apache.commons</groupId>
@@ -771,6 +778,7 @@
                 <version>${commons-cli.version}</version>
             </dependency>
 
+
             <!-- Logging libs -->
             <dependency>
                 <groupId>org.slf4j</groupId>
@@ -833,6 +841,12 @@
                 <version>${httpasyncclient.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>com.ning</groupId>
+                <artifactId>async-http-client</artifactId>
+                <version>${async-http-client.version}</version>
+            </dependency>
+
             <!-- JDBC libs -->
             <dependency>
                 <groupId>mysql</groupId>
@@ -967,7 +981,6 @@
             </dependency>
 
             <!-- gRPC dependency start -->
-
             <dependency>
                 <groupId>io.grpc</groupId>
                 <artifactId>grpc-netty-shaded</artifactId>

test/src/test/java/com/alibaba/nacos/test/config/ConfigTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ * Copyright 1999-2020 Alibaba Group Holding Ltd.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package com.alibaba.nacos.client;
+package com.alibaba.nacos.test.config;
 
 import com.alibaba.nacos.api.NacosFactory;
 import com.alibaba.nacos.api.PropertyKeyConst;
@@ -56,14 +56,14 @@ public class ConfigTest {
         Properties properties = new Properties();
         properties.setProperty(PropertyKeyConst.SERVER_ADDR, "127.0.0.1:8848");
         
-        //properties.setProperty(PropertyKeyConst.SERVER_ADDR, "11.160.144.149:8848");
+        properties.setProperty(PropertyKeyConst.SERVER_ADDR, "11.160.144.149:8848");
-        //properties.setProperty(PropertyKeyConst.SERVER_ADDR, "11.160.67.159:8849");
+        //properties.setProperty(PropertyKeyConst.SERVER_ADDR, "11.160.67.159:7001");
         
         //properties.setProperty(PropertyKeyConst.SERVER_ADDR, "11.160.144.149:8848,11.160.144.148:8848,127.0.0.1:8848");
         //"11.239.114.187:8848,,11.239.113.204:8848,11.239.112.161:8848");
         //"11.239.114.187:8848");
-        properties.setProperty(PropertyKeyConst.USERNAME, "nacos");
+        //properties.setProperty(PropertyKeyConst.USERNAME, "nacos");
-        properties.setProperty(PropertyKeyConst.PASSWORD, "nacos");
+        //properties.setProperty(PropertyKeyConst.PASSWORD, "nacos");
         
         configService = NacosFactory.createConfigService(properties);
         //Thread.sleep(2000L);
@@ -189,11 +189,12 @@ public class ConfigTest {
     public void test2() throws Exception {
         final String dataId = "xiaochun.xxc";
         final String group = "xiaochun.xxc";
+        Random random = new Random();
         Properties properties = new Properties();
-        properties.setProperty(PropertyKeyConst.SERVER_ADDR, "11.160.144.149:8848");
+        properties.setProperty(PropertyKeyConst.SERVER_ADDR, "11.160.144.149:8848,11.160.144.148:8848");
         //"
         List<ConfigService> configServiceList = new ArrayList<ConfigService>();
-        for (int i = 0; i < 300; i++) {
+        for (int i = 0; i < 500; i++) {
             
             ConfigService configService = NacosFactory.createConfigService(properties);
             
@@ -205,8 +206,9 @@ public class ConfigTest {
                     
                 }
             };
-    
+            for (int j = 0; j < 50; j++) {
-            configService.addListener(dataId, group, listener);
+                configService.addListener(dataId + random.nextInt(200), group, listener);
+            }
             configServiceList.add(configService);
             System.out.println(configServiceList.size());
         }
@@ -223,6 +225,7 @@ public class ConfigTest {
                         boolean result = configService.publishConfig(dataId, group, "" + System.currentTimeMillis());
                         
                         times--;
+                        System.out.println("发布配置：" + result);
                         Thread.sleep(1000L);
                     } catch (Exception e) {
                         e.printStackTrace();
@@ -241,21 +244,30 @@ public class ConfigTest {
     public void test() throws Exception {
         
         //SnapShotSwitch.setIsSnapShot(false);
-        final Random random = new Random();
+        final Random random = new Random(System.currentTimeMillis());
         final String dataId = "xiaochun.xxc";
         final String group = "xiaochun.xxc";
-        
+        Listener listener = new AbstractListener() {
+            @Override
+            public void receiveConfigInfo(String configInfo) {
+                String[] s = configInfo.split("__");
+                System.out.println("receiveConfigInfo1 content:" + (System.currentTimeMillis() - Long.valueOf(s[1])));
+            }
+        };
         Thread th = new Thread(new Runnable() {
             @Override
             public void run() {
                 long start = System.currentTimeMillis();
-                int times = 1000;
+                int times = 10000;
                 while (times > 0) {
                     try {
-                        String content1 = System.currentTimeMillis() + "";
+                        String content1 = new String(new byte[5000]) + "__" + System.currentTimeMillis();
-                        boolean b = configService.publishConfig(dataId + random.nextInt(20), group, content1);
+                        System.out.println(content1.length());
+                        boolean b = configService.publishConfig(dataId + random.nextInt(400), group, content1);
                         times--;
-                        Thread.sleep(1000L);
+                        System.out.println("发布配置：" + b);
+                        
+                        Thread.sleep(500L);
                     } catch (Exception e) {
                         e.printStackTrace();
                     }
@@ -266,29 +278,15 @@ public class ConfigTest {
             }
             
         });
-    
         th.start();
         
-        Listener listener = new AbstractListener() {
+        for (int i = 0; i < 500; i++) {
-            @Override
+            String content1 = System.currentTimeMillis() + "";
-            public void receiveConfigInfo(String configInfo) {
+            configService.getConfigAndSignListener(dataId + i, group, 3000L, listener);
-                System.out.println(
-                        "receiveConfigInfo1 content:" + (System.currentTimeMillis() - Long.valueOf(configInfo)));
-                
-            }
-        };
-        
-        for (int i = 0; i < 20; i++) {
-            final int ls = i;
-            configService.addListener(dataId + i, group, listener);
-            
         }
         
         Thread.sleep(1000000L);
         
-        for (int i = 0; i < 20; i++) {
-            configService.removeListener(dataId + i, group, listener);
-        }
         System.out.println("remove listens.");
         
         Scanner scanner = new Scanner(System.in);
@@ -341,7 +339,6 @@ public class ConfigTest {
                         
                     }
                 }
-    
                 System.out.println(times);
                 System.out.println("Write Done");
             }