Java虚拟线程深度实战：高并发系统性能优化与架构设计指南

方式一：使用Thread.ofVirtual()

// 创建并启动虚拟线程
Thread virtualThread = Thread.ofVirtual()
    .name("virtual-worker-", 0)
    .start(() -> {
        System.out.println("虚拟线程执行任务: " + Thread.currentThread());
        // 模拟IO操作
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });

// 等待虚拟线程完成
virtualThread.join();

方式二：使用Executors.newVirtualThreadPerTaskExecutor()

// 创建虚拟线程执行器
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future> futures = new ArrayList();
    
    // 提交10000个任务
    for (int i = 0; i < 10000; i++) {
        final int taskId = i;
        Future future = executor.submit(() -> {
            // 模拟业务处理
            processTask(taskId);
            return "任务-" + taskId + "-完成";
        });
        futures.add(future);
    }
    
    // 收集结果
    for (Future future : futures) {
        String result = future.get();
        System.out.println(result);
    }
}

方式三：使用ThreadFactory

// 创建虚拟线程工厂
ThreadFactory virtualThreadFactory = Thread.ofVirtual()
    .name("db-query-", 0)
    .factory();

// 使用工厂创建线程
Thread dbQueryThread = virtualThreadFactory.newThread(() -> {
    performDatabaseQuery();
});
dbQueryThread.start();

public class VirtualThreadDemo {
    
    // 模拟阻塞IO操作
    public static String fetchDataFromNetwork(String url) {
        try {
            // 当虚拟线程执行阻塞操作时，会自动挂起
            Thread.sleep(2000); // 模拟网络延迟
            return "从 " + url + " 获取的数据";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        }
    }
    
    public static void demonstrateMountUnmount() {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            
            List<CompletableFuture> futures = new ArrayList();
            
            for (int i = 0; i < 1000; i++) {
                final String url = "http://api.example.com/data/" + i;
                CompletableFuture future = CompletableFuture
                    .supplyAsync(() -> fetchDataFromNetwork(url), executor);
                futures.add(future);
            }
            
            // 所有虚拟线程会在阻塞操作时挂起，释放载体线程
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                .thenRun(() -> {
                    System.out.println("所有网络请求完成");
                    for (CompletableFuture future : futures) {
                        System.out.println("结果: " + future.join());
                    }
                })
                .join();
        }
    }
}

迁移前：传统线程池

// 传统固定大小线程池
ExecutorService traditionalExecutor = 
    Executors.newFixedThreadPool(200); // 最大200个平台线程

public void processUserRequests(List requests) {
    List<Future> futures = new ArrayList();
    
    for (Request request : requests) {
        Future future = traditionalExecutor.submit(() -> {
            // 每个请求占用一个平台线程
            return handleRequest(request); // 包含阻塞IO操作
        });
        futures.add(future);
    }
    
    // 处理结果...
    for (Future future : futures) {
        Response response = future.get();
        processResponse(response);
    }
}

迁移后：虚拟线程执行器

// 虚拟线程执行器 - 自动扩展
ExecutorService virtualThreadExecutor = 
    Executors.newVirtualThreadPerTaskExecutor();

public void processUserRequestsWithVirtualThreads(List requests) {
    List<CompletableFuture> futures = new ArrayList();
    
    for (Request request : requests) {
        CompletableFuture future = CompletableFuture
            .supplyAsync(() -> handleRequest(request), virtualThreadExecutor);
        futures.add(future);
    }
    
    // 使用CompletableFuture组合操作
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
        .thenApply(v -> futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList()))
        .thenAccept(this::processAllResponses)
        .join();
}

从CompletableFuture到结构化并发

// 传统异步编程 - 回调地狱风险
public CompletableFuture getUserProfileAsync(Long userId) {
    return CompletableFuture.supplyAsync(() -> getUserBasicInfo(userId))
        .thenCompose(basicInfo -> 
            CompletableFuture.supplyAsync(() -> getUserPreferences(userId))
                .thenApply(preferences -> 
                    new UserProfile(basicInfo, preferences)))
        .exceptionally(throwable -> {
            log.error("获取用户资料失败", throwable);
            return UserProfile.defaultProfile();
        });
}

