Java虚拟线程生产级性能调优与监控实战指南

/**
 * 虚拟线程调度器核心参数
 * 
 * jdk.virtualThreadScheduler.maxPoolSize: 载体线程池最大大小
 *   - IO密集型：建议设为 CPU核心数 * 2 ~ 4
 *   - CPU密集型：保持默认，过大反而增加上下文切换
 * jdk.virtualThreadScheduler.parallelism: 并行度，默认等于CPU核心数
 * 
 * 生产环境通过JVM参数设置：
 * -XX:jdk.virtualThreadScheduler.maxPoolSize=32
 */
public class VtSchedulerConfig {
    public static void printStatus() {
        var fjp = ForkJoinPool.commonPool();
        System.out.printf("""
            调度器状态: 并行度=%d, 池大小=%d, 活跃=%d, 排队=%d%n
            """, fjp.getParallelism(), fjp.getPoolSize(),
            fjp.getActiveThreadCount(), fjp.getQueuedTaskCount());
    }
}

/**
 * 三种任务提交方式性能对比（10000个IO任务，8核CPU）
 * 
 * 1. ExecutorService（推荐）: 128ms, 12MB ⭐⭐⭐⭐⭐
 * 2. Thread.startVirtualThread: 356ms, 28MB ⭐⭐⭐
 * 3. ThreadFactory: 142ms, 15MB ⭐⭐⭐⭐
 */
public class VtSubmitBenchmark {
    /** ✅ 推荐：try-with-resources自动管理生命周期 */
    public static void testExecutor() throws Exception {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < 10000; i++) {
                executor.submit(() -> LockSupport.parkNanos(1_000_000));
            }
        } // 自动等待所有任务完成
    }
    /** ❌ 不推荐批量使用：每个任务都创建新对象 */
    public static void testStartVt() throws Exception {
        var latch = new CountDownLatch(10000);
        for (int i = 0; i < 10000; i++) {
            Thread.startVirtualThread(() -> {
                LockSupport.parkNanos(1_000_000);
                latch.countDown();
            });
        }
        latch.await();
    }
}

/**
 * ThreadLocal 内存泄漏与解决方案
 * 
 * ❌ 错误：大对象 + 虚拟线程 = 内存爆炸
 * private static final ThreadLocal<byte[]> BAD = 
 *     ThreadLocal.withInitial(() -> new byte[1024*1024]);
 */
public class ThreadLocalSolution {
    /** ✅ 方案一：ScopedValue（Java 21+ 推荐）*/
    private static final ScopedValue<UserContext> USER_CTX = ScopedValue.newInstance();
    public void process(String traceId, UserContext ctx) {
        ScopedValue.where(USER_CTX, ctx).run(() -> {
            // 业务逻辑中直接获取，自动回收，支持父子传递
            UserContext context = USER_CTX.get();
            doBusiness(context);
        });
    }
    /** ✅ 方案二：小对象 + 显式remove */
    private static final ThreadLocal<String> TRACE_ID = 
        ThreadLocal.withInitial(() -> "");
    public void safeUsage() {
        try {
            TRACE_ID.set(UUID.randomUUID().toString());
            // 业务逻辑...
        } finally {
            TRACE_ID.remove(); // 必须手动清理
        }
    }
}

/**
 * 载体线程Pinning问题与解决方案
 * 
 * 检测方式：-Djdk.tracePinnedThreads=full/short
 * JFR事件：jdk.VirtualThreadPinned
 */
public class PinningSolution {
    private final Object syncLock = new Object();
    private final ReentrantLock reentrantLock = new ReentrantLock();
    /** ❌ 会导致Pinning：synchronized内阻塞 */
    public void badSync() {
        synchronized (syncLock) {
            try {
                Thread.sleep(100); // 阻塞导致载体线程被占用
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    /** ✅ 推荐：ReentrantLock不会Pinning */
    public void goodLock() {
        reentrantLock.lock();
        try {
            Thread.sleep(100); // park/unpark机制不影响调度
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            reentrantLock.unlock();
        }
    }
}

/**
 * 虚拟线程并发限制方案
 */
public class VtConcurrencyLimiter {
    private final Semaphore limiter;
    public VtConcurrencyLimiter(int maxConcurrency) {
        this.limiter = new Semaphore(maxConcurrency);
    }
    /** 带限流的任务提交 */
    public <T> CompletableFuture<T> submit(Callable<T> task) {
        var future = new CompletableFuture<T>();
        Thread.startVirtualThread(() -> {
            try {
                limiter.acquire();
                try {
                    future.complete(task.call());
                } finally {
                    limiter.release();
                }
            } catch (Exception e) {
                future.completeExceptionally(e);
            }
        });
        return future;
    }
    /** Java 26 StructuredTaskScope 更优雅的方式 */
    public <T> List<T> batchExecute(List<Callable<T>> tasks, Duration timeout) 
            throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            var subtasks = tasks.stream().map(scope::fork).toList();
            scope.joinUntil(Instant.now().plus(timeout));
            scope.throwIfFailed();
            return subtasks.stream().map(Subtask::get).toList();
        }
    }
}

@Configuration
public class VtMetricsConfig {
    @Bean
    public MeterBinder virtualThreadMetrics() {
        return registry -> {
            // 活跃虚拟线程数
            Gauge.builder("vt.active.count",
                    () -> Thread.getAllStackTraces().keySet().stream()
                            .filter(Thread::isVirtual).count())
                 .description("活跃虚拟线程数")
                 .register(registry);
            // 载体线程池指标
            var fjp = ForkJoinPool.commonPool();
            Gauge.builder("carrier.pool.size", fjp, ForkJoinPool::getPoolSize)
                 .register(registry);
            Gauge.builder("carrier.active", fjp, ForkJoinPool::getActiveThreadCount)
                 .register(registry);
            Gauge.builder("carrier.queued", fjp, ForkJoinPool::getQueuedTaskCount)
                 .register(registry);
        };
    }
}