Java虚拟线程与RecursiveAction/Task：兼容性与替代方案

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Supplier;

public class VirtualThreadRecursiveTask {

    // 示例1：基本递归任务，每个子任务启动一个虚拟线程
    record PseudoTask(int from, int to) {
        public static CompletableFuture<Void> run(int from, int to) {
            // 使用CompletableFuture.runAsync在虚拟线程上执行任务
            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 range [" + from + ", " + to + "]");
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(50)); // 模拟耗时操作
            } else {
                // 递归地创建并运行子任务
                CompletableFuture<Void> sub1 = run(from, mid);
                CompletableFuture<Void> sub2 = run(mid, to);
                sub1.join(); // 等待子任务完成
                sub2.join(); // 等待子任务完成
            }
        }
    }

    // 示例2：优化后的递归任务，减少虚拟线程创建数量
    record OptimizedPseudoTask(int from, int to) {
        public static CompletableFuture<Void> run(int from, int to) {
            return CompletableFuture.runAsync(
                new OptimizedPseudoTask(from, to)::compute, Thread::startVirtualThread);
        }

        protected void compute() {
            CompletableFuture<Void> pendingFutures = null;
            // 循环处理一部分任务，另一部分提交给新线程
            for (int currentFrom = this.from; ; ) {
                int mid = (currentFrom + to) >>> 1;
                if (mid == currentFrom) {
                    // 模拟实际处理
                    System.out.println(Thread.currentThread().getName() + ": Processing range [" + currentFrom + ", " + to + "]");
                    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(50));
                    break;
                } else {
                    // 提交一个子任务到新的虚拟线程
                    CompletableFuture<Void> sub = run(currentFrom, mid);
                    if (pendingFutures == null) {
                        pendingFutures = sub;
                    } else {
                        pendingFutures = CompletableFuture.allOf(pendingFutures, sub);
                    }
                    currentFrom = mid; // 当前线程处理另一半
                }
            }
            if (pendingFutures != null) {
                pendingFutures.join(); // 等待所有提交的子任务完成
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println("--- Running Basic PseudoTask ---");
        long startTime = System.currentTimeMillis();
        PseudoTask.run(0, 1_000).join(); // 执行任务并等待完成
        long endTime = System.currentTimeMillis();
        System.out.println("Basic PseudoTask completed in " + (endTime - startTime) + " ms");

        System.out.println("\n--- Running Optimized PseudoTask ---");
        startTime = System.currentTimeMillis();
        OptimizedPseudoTask.run(0, 1_000_000).join(); // 执行优化后的任务
        endTime = System.currentTimeMillis();
        System.out.println("Optimized PseudoTask completed in " + (endTime - startTime) + " ms");
    }
}

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

public class StructuredVirtualThreadTask {

    // 示例：使用StructuredTaskScope实现递归任务
    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(); // 等待所有子任务完成
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("Task interrupted", e);
            }
        }

        protected Void compute(StructuredTaskScope<Object> scope) {
            for (int currentFrom = this.from; ; ) {
                int mid = (currentFrom + to) >>> 1;
                if (mid == currentFrom) {
                    // 模拟实际处理
                    System.out.println(Thread.currentThread().getName() + ": Processing range [" + currentFrom + ", " + to + "]");
                    LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(50));
                    break;
                } else {
                    // 使用scope.fork()在虚拟线程中启动子任务
                    var sub = new PseudoTask(currentFrom, mid);
                    scope.fork(() -> sub.compute(scope));
                    currentFrom = mid; // 当前线程处理另一半
                }
            }
            return null;
        }
    }

    public static void main(String[] args) {
        System.out.println("--- Running StructuredTaskScope PseudoTask ---");
        long startTime = System.currentTimeMillis();
        PseudoTask.run(0, 1_000_000); // 执行任务
        long endTime = System.currentTimeMillis();
        System.out.println("StructuredTaskScope PseudoTask completed in " + (endTime - startTime) + " ms");
    }
}