// 使用虚拟线程的结构化并发 - 同步式编程体验
public UserProfile getUserProfileStructured(Long userId) {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        
        // 并发执行多个任务
        Subtask basicInfoTask = scope.fork(() -> 
            getUserBasicInfo(userId));
        Subtask preferencesTask = scope.fork(() -> 
            getUserPreferences(userId));
        
        // 等待所有任务完成或失败
        scope.join();
        scope.throwIfFailed();
        
        // 组合结果
        return new UserProfile(basicInfoTask.get(), preferencesTask.get());
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException("操作被中断", e);
    } catch (ExecutionException e) {
        log.error("获取用户资料失败", e);
        return UserProfile.defaultProfile();
    }
}

public class OrderProcessingService {
    
    public OrderResult processOrder(OrderRequest request) {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            
            // 并发执行订单验证、库存检查、价格计算
            Subtask validationTask = scope.fork(() -> 
                validateOrder(request));
            Subtask inventoryTask = scope.fork(() -> 
                checkInventory(request));
            Subtask priceTask = scope.fork(() -> 
                calculatePrice(request));
            
            // 等待所有必要任务完成
            scope.join();
            scope.throwIfFailed();
            
            // 检查业务规则
            if (!validationTask.get().isValid()) {
                throw new ValidationException("订单验证失败");
            }
            
            if (!inventoryTask.get().isInStock()) {
                throw new InventoryException("库存不足");
            }
            
            // 创建订单
            return createOrder(request, priceTask.get());
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("订单处理被中断", e);
        }
    }
    
    // 带超时的结构化并发
    public OrderResult processOrderWithTimeout(OrderRequest request, 
                                             Duration timeout) {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            
            Subtask validationTask = scope.fork(() -> 
                validateOrder(request));
            Subtask inventoryTask = scope.fork(() -> 
                checkInventory(request));
            
            // 带超时的等待
            scope.joinUntil(Instant.now().plus(timeout));
            scope.throwIfFailed();
            
            return createOrderBasedOnValidation(
                validationTask.get(), inventoryTask.get());
                
        } catch (TimeoutException e) {
            throw new BusinessException("订单处理超时", e);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("订单处理被中断", e);
        }
    }
}

@Service
public class ReactiveVirtualThreadService {
    
    private final Executor virtualThreadExecutor = 
        Executors.newVirtualThreadPerTaskExecutor();
    
    // 结合虚拟线程和反应式编程
    public Mono<List> searchProductsReactive(SearchCriteria criteria) {
        return Mono.fromCallable(() -> performSearch(criteria))
            .subscribeOn(Schedulers.fromExecutor(virtualThreadExecutor))
            .flatMapIterable(Function.identity())
            .filter(this::filterByAvailability)
            .collectList()
            .timeout(Duration.ofSeconds(10))
            .onErrorResume(throwable -> {
                log.error("搜索产品失败", throwable);
                return Mono.just(Collections.emptyList());
            });
    }
    
    // 批量处理优化
    public Flux batchProcessItems(List items) {
        return Flux.fromIterable(items)
            .parallel()
            .runOn(Schedulers.fromExecutor(virtualThreadExecutor))
            .flatMap(this::processItemAsync)
            .sequential()
            .buffer(100) // 每100个结果批量处理
            .flatMap(this::persistBatch);
    }
    
    private Mono processItemAsync(Item item) {
        return Mono.fromCallable(() -> processItem(item))
            .subscribeOn(Schedulers.fromExecutor(virtualThreadExecutor));
    }
}

public class ResourceUsageBenchmark {
    
