// 改造前的Spring Boot配置
// application.properties
server.tomcat.threads.max=200
server.tomcat.threads.min-spare=10

// 改造前的业务代码
@Service
public class DataAggregationService {
    
    @Autowired
    private UserServiceClient userClient;
    @Autowired  
    private OrderServiceClient orderClient;
    @Autowired
    private ProductServiceClient productClient;
    
    // 串行调用，总耗时 = 所有下游服务耗时之和
    public AggregationResult aggregate(Long userId) {
        UserInfo user = userClient.getUser(userId);           // 50ms
        List<Order> orders = orderClient.getOrders(userId);   // 80ms
        List<Product> products = productClient.getRecommendations(userId); // 60ms
        return new AggregationResult(user, orders, products); // 总计190ms
    }
}

3.2 改造第一步：启用虚拟线程

// Spring Boot 3.2+ 最简单的方式
// application.properties
spring.threads.virtual.enabled=true

// 这一行配置让Tomcat使用虚拟线程处理HTTP请求
// 业务代码完全不需要改动！

// 如果要手动配置（更精细控制）
@Configuration
public class VirtualThreadConfig {
    
    @Bean
    public TomcatProtocolHandlerCustomizer<?> virtualThreadTomcatCustomizer() {
        return handler -> handler.setExecutor(
            Executors.newVirtualThreadPerTaskExecutor()
        );
    }
    
    // 异步任务也用虚拟线程
    @Bean
    public AsyncTaskExecutor applicationTaskExecutor() {
        return new TaskExecutorAdapter(
            Executors.newVirtualThreadPerTaskExecutor()
        );
    }
}

3.3 改造第二步：并行化阻塞调用

虚拟线程让我们可以用简单的同步代码实现并行调用，不需要CompletableFuture的复杂语法：

@Service
public class ImprovedAggregationService {
    
    @Autowired
    private UserServiceClient userClient;
    @Autowired
    private OrderServiceClient orderClient;
    @Autowired
    private ProductServiceClient productClient;
    
    /**
     * 用虚拟线程并行调用，代码清晰，耗时 = max(50, 80, 60) = 80ms
     */
    public AggregationResult aggregateParallel(Long userId) throws InterruptedException {
        // 用StructuredTaskScope（Java 21预览特性）
        // 或者用ExecutorService简单实现
        
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        
        Future<UserInfo> userFuture = executor.submit(() -> userClient.getUser(userId));
        Future<List<Order>> ordersFuture = executor.submit(() -> orderClient.getOrders(userId));
        Future<List<Product>> productsFuture = executor.submit(() -> productClient.getRecommendations(userId));
        
        try {
            UserInfo user = userFuture.get(2, TimeUnit.SECONDS);
            List<Order> orders = ordersFuture.get(2, TimeUnit.SECONDS);
            List<Product> products = productsFuture.get(2, TimeUnit.SECONDS);
            return new AggregationResult(user, orders, products);
        } catch (ExecutionException | TimeoutException e) {
            executor.shutdownNow();
            throw new ServiceException("聚合数据失败", e);
        } finally {
            executor.close();
        }
    }
}

3.4 处理Pinning问题：替换synchronized

// 问题：某些遗留代码用synchronized，会导致虚拟线程被固定
public class LegacyCache {
    private final Map<String, Object> cache = new HashMap<>();
    
    // 这个synchronized会导致Pinning！
    public synchronized Object get(String key) {
        // 如果这里有IO操作（比如缓存未命中查数据库），线程会被固定
        if (!cache.containsKey(key)) {
            Object value = loadFromDatabase(key); // IO操作！
            cache.put(key, value);
        }
        return cache.get(key);
    }
}

// 修复方案：将synchronized替换为ReentrantLock
public class ImprovedCache {
    private final Map<String, Object> cache = new HashMap<>();
    private final ReentrantLock lock = new ReentrantLock(); // 替换synchronized
    
