Java 虚拟线程启动流程及调度器源码解析

启动虚拟线程方式

方式一

Thread.ofVirtual().start(() -> {

});

方式二

Runnable runnable = () -> {};

Thread.startVirtualThread(runnable);

源码

startVirtualThread是Thread的静态方法，在Java21中引入。

在方法内部调用了ThreadBuilders类中的静态方法newVirtualThread，并开启了线程的运行。

package java.lang;

public class Thread implements Runnable {

    /**
     * Creates a virtual thread to execute a task and schedules it to execute.
     *
     * <p> This method is equivalent to:
     * <pre>{@code Thread.ofVirtual().start(task); }</pre>
     *
     * @param task the object to run when the thread executes
     * @return a new, and started, virtual thread
     * @see <a href="#inheritance">Inheritance when creating threads</a>
     * @since 21
     */
    public static Thread startVirtualThread(Runnable task) {
        Objects.requireNonNull(task);
        var thread = ThreadBuilders.newVirtualThread(null, null, 0, task);
        thread.start();
        return thread;
    }
    
}

newVirtualThread中第一个参数是调度器，在startVirtualThread方法中调用时传入的是null。

package java.lang;

class ThreadBuilders {
    private ThreadBuilders() { }
    
    /**
     * Creates a new virtual thread to run the given task.
     */
    static Thread newVirtualThread(Executor scheduler,
                                   String name,
                                   int characteristics,
                                   Runnable task) {
        if (ContinuationSupport.isSupported()) {
            return new VirtualThread(scheduler, name, characteristics, task);
        } else {
            if (scheduler != null)
                throw new UnsupportedOperationException();
            return new BoundVirtualThread(name, characteristics, task);
        }
    }
    
    /**
     * ThreadBuilder.OfVirtual implementation.
     */
    static final class VirtualThreadBuilder
            extends BaseThreadBuilder implements OfVirtual {
        private Executor scheduler;

        VirtualThreadBuilder() {
        }
        
        @Override
        public Thread unstarted(Runnable task) {
            Objects.requireNonNull(task);
            var thread = newVirtualThread(scheduler, nextThreadName(), characteristics(), task);
            UncaughtExceptionHandler uhe = uncaughtExceptionHandler();
            if (uhe != null)
                thread.uncaughtExceptionHandler(uhe);
            return thread;
        }

        @Override
        public Thread start(Runnable task) {
            Thread thread = unstarted(task);
            thread.start();
            return thread;
        }

        @Override
        public ThreadFactory factory() {
            return new VirtualThreadFactory(scheduler, name(), counter(), characteristics(),
                    uncaughtExceptionHandler());
        }
        
    }
    
}

虚拟线程默认的调度器是ForkJoinPool

package java.lang;

final class VirtualThread extends BaseVirtualThread {
    private static final ForkJoinPool DEFAULT_SCHEDULER = createDefaultScheduler();
    
    @SuppressWarnings("removal")
    private static ForkJoinPool createDefaultScheduler() {
        ForkJoinWorkerThreadFactory factory = pool -> {
            PrivilegedAction<ForkJoinWorkerThread> pa = () -> new CarrierThread(pool);
            return AccessController.doPrivileged(pa);
        };
        PrivilegedAction<ForkJoinPool> pa = () -> {
            int parallelism, maxPoolSize, minRunnable;
            String parallelismValue = System.getProperty("jdk.virtualThreadScheduler.parallelism");
            String maxPoolSizeValue = System.getProperty("jdk.virtualThreadScheduler.maxPoolSize");
            String minRunnableValue = System.getProperty("jdk.virtualThreadScheduler.minRunnable");
            ...
        }
    }
    
    /**
     * Creates a new {@code VirtualThread} to run the given task with the given
     * scheduler. If the given scheduler is {@code null} and the current thread
     * is a platform thread then the newly created virtual thread will use the
     * default scheduler. If given scheduler is {@code null} and the current
     * thread is a virtual thread then the current thread's scheduler is used.
     *
     * @param scheduler the scheduler or null
     * @param name thread name
     * @param characteristics characteristics
     * @param task the task to execute
     */
    VirtualThread(Executor scheduler, String name, int characteristics, Runnable task) {
        super(name, characteristics, /*bound*/ false);
        Objects.requireNonNull(task);

        // choose scheduler if not specified
        if (scheduler == null) {
            Thread parent = Thread.currentThread();
            if (parent instanceof VirtualThread vparent) {
                scheduler = vparent.scheduler;
            } else {
                scheduler = DEFAULT_SCHEDULER;
            }
        }

        this.scheduler = scheduler;
        this.cont = new VThreadContinuation(this, task);
        this.runContinuation = this::runContinuation;
    }
    
}