    public static void benchmarkThreadCreation() {
        int threadCount = 100000;
        
        // 平台线程测试
        long startTime = System.currentTimeMillis();
        try {
            List platformThreads = new ArrayList();
            for (int i = 0; i  {
                    try { Thread.sleep(100); } catch (InterruptedException e) {}
                });
                platformThreads.add(thread);
                thread.start();
            }
            // 通常会因为资源不足而失败
        } catch (OutOfMemoryError e) {
            System.out.println("平台线程在创建 " + threadCount + " 个线程时内存不足");
        }
        
        // 虚拟线程测试
        long virtualStartTime = System.currentTimeMillis();
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future> futures = new ArrayList();
            for (int i = 0; i < threadCount; i++) {
                Future future = executor.submit(() -> {
                    try { Thread.sleep(100); } catch (InterruptedException e) {}
                });
                futures.add(future);
            }
            // 等待所有任务完成
            for (Future future : futures) {
                future.get();
            }
            long virtualEndTime = System.currentTimeMillis();
            System.out.println("虚拟线程完成 " + threadCount + " 个任务，耗时: " + 
                (virtualEndTime - virtualStartTime) + "ms");
        } catch (Exception e) {
            System.out.println("虚拟线程处理异常: " + e.getMessage());
        }
    }
}

重构前：基于线程池的订单处理系统

@Service
public class LegacyOrderService {
    private final ExecutorService orderExecutor = 
        Executors.newFixedThreadPool(500); // 最大500并发
    
    private final ExecutorService paymentExecutor = 
        Executors.newFixedThreadPool(200); // 支付处理线程池
    
    public CompletableFuture processOrder(Order order) {
        return CompletableFuture.supplyAsync(() -> validateOrder(order), orderExecutor)
            .thenComposeAsync(validation -> {
                if (!validation.isValid()) {
                    return CompletableFuture.completedFuture(
                        OrderResult.failed("验证失败"));
                }
                return processPaymentAsync(order);
            }, paymentExecutor)
            .exceptionally(throwable -> {
                log.error("订单处理失败", throwable);
                return OrderResult.failed("系统错误");
            });
    }
    
    // 系统瓶颈：线程池大小限制，内存消耗大，上下文切换开销高

重构后：基于虚拟线程的现代化架构

@Service
public class ModernOrderService {
    private final Executor virtualThreadExecutor = 
        Executors.newVirtualThreadPerTaskExecutor();
    
    public OrderResult processOrder(Order order) {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            
            // 并发执行所有验证步骤
            Subtask orderValidation = scope.fork(() -> 
                validateOrder(order));
            Subtask inventoryCheck = scope.fork(() -> 
                checkInventory(order));
            Subtask fraudDetection = scope.fork(() -> 
                detectFraud(order));
            
            // 等待所有验证完成
            scope.join();
            scope.throwIfFailed();
            
            // 检查验证结果
            if (!orderValidation.get().isValid()) {
                return OrderResult.failed("订单验证失败");
            }
            if (!inventoryCheck.get().isAvailable()) {
                return OrderResult.failed("库存不足");
            }
            if (fraudDetection.get().isSuspicious()) {
                return OrderResult.failed("欺诈检测失败");
            }
            
            // 处理支付
            PaymentResult paymentResult = processPayment(order);
            if (!paymentResult.isSuccess()) {
                return OrderResult.failed("支付失败: " + paymentResult.getMessage());
            }
            
            // 创建订单
            return createOrder(order, paymentResult);
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("订单处理被中断", e);
        } catch (ExecutionException e) {
            log.error("订单处理失败", e);
            return OrderResult.failed("系统错误");
        }
    }
    
    // 批量订单处理
    public List processOrdersBatch(List orders) {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            
            List<CompletableFuture> futures = orders.stream()
                .map(order -> CompletableFuture.supplyAsync(
                    () -> processOrder(order), executor))
                .collect(Collectors.toList());
            
            return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                .thenApply(v -> futures.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList()))
                .join();
        }
    }
}