在Loom中利用虚拟线程实现递归任务：告别ForkJoinPool的限制

本文探讨了Java Loom中RecursiveAction和RecursiveTask与虚拟线程的兼容性。由于它们设计上依赖于ForkJoinPool及其特定的工作线程，无法直接与虚拟线程配合使用。文章提供了两种替代方案：一是利用CompletableFuture结合虚拟线程工厂实现自定义递归任务，二是介绍孵化阶段的StructuredTaskScope API，以更结构化的方式管理并发子任务。这些方法允许开发者在虚拟线程环境下高效地执行递归分解任务，同时避免了传统ForkJoinPool的限制。

RecursiveAction与RecursiveTask不兼容虚拟线程的原因

recursiveaction和recursivetask是java并发api中用于支持分治算法的抽象基类，它们专为forkjoinpool设计。正如其文档所述，它们是“在forkjoinpool中运行任务的抽象基类”。forkjoinpool虽然允许通过线程工厂进行定制，但它要求的是forkjoinpool.forkjoinworkerthreadfactory接口的实现，用于生产forkjoinworkerthread实例。

ForkJoinWorkerThread是Thread的子类，这意味着它们是平台线程（Platform Thread）的特化。虚拟线程（Virtual Thread）则是一种轻量级的线程，由Thread.Builder.OfVirtual创建，与传统的平台线程在实现机制上存在根本差异。因此，ForkJoinPool无法通过其内部的线程工厂机制来创建或使用虚拟线程。这意味着，RecursiveAction和RecursiveTask无法直接与虚拟线程结合使用。

重新思考递归任务在虚拟线程环境下的实现

RecursiveAction和RecursiveTask的核心价值在于提供了任务分解的抽象，并与ForkJoinPool的窃取式工作队列（work-stealing queue）机制相结合，以有效地平衡平台线程上的工作负载。然而，当引入虚拟线程时，这种工作负载平衡的必要性发生了变化。虚拟线程非常轻量，可以大量创建，并且由JVM运行时高效地映射到少量平台线程上。如果每个子任务都可以在自己的虚拟线程上运行，那么ForkJoinPool提供的复杂工作调度和负载平衡机制可能不再是必需的。开发者可以更直接地实现任务的递归分解，并利用虚拟线程的优势。

方案一：基于CompletableFuture的自定义递归任务

在虚拟线程环境下，我们可以利用CompletableFuture的异步执行能力来构建自定义的递归任务。CompletableFuture.runAsync()或CompletableFuture.supplyAsync()方法可以接受一个Executor，我们可以传入一个创建虚拟线程的Executor，例如Thread::startVirtualThread。

以下是一个简单的示例，演示如何使用CompletableFuture和虚拟线程实现一个递归任务：

