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

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<Void> 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<Void> sub1 = run(from, mid);
                CompletableFuture<Void> 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.");
    }
}

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<Void> run(int from, int to) {
            return CompletableFuture.runAsync(
                new PseudoTask(from, to)::compute, Thread::startVirtualThread);
        }

        protected void compute() {
            CompletableFuture<Void> 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<Void> 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.");
    }
}

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<Object> scope) {
            // 同样采用只提交一半任务的优化策略
            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 {
                    var sub = new PseudoTask(currentFrom, mid);
                    // 使用scope.fork()启动子任务，它会在虚拟线程中运行
                    scope.fork(() -> sub.compute(scope));
                }
            }
            return null; // 返回Void
        }
    }

    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();
        System.out.println("Starting recursive task with StructuredTaskScope...");
        PseudoTask.run(0, 1_000_000); // 启动根任务
        long endTime = System.currentTimeMillis();
        System.out.println("StructuredTaskScope recursive task completed in " + (endTime - startTime) + " ms.");
    }
}