Java 为虚拟线程配置线程池（执行器）

作用

为虚拟线程配置线程池（应叫做：执行器）的作用：使用执行器去开启虚拟线程，并用执行器管理虚拟线程的生命周期。

本质

为虚拟线程创建的ExecutorService，本质是虚拟线程创建器和任务提交器。

虚拟线程创建器：使用VirtualThread类中的私有静态final成员变量DEFAULT_SCHEDULER为平台线程，VirtualThreadFactory创建虚拟线程。

任务提交器：ExecutorService的submit方法将Runnable任务提交给ForkJoinPool。

代码

try (ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10; i++) {
        int finalI = i;
        executorService.submit(() -> {
            Thread thread = Thread.currentThread();
            System.out.println("executorService: " + thread + "," + thread.isVirtual() + ", isDaemon: " + thread.isDaemon());
            try {
                Thread.sleep(2000);
                System.out.println("executorService 2000 -----------------------------" + finalI);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }
}

源码

使用Executors类提供的静态方法创建虚拟线程的执行器。

package java.util.concurrent;

public final class Executors {

    /**
     * Creates an Executor that starts a new virtual Thread for each task.
     * The number of threads created by the Executor is unbounded.
     *
     * <p> This method is equivalent to invoking
     * {@link #newThreadPerTaskExecutor(ThreadFactory)} with a thread factory
     * that creates virtual threads.
     *
     * @return a new executor that creates a new virtual Thread for each task
     * @since 21
     */
    public static ExecutorService newVirtualThreadPerTaskExecutor() {
        ThreadFactory factory = Thread.ofVirtual().factory();
        return newThreadPerTaskExecutor(factory);
    }
    
    /**
     * Creates an Executor that starts a new Thread for each task.
     * The number of threads created by the Executor is unbounded.
     *
     * <p> Invoking {@link Future#cancel(boolean) cancel(true)} on a {@link
     * Future Future} representing the pending result of a task submitted to
     * the Executor will {@link Thread#interrupt() interrupt} the thread
     * executing the task.
     *
     * @param threadFactory the factory to use when creating new threads
     * @return a new executor that creates a new Thread for each task
     * @throws NullPointerException if threadFactory is null
     * @since 21
     */
    public static ExecutorService newThreadPerTaskExecutor(ThreadFactory threadFactory) {
        return ThreadPerTaskExecutor.create(threadFactory);
    }

虚拟线程的执行器ThreadPerTaskExecutor重写了ExecutorService的close()方法，awaitTermination将等待所有提交的虚拟线程任务执行完成后退出。

package java.util.concurrent;

class ThreadPerTaskExecutor extends ThreadContainer implements ExecutorService {

    @Override
    public void close() {
        awaitTermination();
    }
    
    /**
     * Waits for executor to terminate.
     */
    private void awaitTermination() {
        boolean terminated = isTerminated();
        if (!terminated) {
            tryShutdownAndTerminate(false);
            boolean interrupted = false;
            while (!terminated) {
                try {
                    terminated = awaitTermination(1L, TimeUnit.DAYS);
                } catch (InterruptedException e) {
                    if (!interrupted) {
                        tryShutdownAndTerminate(true);
                        interrupted = true;
                    }
                }
            }
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
}