Qos heart

dubbo-common/src/main/java/org/apache/dubbo/common/utils/ExecutorUtil.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ScheduledFuture;
 
 public class ExecutorUtil {
     private static final Logger logger = LoggerFactory.getLogger(ExecutorUtil.class);
@@ -45,9 +46,10 @@ public static boolean isTerminated(Executor executor) {
 
     /**
      * Use the shutdown pattern from:
-     *  https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
+     * https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
+     *
      * @param executor the Executor to shutdown
-     * @param timeout the timeout in milliseconds before termination
+     * @param timeout  the timeout in milliseconds before termination
      */
     public static void gracefulShutdown(Executor executor, int timeout) {
         if (!(executor instanceof ExecutorService) || isTerminated(executor)) {
@@ -131,4 +133,11 @@ public static URL setThreadName(URL url, String defaultName) {
         url = url.addParameter(Constants.THREAD_NAME_KEY, name);
         return url;
     }
+
+    public static void cancelScheduledFuture(ScheduledFuture<?> scheduledFuture) {
+        ScheduledFuture<?> future = scheduledFuture;
+        if (future != null && !future.isCancelled()) {
+            future.cancel(true);
+        }
+    }
 }

dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/DubboMonitor.java

@@ -20,6 +20,7 @@
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.ExecutorUtil;
 import org.apache.dubbo.common.utils.NamedThreadFactory;
 import org.apache.dubbo.monitor.Monitor;
 import org.apache.dubbo.monitor.MonitorService;
@@ -209,7 +210,7 @@ public boolean isAvailable() {
     @Override
     public void destroy() {
         try {
-            sendFuture.cancel(true);
+            ExecutorUtil.cancelScheduledFuture(sendFuture);
         } catch (Throwable t) {
             logger.error("Unexpected error occur at cancel sender timer, cause: " + t.getMessage(), t);
         }

dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/server/handler/QosProcessHandler.java

@@ -26,9 +26,11 @@
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.string.StringDecoder;
 import io.netty.handler.codec.string.StringEncoder;
+import io.netty.handler.timeout.IdleStateEvent;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.CharsetUtil;
 import io.netty.util.concurrent.ScheduledFuture;
+import org.apache.dubbo.common.utils.ExecutorUtil;
 
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -93,6 +95,14 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
         }
     }
 
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            ExecutorUtil.cancelScheduledFuture(welcomeFuture);
+            ctx.close();
+        }
+    }
+
     // G for GET, and P for POST
     private static boolean isHttp(int magic) {
         return magic == 'G' || magic == 'P';

dubbo-registry/dubbo-registry-default/src/main/java/org/apache/dubbo/registry/dubbo/DubboRegistry.java

@@ -38,7 +38,6 @@
 
 /**
  * DubboRegistry
- *
  */
 public class DubboRegistry extends FailbackRegistry {
 
@@ -124,9 +123,7 @@ public void destroy() {
         super.destroy();
         try {
             // Cancel the reconnection timer
-            if (!reconnectFuture.isCancelled()) {
-                reconnectFuture.cancel(true);
-            }
+            ExecutorUtil.cancelScheduledFuture(reconnectFuture);
         } catch (Throwable t) {
             logger.warn("Failed to cancel reconnect timer", t);
         }

dubbo-registry/dubbo-registry-multicast/src/main/java/org/apache/dubbo/registry/multicast/MulticastRegistry.java

@@ -297,9 +297,7 @@ public boolean isAvailable() {
     public void destroy() {
         super.destroy();
         try {
-            if (cleanFuture != null) {
-                cleanFuture.cancel(true);
-            }
+            ExecutorUtil.cancelScheduledFuture(cleanFuture);
         } catch (Throwable t) {
             logger.warn(t.getMessage(), t);
         }
@@ -341,8 +339,8 @@ protected void unregistered(URL url) {
                 if (urls != null) {
                     urls.remove(url);
                 }
-                if (urls == null || urls.isEmpty()){
-                    if (urls == null){
+                if (urls == null || urls.isEmpty()) {
+                    if (urls == null) {
                         urls = new ConcurrentHashSet<URL>();
                     }
                     URL empty = url.setProtocol(Constants.EMPTY_PROTOCOL);

dubbo-remoting/dubbo-remoting-p2p/src/main/java/org/apache/dubbo/remoting/p2p/exchange/support/FileExchangeGroup.java

@@ -17,6 +17,7 @@
 package org.apache.dubbo.remoting.p2p.exchange.support;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ExecutorUtil;
 import org.apache.dubbo.common.utils.IOUtils;
 import org.apache.dubbo.common.utils.NamedThreadFactory;
 import org.apache.dubbo.common.utils.NetUtils;
@@ -70,9 +71,7 @@ public void run() {
     public void close() {
         super.close();
         try {
-            if (!checkModifiedFuture.isCancelled()) {
-                checkModifiedFuture.cancel(true);
-            }
+            ExecutorUtil.cancelScheduledFuture(checkModifiedFuture);
         } catch (Throwable t) {
             logger.error(t.getMessage(), t);
         }

dubbo-remoting/dubbo-remoting-p2p/src/main/java/org/apache/dubbo/remoting/p2p/support/FileGroup.java

@@ -17,6 +17,7 @@
 package org.apache.dubbo.remoting.p2p.support;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.utils.ExecutorUtil;
 import org.apache.dubbo.common.utils.IOUtils;
 import org.apache.dubbo.common.utils.NamedThreadFactory;
 import org.apache.dubbo.common.utils.NetUtils;
@@ -67,9 +68,7 @@ public void run() {
     public void close() {
         super.close();
         try {
-            if (!checkModifiedFuture.isCancelled()) {
-                checkModifiedFuture.cancel(true);
-            }
+            ExecutorUtil.cancelScheduledFuture(checkModifiedFuture);
         } catch (Throwable t) {
             logger.error(t.getMessage(), t);
         }