从Project Loom到Virtual Threads：Java高并发编程的范式革命与性能实测

// build.gradle (或Maven)
plugins {
    id 'java'
}
sourceCompatibility = '21'
repositories {
    mavenCentral()
}

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.*;
import java.util.stream.IntStream;

public class VirtualThreadBenchmark {

    private static final int TASK_COUNT = 100_000;
    private static final int SIMULATED_IO_MS = 20;

    // 模拟阻塞IO任务
    private static Runnable blockingTask(int id) {
        return () -> {
            try {
                Thread.sleep(SIMULATED_IO_MS); // 模拟网络/DB阻塞
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };
    }

    // 方案A：传统平台线程池（固定大小，极限线程数）
    private static void platformThreadPoolTest() throws InterruptedException {
        System.out.println("=== Platform Thread Pool (Fixed 500) ===");
        ExecutorService executor = Executors.newFixedThreadPool(500);
        CountDownLatch latch = new CountDownLatch(TASK_COUNT);
        Instant start = Instant.now();

        IntStream.range(0, TASK_COUNT).forEach(i -> {
            executor.submit(() -> {
                blockingTask(i).run();
                latch.countDown();
            });
        });
        latch.await();
        executor.shutdown();
        System.out.println("Elapsed: " + Duration.between(start, Instant.now()).toMillis() + "ms");
    }

    // 方案B：虚拟线程（每个任务新建虚拟线程，轻量）
    private static void virtualThreadPerTaskTest() throws InterruptedException {
        System.out.println("=== Virtual Thread (Per-Task) ===");
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            CountDownLatch latch = new CountDownLatch(TASK_COUNT);
            Instant start = Instant.now();

            IntStream.range(0, TASK_COUNT).forEach(i -> {
                executor.submit(() -> {
                    blockingTask(i).run();
                    latch.countDown();
                });
            });
            latch.await();
            System.out.println("Elapsed: " + Duration.between(start, Instant.now()).toMillis() + "ms");
        }
    }

    // 方案C：虚拟线程 + 有界调度器（模拟限制并发数）
    private static void virtualThreadBoundedTest(int concurrencyLimit) throws InterruptedException {
        System.out.println("=== Virtual Thread (Bounded " + concurrencyLimit + " semaphore) ===");
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            Semaphore semaphore = new Semaphore(concurrencyLimit);
            CountDownLatch latch = new CountDownLatch(TASK_COUNT);
            Instant start = Instant.now();

            IntStream.range(0, TASK_COUNT).forEach(i -> {
                executor.submit(() -> {
                    try {
                        semaphore.acquire();
                        blockingTask(i).run();
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } finally {
                        semaphore.release();
                        latch.countDown();
                    }
                });
            });
            latch.await();
            System.out.println("Elapsed: " + Duration.between(start, Instant.now()).toMillis() + "ms");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println("Available processors: " + Runtime.getRuntime().availableProcessors());
        // 预热JIT
        for (int i = 0; i < 1000; i++) {
            blockingTask(i).run();
        }

        // 跑测（顺序执行，避免资源争抢）
        platformThreadPoolTest();          // 预计 4000~5000ms (因为500线程分20批完成，每批20ms，总计约4000ms)
        virtualThreadPerTaskTest();        // 预计 20~30ms (十万虚拟线程瞬间创建，载体线程自动调度)
        virtualThreadBoundedTest(200);     // 预计 10000ms (限流200并发，每批20ms，共500批)
    }
}

// java.lang.VirtualThread 核心调度循环（简化）
private void submitRunContinuation() {
    // 将当前虚拟线程提交给调度器（ForkJoinPool）
    scheduler.execute(() -> {
        try {
            // 绑定到当前载体线程（Carrier Thread）
            carrierThread = Thread.currentThread();
            // 执行延续（继续上次yield的位置）
            continuation.run();
        } finally {
            carrierThread = null;
        }
    });
}

// 反例：计算密集型使用虚拟线程
Runnable cpuBound = () -> {
    BigInteger.probablePrime(2048, new Random()); // 高CPU消耗
};
// 应使用 newFixedThreadPool(cores)

// 坏实践
synchronized(lock) {
    Thread.sleep(1000); // 载体线程会被阻塞
}

// 好实践
ReentrantLock lock = new ReentrantLock();
lock.lock();
try {
    // 阻塞操作，虚拟线程会yield，载体释放
    Thread.sleep(1000);
} finally {
    lock.unlock();
}

spring:
  threads:
    virtual:
      enabled: true

@Configuration
public class VirtualThreadConfig {
    @Bean
    public Executor virtualTaskExecutor() {
        return Executors.newVirtualThreadPerTaskExecutor();
    }
}

@Service
public class AsyncService {
    @Async("virtualTaskExecutor")
    public CompletableFuture<String> fetchData() {
        // 阻塞调用外部API，虚拟线程自动处理
        return CompletableFuture.completedFuture(restTemplate.getForObject("...", String.class));
    }
}

jcmd <pid> Thread.dump_to_file -format=json threads.json

Thread.getAllStackTraces().keySet().stream()
    .filter(t -> t.isVirtual())
    .forEach(t -> System.out.println(t.getName() + " carrier: " + t.getCarrierThread()));