Support use virtual thread as platform thread

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/FailbackClusterInvoker.java

@@ -19,11 +19,10 @@
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.timer.HashedWheelTimer;
+import org.apache.dubbo.common.threadpool.ExecutorsUtil;
 import org.apache.dubbo.common.timer.Timeout;
 import org.apache.dubbo.common.timer.Timer;
 import org.apache.dubbo.common.timer.TimerTask;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
 import org.apache.dubbo.rpc.AsyncRpcResult;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
@@ -91,12 +90,8 @@ private void addFailed(
         if (failTimer == null) {
             synchronized (this) {
                 if (failTimer == null) {
-                    failTimer = new HashedWheelTimer(
-                            new NamedThreadFactory("failback-cluster-timer", true),
-                            1,
-                            TimeUnit.SECONDS,
-                            32,
-                            failbackTasks);
+                    failTimer =
+                            ExecutorsUtil.newTimer("failback-cluster-timer", 1, TimeUnit.SECONDS, 32, failbackTasks);
                 }
             }
         }

dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/AbstractDynamicConfiguration.java

@@ -19,10 +19,11 @@
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.threadpool.ExecutorsUtil;
 import org.apache.dubbo.common.utils.StringUtils;
 
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -82,7 +83,7 @@ public abstract class AbstractDynamicConfiguration implements DynamicConfigurati
     /**
      * The thread pool for workers who execute the tasks
      */
-    private final ThreadPoolExecutor workersThreadPool;
+    private final ExecutorService workersThreadPool;
 
     private final String group;
 
@@ -221,7 +222,7 @@ protected final <V> V execute(Callable<V> task, long timeout) {
         return value;
     }
 
-    protected ThreadPoolExecutor getWorkersThreadPool() {
+    protected ExecutorService getWorkersThreadPool() {
         return workersThreadPool;
     }
 
@@ -235,15 +236,15 @@ private void shutdownWorkersThreadPool() {
         }
     }
 
-    protected ThreadPoolExecutor initWorkersThreadPool(
+    protected ExecutorService initWorkersThreadPool(
             String threadPoolPrefixName, int threadPoolSize, long keepAliveTime) {
-        return new ThreadPoolExecutor(
+        return ExecutorsUtil.newExecutorService(
                 threadPoolSize,
                 threadPoolSize,
                 keepAliveTime,
                 TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<>(),
-                new NamedThreadFactory(threadPoolPrefixName, true));
+                threadPoolPrefixName);
     }
 
     protected static String getThreadPoolPrefixName(URL url) {

dubbo-common/src/main/java/org/apache/dubbo/common/resource/GlobalResourcesRepository.java

@@ -19,13 +19,13 @@
 import org.apache.dubbo.common.config.ConfigurationUtils;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
+import org.apache.dubbo.common.threadpool.ExecutorsUtil;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT;
@@ -95,8 +95,13 @@ public ExecutorService getExecutorService() {
                     if (logger.isInfoEnabled()) {
                         logger.info("Creating global shared handler ...");
                     }
-                    executorService =
-                            Executors.newCachedThreadPool(new NamedThreadFactory("Dubbo-global-shared-handler", true));
+                    executorService = ExecutorsUtil.newExecutorService(
+                            0,
+                            Integer.MAX_VALUE,
+                            60L,
+                            TimeUnit.SECONDS,
+                            new SynchronousQueue<>(),
+                            "Dubbo-global-shared-handler");
                 }
             }
         }

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/DefaultExecutorAdaptor.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.common.threadpool;
+
+import org.apache.dubbo.common.timer.HashedWheelTimer;
+import org.apache.dubbo.common.timer.Timer;
+import org.apache.dubbo.common.utils.NamedThreadFactory;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class DefaultExecutorAdaptor implements ExecutorAdaptor {
+    @Override
+    public ExecutorService newSingleThreadExecutorService(String namePrefix) {
+        return Executors.newSingleThreadExecutor(new NamedThreadFactory(namePrefix, true));
+    }
+
+    @Override
+    public ExecutorService newExecutorService(
+            int corePoolSize,
+            int maximumPoolSize,
+            long keepAliveTime,
+            TimeUnit unit,
+            BlockingQueue<Runnable> workQueue,
+            String namePrefix,
+            RejectedExecutionHandler handler) {
+        return new ThreadPoolExecutor(
+                corePoolSize,
+                maximumPoolSize,
+                keepAliveTime,
+                unit,
+                workQueue,
+                new NamedThreadFactory(namePrefix, true),
+                handler);
+    }
+
+    @Override
+    public ScheduledExecutorService newScheduledExecutorService(int corePoolSize, String namePrefix) {
+        return new ScheduledThreadPoolExecutor(corePoolSize, new NamedThreadFactory(namePrefix, true));
+    }
+
+    @Override
+    public Timer newTimer(
+            String namePrefix, long tickDuration, TimeUnit unit, int ticksPerWheel, long maxPendingTimeouts) {
+        return new HashedWheelTimer(
+                new NamedThreadFactory(namePrefix, true), tickDuration, unit, ticksPerWheel, maxPendingTimeouts);
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/ExecutorAdaptor.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.common.threadpool;
+
+import org.apache.dubbo.common.timer.Timer;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public interface ExecutorAdaptor {
+    ExecutorService newSingleThreadExecutorService(String namePrefix);
+
+    ExecutorService newExecutorService(
+            int corePoolSize,
+            int maximumPoolSize,
+            long keepAliveTime,
+            TimeUnit unit,
+            BlockingQueue<Runnable> workQueue,
+            String namePrefix,
+            RejectedExecutionHandler handler);
+
+    ScheduledExecutorService newScheduledExecutorService(int corePoolSize, String namePrefix);
+
+    Timer newTimer(String namePrefix, long tickDuration, TimeUnit unit, int ticksPerWheel, long maxPendingTimeouts);
+}

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/ExecutorsUtil.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.common.threadpool;
+
+import org.apache.dubbo.common.timer.Timer;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class ExecutorsUtil {
+    private static ExecutorAdaptor getExecutorAdaptor() {
+        return new DefaultExecutorAdaptor();
+    }
+
+    public static ExecutorService newSingleThreadExecutorService(String namePrefix) {
+        return getExecutorAdaptor().newSingleThreadExecutorService(namePrefix);
+    }
+
+    public static ExecutorService newExecutorService(
+            int corePoolSize,
+            int maximumPoolSize,
+            long keepAliveTime,
+            TimeUnit unit,
+            BlockingQueue<Runnable> workQueue,
+            String namePrefix) {
+        return getExecutorAdaptor()
+                .newExecutorService(
+                        corePoolSize,
+                        maximumPoolSize,
+                        keepAliveTime,
+                        unit,
+                        workQueue,
+                        namePrefix,
+                        new ThreadPoolExecutor.AbortPolicy());
+    }
+
+    public static ExecutorService newExecutorService(
+            int corePoolSize,
+            int maximumPoolSize,
+            long keepAliveTime,
+            TimeUnit unit,
+            BlockingQueue<Runnable> workQueue,
+            String namePrefix,
+            RejectedExecutionHandler handler) {
+        return getExecutorAdaptor()
+                .newExecutorService(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, namePrefix, handler);
+    }
+
+    public static ScheduledExecutorService newScheduledExecutorService(int corePoolSize, String namePrefix) {
+        return getExecutorAdaptor().newScheduledExecutorService(corePoolSize, namePrefix);
+    }
+
+    public static Timer newTimer(String namePrefix, long tickDuration, TimeUnit unit) {
+        return newTimer(namePrefix, tickDuration, unit, 512);
+    }
+
+    public static Timer newTimer(String namePrefix, long tickDuration, TimeUnit unit, int ticksPerWheel) {
+        return newTimer(namePrefix, tickDuration, unit, ticksPerWheel, -1);
+    }
+
+    public static Timer newTimer(
+            String namePrefix, long tickDuration, TimeUnit unit, int ticksPerWheel, long maxPendingTimeouts) {
+        return getExecutorAdaptor().newTimer(namePrefix, tickDuration, unit, ticksPerWheel, maxPendingTimeouts);
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/MemoryLimitCalculator.java

@@ -17,9 +17,7 @@
 package org.apache.dubbo.common.threadpool;
 
 import org.apache.dubbo.common.resource.GlobalResourcesRepository;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
 
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -50,7 +48,7 @@ private static void checkAndScheduleRefresh() {
             refresh();
             if (refreshStarted.compareAndSet(false, true)) {
                 ScheduledExecutorService scheduledExecutorService =
-                        Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-Memory-Calculator"));
+                        ExecutorsUtil.newScheduledExecutorService(1, "Dubbo-Memory-Calculator");
                 // check every 50 ms to improve performance
                 scheduledExecutorService.scheduleWithFixedDelay(
                         MemoryLimitCalculator::refresh, 50, 50, TimeUnit.MILLISECONDS);

dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java

@@ -22,10 +22,10 @@
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.common.store.DataStore;
+import org.apache.dubbo.common.threadpool.ExecutorsUtil;
 import org.apache.dubbo.common.threadpool.ThreadPool;
 import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
 import org.apache.dubbo.common.utils.ExecutorUtil;
-import org.apache.dubbo.common.utils.NamedThreadFactory;
 import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.config.ConsumerConfig;
 import org.apache.dubbo.config.ModuleConfig;
@@ -42,9 +42,10 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SHARED_EXECUTOR_SERVICE_COMPONENT_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE;
@@ -298,8 +299,8 @@ public ScheduledExecutorService getServiceExportExecutor() {
                 int coreSize = getExportThreadNum();
                 String applicationName = applicationModel.tryGetApplicationName();
                 applicationName = StringUtils.isEmpty(applicationName) ? "app" : applicationName;
-                serviceExportExecutor = Executors.newScheduledThreadPool(
-                        coreSize, new NamedThreadFactory("Dubbo-" + applicationName + "-service-export", true));
+                serviceExportExecutor = ExecutorsUtil.newScheduledExecutorService(
+                        coreSize, "Dubbo-" + applicationName + "-service-export");
             }
         }
         return serviceExportExecutor;
@@ -327,8 +328,13 @@ public ExecutorService getServiceReferExecutor() {
                 int coreSize = getReferThreadNum();
                 String applicationName = applicationModel.tryGetApplicationName();
                 applicationName = StringUtils.isEmpty(applicationName) ? "app" : applicationName;
-                serviceReferExecutor = Executors.newFixedThreadPool(
-                        coreSize, new NamedThreadFactory("Dubbo-" + applicationName + "-service-refer", true));
+                serviceReferExecutor = ExecutorsUtil.newExecutorService(
+                        coreSize,
+                        coreSize,
+                        0L,
+                        TimeUnit.MILLISECONDS,
+                        new LinkedBlockingQueue<>(),
+                        "Dubbo-" + applicationName + "-service-refer");
             }
         }
         return serviceReferExecutor;