在Loom中利用虚拟线程实现递归任务：告别ForkJoinPool的限制-java教程-PHP中文网

在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万个。这体现了在虚拟线程环境下，开发者仍然需要考虑任务分解的粒度与效率，尽管虚拟线程本身非常轻量。

BibiGPT-哔哔终结者

B站视频总结器-一键总结音视频内容

下载

方案二：利用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