import java.util.concurrent.CompletableFuture;import java.util.concurrent.TimeUnit;import java.util.concurrent.locks.LockSupport;public class VirtualThreadRecursiveTask {    /**     * 模拟一个递归任务，使用虚拟线程执行子任务。     * 每个子任务都会在新的虚拟线程中运行。     */    record PseudoTask(int from, int to) {        public static CompletableFuture run(int from, int to) {            // 使用Thread::startVirtualThread作为Executor，确保任务在虚拟线程中运行            return CompletableFuture.runAsync(                new PseudoTask(from, to)::compute, Thread::startVirtualThread);        }        protected void compute() {            int mid = (from + to) >>> 1; // 计算中间值            if (mid == from) {                // 达到基本情况，模拟实际处理，可能包含阻塞操作                System.out.println(Thread.currentThread().getName() + " processing: [" + from + ", " + to + ")");                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(50)); // 模拟耗时操作            } else {                // 递归分解任务                CompletableFuture sub1 = run(from, mid);                CompletableFuture sub2 = run(mid, to);                // 等待所有子任务完成。这里的join()会阻塞当前虚拟线程，                // 但由于虚拟线程的特性，它可以在阻塞时被卸载，不会阻塞底层平台线程。                sub1.join();                 sub2.join();            }        }    }    public static void main(String[] args) throws InterruptedException {        long startTime = System.currentTimeMillis();        System.out.println("Starting recursive task with virtual threads...");        PseudoTask.run(0, 1_000).join(); // 启动根任务并等待其完成        long endTime = System.currentTimeMillis();        System.out.println("Recursive task completed in " + (endTime - startTime) + " ms.");    }}

注意事项：

join()的阻塞性：尽管CompletableFuture.join()是阻塞的，但在虚拟线程中，这种阻塞不会像在平台线程中那样导致底层操作系统线程的浪费。当虚拟线程被阻塞时，它会被JVM卸载，允许底层平台线程执行其他可运行的虚拟线程。线程数量：上述示例中，每个子任务都会创建一个新的虚拟线程，这在任务分解粒度很细时可能导致创建大量的虚拟线程。对于PseudoTask.run(0, 1_000)这样的调用，可能创建接近2000个虚拟线程。虽然虚拟线程非常轻量，但过多的创建和调度仍可能带来开销。

为了优化线程创建数量，可以采用类似ForkJoinPool中“工作窃取”的策略，即一个线程处理一部分任务，另一部分任务提交给其他线程。例如，只将一半范围的任务提交到新的虚拟线程，而另一半则在当前虚拟线程中局部处理：

import java.util.concurrent.CompletableFuture;import java.util.concurrent.TimeUnit;import java.util.concurrent.locks.LockSupport;public class OptimizedVirtualThreadRecursiveTask {    /**     * 优化后的递归任务，将一半子任务提交到新的虚拟线程，另一半在当前线程处理。     */    record PseudoTask(int from, int to) {        public static CompletableFuture run(int from, int to) {            return CompletableFuture.runAsync(                new PseudoTask(from, to)::compute, Thread::startVirtualThread);        }        protected void compute() {            CompletableFuture pendingFutures = null; // 用于收集异步子任务的CompletableFuture            // 循环处理，直到达到基本情况            for (int currentFrom = this.from; ; currentFrom = (currentFrom + to) >>> 1) {                int mid = (currentFrom + to) >>> 1;                if (mid == currentFrom) {                    // 达到基本情况，模拟实际处理                    System.out.println(Thread.currentThread().getName() + " processing: [" + currentFrom + ", " + to + ")");                    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(50));                    break; // 退出循环                } else {                    // 提交一半任务到新的虚拟线程                    CompletableFuture subTaskFuture = run(currentFrom, mid);                    if (pendingFutures == null) {                        pendingFutures = subTaskFuture;                    } else {                        // 将新的子任务Future与之前的Future组合                        pendingFutures = CompletableFuture.allOf(pendingFutures, subTaskFuture);                    }                    // 另一半任务 (mid到to) 将在下一次循环中由当前虚拟线程处理                }            }            // 等待所有异步提交的子任务完成            if (pendingFutures != null) {                pendingFutures.join();            }        }    }    public static void main(String[] args) throws InterruptedException {        long startTime = System.currentTimeMillis();        System.out.println("Starting optimized recursive task with virtual threads...");        PseudoTask.run(0, 1_000_000).join(); // 启动根任务并等待其完成，处理更大范围        long endTime = System.currentTimeMillis();        System.out.println("Optimized recursive task completed in " + (endTime - startTime) + " ms.");    }}

这种优化策略对于PseudoTask.run(0, 1_000_000)这样的大范围任务会产生显著差异，它将创建大约100万个虚拟线程，而不是200万个。这体现了在虚拟线程环境下，开发者仍然需要考虑任务分解的粒度与效率，尽管虚拟线程本身非常轻量。

方案二：利用StructuredTaskScope进行结构化并发 (孵化中)

Java的孵化模块jdk.incubator.concurrent中引入了StructuredTaskScope，它提供了一种更结构化的方式来管理并发任务的生命周期。StructuredTaskScope允许在一个作用域内启动多个子任务，并等待所有子任务完成，或者在任何子任务失败时关闭整个作用域。这与递归任务的分解和聚合模式非常契合。

import jdk.incubator.concurrent.StructuredTaskScope;import java.util.concurrent.TimeUnit;import java.util.concurrent.locks.LockSupport;public class StructuredTaskScopeRecursiveTask {    /**     * 使用StructuredTaskScope实现递归任务。     * 注意：StructuredTaskScope是孵化API，可能在未来版本中发生变化。     */    record PseudoTask(int from, int to) {        public static void run(int from, int to) {            // 使用ShutdownOnFailure策略，任何子任务失败都会导致整个作用域关闭            try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {                new PseudoTask(from, to).compute(scope);                scope.join(); // 等待所有子任务完成                scope.throwIfFailed(); // 抛出任何失败的异常            } catch (InterruptedException e) {                Thread.currentThread().interrupt();                throw new IllegalStateException("Task interrupted", e);            } catch (Exception e) {                throw new RuntimeException("Task failed", e);            }        }        protected Void compute(StructuredTaskScope