从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()));

// 这段代码隐藏了哪些逻辑？旁人完全看不懂
public void process(User u, int a) {
    if (u.getLevel() > 5 && a > 0) {
        int b = u.getScore() + a * 10;
        u.setScore(b);
        // 写日志
        log("score changed");
        // 假设这里还要调用外部奖励接口...
        if (b > 10000) {
            // 发送系统通知...
        }
    }
}

public void handleScoreUpgrade(User user, int winStreak) {
    // 只有高级玩家且连胜才能触发倍数奖励
    if (user.getLevel() > LEVEL_THRESHOLD && winStreak > 0) {
        int bonusMultiplier = 10;
        int newScore = calculateBonusScore(user.getScore(), winStreak, bonusMultiplier);
        user.setScore(newScore);
        
        logScoreChange(user.getId(), newScore);
        // 如果达到传奇分数，触发额外庆祝逻辑
        if (newScore >= LEGENDARY_SCORE_THRESHOLD) {
            triggerLegendaryCelebration(user);
        }
    }
}

private int calculateBonusScore(int currentScore, int streak, int multiplier) {
    return currentScore + streak * multiplier;
}

public void handleScoreUpgrade(User user, int winStreak) {
    // 防御性校验：Fail-Fast 原则
    if (user == null) {
        throw new IllegalArgumentException("User cannot be null for score upgrade");
    }
    if (winStreak < 0) {
        // 业务异常，记录警告并正常返回，而非抛错中断业务
        log.warn("Negative winStreak detected for user: {}, ignoring.", user.getId());
        return;
    }
    
    if (user.getLevel() > LEVEL_THRESHOLD && winStreak > 0) {
        try {
            int newScore = calculateBonusScore(user.getScore(), winStreak, DEFAULT_BONUS_MULTIPLIER);
            // 使用AtomicReference或数据库乐观锁更新，这里仅作示意
            user.setScore(newScore);
            publishScoreChangedEvent(user); // 解耦后续日志和通知，改为事件驱动
        } catch (Exception e) {
            // 记录完整上下文，便于排查
            log.error("Failed to process score upgrade for user: {}, streak: {}", user.getId(), winStreak, e);
            // 根据业务决定是否重试或降级
        }
    }
}

// 利用虚拟线程执行器，无需改为异步回调，代码依然线性清晰
private CompletableFuture<Integer> fetchRemoteScoreRule(User user) {
    // 在虚拟线程环境下，这里的阻塞调用不会导致平台线程挂起
    return CompletableFuture.supplyAsync(() -> {
        // 模拟远程配置中心调用
        return remoteService.getScoreMultiplier(user.getLevel());
    }, virtualThreadExecutor);
}