    public Object get(String key) throws InterruptedException {
        lock.lock();
        try {
            if (!cache.containsKey(key)) {
                lock.unlock(); // 在IO前释放锁
                Object value = loadFromDatabase(key);
                lock.lock();   // IO后重新获取锁
                cache.put(key, value);
            }
            return cache.get(key);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

// 更简洁的方式：用ConcurrentHashMap的computeIfAbsent
// 注意：computeIfAbsent内部也有同步，但IO不在其内部
public class BetterCache {
    private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
    
    public Object get(String key) {
        return cache.computeIfAbsent(key, k -> loadFromDatabase(k));
        // 警告：如果loadFromDatabase很慢，computeIfAbsent会阻塞同key的其他请求
        // 更好的方案是用CompletableFuture缓存Mono，避免重复查询
    }
}

3.5 监控Pinning事件

// 开启Pinning监控
// JVM参数添加：
// -Djdk.tracePinnedThreads=full
// 发现Pinning时会打印详细栈信息

// 或者在代码中监控
@Configuration
public class VirtualThreadMonitor {
    
    @PostConstruct
    public void setupPinningMonitor() {
        // 监控虚拟线程事件（需要JFR支持）
        Thread.ofVirtual().name("pinning-monitor").start(() -> {
            // 用JFR监控VirtualThreadPinned事件
            // 实际实现需要引入JFR工具类
        });
    }
}

3.6 数据库连接池适配

// HikariCP对虚拟线程的配置调整
// 虚拟线程场景下，连接池的最大连接数应该和实际数据库处理能力对齐
// 不是越大越好！
@Configuration
public class DatabaseConfig {
    
    @Bean
    public HikariDataSource dataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://localhost:5432/mydb");
        config.setUsername("postgres");
        config.setPassword("secret");
        
        // 虚拟线程场景：连接池大小根据数据库能力设置，而不是线程数
        // PostgreSQL默认max_connections=100，保守设置
        config.setMaximumPoolSize(20);
        config.setMinimumIdle(5);
        
        // 重要：禁用connectionTimeout的指数退避，避免虚拟线程长时间等待
        config.setConnectionTimeout(10000); // 10秒超时
        config.setIdleTimeout(600000);
        
        return new HikariDataSource(config);
    }
}

四、工程实践与最佳实践

4.1 压测对比数据

我们项目的实际改造结果（同一个数据聚合服务）：

配置	QPS	P99延迟	线程数	内存(RSS)
传统200线程(串行)	1200	580ms	210	480MB
传统200线程(CompletableFuture并行)	3800	180ms	210	490MB
虚拟线程(串行调用)	6500	220ms	35	280MB
虚拟线程(并行调用)	8200	95ms	35	285MB

最让我惊喜的是内存从480MB降到280MB，线程数从210降到35。这对容器化部署的成本影响很大。

4.2 不适合虚拟线程的场景

CPU密集型：虚拟线程没有帮助，平台线程本来就不会阻塞
需要精细线程控制：比如给特定任务绑定CPU核（thread affinity），虚拟线程不支持
大量使用ThreadLocal的遗留代码：虚拟线程的ThreadLocal行为有微妙变化（ScopedValue是新方案）

// 问题场景
ThreadLocal<String> tenantId = new ThreadLocal<>();
tenantId.set("tenant-001");

// 提交到虚拟线程池的任务，看不到父线程的tenantId
executor.submit(() -> {
    System.out.println(tenantId.get()); // null！
});

// 解决方案：手动传递
String currentTenant = tenantId.get();
executor.submit(() -> {
    tenantId.set(currentTenant); // 手动设置
    try {
        // 业务逻辑
    } finally {
        tenantId.remove();
    }
});

// 或者用InheritableThreadLocal（有继承语义，但也有坑）
InheritableThreadLocal<String> inheritableTenantId = new InheritableThreadLocal<>();

坑三：第三方库的synchronized Pinning

我们用的一个老版本Redis客户端（Jedis某个版本），内部大量使用synchronized。在高并发场景下，Pinning导致平台线程被占满，反而不如改造前。

解决方案：升级到支持虚拟线程友好的新版Jedis，或者换用Lettuce（非阻塞Redis客户端）。

排查工具：

# 添加JVM参数，Pinning时会在日志输出
-Djdk.tracePinnedThreads=short

坑四：虚拟线程数量失控

有人觉得虚拟线程便宜就无限创建，结果创建了几十万个虚拟线程，虽然内存占用不大，但调度器的工作量也增加了，同时GC压力也变大。

合理做法：还是要限制并发度，用Semaphore或者有界ExecutorService。

// 用Semaphore限制并发
private final Semaphore concurrencyLimit = new Semaphore(1000);

public void processWithLimit(Runnable task) throws InterruptedException {
    concurrencyLimit.acquire();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        executor.submit(() -> {
            try {
                task.run();
            } finally {
                concurrencyLimit.release();
            }
        });
    }
}

六、总结与个人判断

虚拟线程是Java这十年来最重要的特性之一，没有之一。它让Java在IO密集型场景下，既能有接近响应式编程的并发效率，又能保持同步代码的简洁可读性。

但我要说清楚它的定位：虚拟线程不是响应式编程的替代品，而是一条不同的路。

对于大多数业务系统，用虚拟线程改造传统阻塞IO服务，是性价比最高的方案：改动小、风险低、效果好。

对于需要真正端到端非阻塞（连数据库驱动都是异步的）、需要背压控制、需要精细流控的场景，响应式编程依然不可替代。

我的实际建议：先升Java 21，开启虚拟线程，看能解决多少问题。剩下那部分解决不了的，再考虑响应式编程。从最简单的改动开始，别一上来就想重写系统。

Java 21 Virtual Threads实战：改造传统阻塞IO服务的完整案例

Java 21 Virtual Threads实战：改造传统阻塞IO服务的完整案例

开篇故事

一、核心问题：传统线程的开销在哪里

二、原理深度解析

2.1 虚拟线程调度原理

2.2 Pinning问题：虚拟线程的最大限制

三、完整改造案例

3.1 改造前：传统线程池配置