第8部分：Java 21虚拟线程（Virtual Threads）

核心目标

理解Project Loom带来的并发革命。

1. 虚拟线程的概念与设计目标

虚拟线程基本概念

虚拟线程（Virtual Threads）是Java 21引入的一种轻量级线程，也被称为"用户态线程"或"绿色线程"。它们由JVM管理，而不是操作系统，可以创建数百万个虚拟线程而不会消耗大量系统资源。

设计目标

public class VirtualThreadConcepts {
    public static void main(String[] args) {
        // 1. 高并发支持
        // 目标：支持数百万个并发线程
        demonstrateHighConcurrency();
        
        // 2. 简化并发编程
        // 目标：使用同步API编写异步代码
        demonstrateSimplifiedConcurrency();
        
        // 3. 提高资源利用率
        // 目标：减少线程创建和上下文切换开销
        demonstrateResourceEfficiency();
    }
    
    private static void demonstrateHighConcurrency() {
        // 创建大量虚拟线程
        for (int i = 0; i < 1_000_000; i++) {
            final int threadId = i;
            Thread.startVirtualThread(() -> {
                System.out.println("虚拟线程 " + threadId + " 执行中");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
    }
    
    private static void demonstrateSimplifiedConcurrency() {
        // 使用同步API，但实际是异步执行
        Thread.startVirtualThread(() -> {
            try {
                // 阻塞I/O操作，但不会阻塞平台线程
                Thread.sleep(1000);
                System.out.println("虚拟线程完成I/O操作");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
    
    private static void demonstrateResourceEfficiency() {
        // 虚拟线程消耗很少的内存（约几KB）
        // 而平台线程消耗约2MB内存
        System.out.println("虚拟线程内存消耗: ~几KB");
        System.out.println("平台线程内存消耗: ~2MB");
    }
}

虚拟线程 vs 平台线程

public class VirtualVsPlatformThreads {
    public static void main(String[] args) throws InterruptedException {
        // 平台线程示例
        demonstratePlatformThreads();
        
        // 虚拟线程示例
        demonstrateVirtualThreads();
    }
    
    private static void demonstratePlatformThreads() throws InterruptedException {
        System.out.println("=== 平台线程示例 ===");
        
        long start = System.currentTimeMillis();
        
        // 创建平台线程
        Thread platformThread = new Thread(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("平台线程执行完成");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        platformThread.start();
        platformThread.join();
        
        long time = System.currentTimeMillis() - start;
        System.out.println("平台线程耗时: " + time + "ms");
    }
    
    private static void demonstrateVirtualThreads() throws InterruptedException {
        System.out.println("=== 虚拟线程示例 ===");
        
        long start = System.currentTimeMillis();
        
        // 创建虚拟线程
        Thread virtualThread = Thread.startVirtualThread(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("虚拟线程执行完成");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        virtualThread.join();
        
        long time = System.currentTimeMillis() - start;
        System.out.println("虚拟线程耗时: " + time + "ms");
    }
}

2. 平台线程 vs 虚拟线程

详细对比

public class ThreadComparison {
    public static void main(String[] args) {
        // 1. 创建方式对比
        demonstrateCreationMethods();
        
        // 2. 资源消耗对比
        demonstrateResourceConsumption();
        
        // 3. 性能对比
        demonstratePerformance();
    }
    
    private static void demonstrateCreationMethods() {
        System.out.println("=== 创建方式对比 ===");
        
        // 平台线程创建
        Thread platformThread = new Thread(() -> {
            System.out.println("平台线程执行");
        });
        platformThread.start();
        
        // 虚拟线程创建方式1
        Thread virtualThread1 = Thread.ofVirtual().start(() -> {
            System.out.println("虚拟线程1执行");
        });
        
        // 虚拟线程创建方式2
        Thread virtualThread2 = Thread.startVirtualThread(() -> {
            System.out.println("虚拟线程2执行");
        });
        
        // 虚拟线程创建方式3
        Thread virtualThread3 = Thread.ofVirtual()
            .name("custom-virtual-thread")
            .start(() -> {
                System.out.println("虚拟线程3执行");
            });
    }
    
    private static void demonstrateResourceConsumption() {
        System.out.println("=== 资源消耗对比 ===");
        
        // 平台线程资源消耗
        System.out.println("平台线程:");
        System.out.println("  内存消耗: ~2MB");
        System.out.println("  创建开销: 高");
        System.out.println("  上下文切换: 昂贵");
        System.out.println("  最大数量: 数千个");
        
        // 虚拟线程资源消耗
        System.out.println("虚拟线程:");
        System.out.println("  内存消耗: ~几KB");
        System.out.println("  创建开销: 低");
        System.out.println("  上下文切换: 便宜");
        System.out.println("  最大数量: 数百万个");
    }
    
    private static void demonstratePerformance() {
        System.out.println("=== 性能对比 ===");
        
        // 测试平台线程性能
        long platformTime = testPlatformThreads();
        
        // 测试虚拟线程性能
        long virtualTime = testVirtualThreads();
        
        System.out.println("平台线程耗时: " + platformTime + "ms");
        System.out.println("虚拟线程耗时: " + virtualTime + "ms");
        System.out.println("性能提升: " + (double) platformTime / virtualTime + "x");
    }
    
    private static long testPlatformThreads() {
        long start = System.currentTimeMillis();
        
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            Thread thread = new Thread(() -> {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            threads.add(thread);
            thread.start();
        }
        
        for (Thread thread : threads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        
        return System.currentTimeMillis() - start;
    }
    
    private static long testVirtualThreads() {
        long start = System.currentTimeMillis();
        
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            threads.add(thread);
        }
        
        for (Thread thread : threads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        
        return System.currentTimeMillis() - start;
    }
}

线程模型对比

public class ThreadModelComparison {
    public static void main(String[] args) {
        // 1. 平台线程模型（1:1映射）
        demonstratePlatformThreadModel();
        
        // 2. 虚拟线程模型（M:N映射）
        demonstrateVirtualThreadModel();
    }
    
    private static void demonstratePlatformThreadModel() {
        System.out.println("=== 平台线程模型 ===");
        System.out.println("特点:");
        System.out.println("  1. 1:1映射（一个Java线程对应一个操作系统线程）");
        System.out.println("  2. 由操作系统调度");
        System.out.println("  3. 创建和销毁开销大");
        System.out.println("  4. 上下文切换成本高");
        System.out.println("  5. 数量受限于操作系统");
        
        // 平台线程示例
        Thread platformThread = new Thread(() -> {
            System.out.println("平台线程执行中");
        });
        platformThread.start();
    }
    
    private static void demonstrateVirtualThreadModel() {
        System.out.println("=== 虚拟线程模型 ===");
        System.out.println("特点:");
        System.out.println("  1. M:N映射（多个虚拟线程映射到少量平台线程）");
        System.out.println("  2. 由JVM调度");
        System.out.println("  3. 创建和销毁开销小");
        System.out.println("  4. 上下文切换成本低");
        System.out.println("  5. 数量可达数百万个");
        
        // 虚拟线程示例
        Thread virtualThread = Thread.startVirtualThread(() -> {
            System.out.println("虚拟线程执行中");
        });
    }
}

3. 创建方式

基本创建方式

public class VirtualThreadCreation {
    public static void main(String[] args) throws InterruptedException {
        // 方式1：使用Thread.startVirtualThread()
        demonstrateStartVirtualThread();
        
        // 方式2：使用Thread.ofVirtual()
        demonstrateOfVirtual();
        
        // 方式3：使用Executors.newVirtualThreadPerTaskExecutor()
        demonstrateVirtualThreadExecutor();
        
        // 方式4：使用ForkJoinPool
        demonstrateForkJoinPool();
    }
    
    private static void demonstrateStartVirtualThread() throws InterruptedException {
        System.out.println("=== 使用Thread.startVirtualThread() ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            System.out.println("虚拟线程执行中: " + Thread.currentThread().getName());
        });
        
        virtualThread.join();
    }
    
    private static void demonstrateOfVirtual() throws InterruptedException {
        System.out.println("=== 使用Thread.ofVirtual() ===");
        
        // 基本用法
        Thread virtualThread1 = Thread.ofVirtual()
            .start(() -> {
                System.out.println("虚拟线程1: " + Thread.currentThread().getName());
            });
        
        // 设置名称
        Thread virtualThread2 = Thread.ofVirtual()
            .name("custom-virtual-thread")
            .start(() -> {
                System.out.println("虚拟线程2: " + Thread.currentThread().getName());
            });
        
        // 设置未捕获异常处理器
        Thread virtualThread3 = Thread.ofVirtual()
            .name("virtual-thread-with-handler")
            .uncaughtExceptionHandler((thread, exception) -> {
                System.out.println("虚拟线程异常: " + exception.getMessage());
            })
            .start(() -> {
                throw new RuntimeException("测试异常");
            });
        
        virtualThread1.join();
        virtualThread2.join();
        virtualThread3.join();
    }
    
    private static void demonstrateVirtualThreadExecutor() throws InterruptedException {
        System.out.println("=== 使用Executors.newVirtualThreadPerTaskExecutor() ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            // 提交任务
            Future<String> future1 = executor.submit(() -> {
                Thread.sleep(1000);
                return "任务1完成";
            });
            
            Future<String> future2 = executor.submit(() -> {
                Thread.sleep(1500);
                return "任务2完成";
            });
            
            // 获取结果
            System.out.println(future1.get());
            System.out.println(future2.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static void demonstrateForkJoinPool() throws InterruptedException {
        System.out.println("=== 使用ForkJoinPool ===");
        
        // 创建虚拟线程池
        ForkJoinPool virtualPool = new ForkJoinPool(
            Runtime.getRuntime().availableProcessors(),
            ForkJoinPool.defaultForkJoinWorkerThreadFactory,
            null,
            true // 使用虚拟线程
        );
        
        try {
            // 提交任务
            Future<String> future = virtualPool.submit(() -> {
                Thread.sleep(1000);
                return "ForkJoinPool虚拟线程任务完成";
            });
            
            System.out.println(future.get());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            virtualPool.shutdown();
        }
    }
}

高级创建方式

public class AdvancedVirtualThreadCreation {
    public static void main(String[] args) throws InterruptedException {
        // 1. 自定义线程工厂
        demonstrateCustomThreadFactory();
        
        // 2. 批量创建虚拟线程
        demonstrateBatchCreation();
        
        // 3. 虚拟线程池管理
        demonstrateVirtualThreadPoolManagement();
    }
    
    private static void demonstrateCustomThreadFactory() throws InterruptedException {
        System.out.println("=== 自定义线程工厂 ===");
        
        ThreadFactory virtualThreadFactory = Thread.ofVirtual()
            .name("custom-virtual-", 0)
            .factory();
        
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            Thread thread = virtualThreadFactory.newThread(() -> {
                System.out.println("自定义虚拟线程: " + Thread.currentThread().getName());
            });
            threads.add(thread);
            thread.start();
        }
        
        for (Thread thread : threads) {
            thread.join();
        }
    }
    
    private static void demonstrateBatchCreation() throws InterruptedException {
        System.out.println("=== 批量创建虚拟线程 ===");
        
        int threadCount = 1000;
        List<Thread> threads = new ArrayList<>();
        
        long start = System.currentTimeMillis();
        
        for (int i = 0; i < threadCount; i++) {
            final int threadId = i;
            Thread thread = Thread.startVirtualThread(() -> {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            threads.add(thread);
        }
        
        for (Thread thread : threads) {
            thread.join();
        }
        
        long time = System.currentTimeMillis() - start;
        System.out.println("创建 " + threadCount + " 个虚拟线程耗时: " + time + "ms");
    }
    
    private static void demonstrateVirtualThreadPoolManagement() throws InterruptedException {
        System.out.println("=== 虚拟线程池管理 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            // 提交多个任务
            List<Future<String>> futures = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                final int taskId = i;
                Future<String> future = executor.submit(() -> {
                    Thread.sleep(1000);
                    return "任务 " + taskId + " 完成";
                });
                futures.add(future);
            }
            
            // 等待所有任务完成
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

4. 虚拟线程与阻塞I/O的关系

阻塞I/O的优势

public class VirtualThreadBlockingIO {
    public static void main(String[] args) throws InterruptedException {
        // 1. 阻塞I/O在虚拟线程中的表现
        demonstrateBlockingIO();
        
        // 2. 与传统异步I/O的对比
        demonstrateAsyncIOComparison();
        
        // 3. 实际应用场景
        demonstrateRealWorldUsage();
    }
    
    private static void demonstrateBlockingIO() throws InterruptedException {
        System.out.println("=== 阻塞I/O在虚拟线程中 ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            try {
                System.out.println("开始阻塞I/O操作");
                Thread.sleep(2000); // 模拟阻塞I/O
                System.out.println("阻塞I/O操作完成");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        virtualThread.join();
    }
    
    private static void demonstrateAsyncIOComparison() {
        System.out.println("=== 阻塞I/O vs 异步I/O ===");
        
        // 传统异步I/O（复杂）
        System.out.println("传统异步I/O:");
        System.out.println("  - 需要回调函数");
        System.out.println("  - 错误处理复杂");
        System.out.println("  - 代码可读性差");
        
        // 虚拟线程中的阻塞I/O（简单）
        System.out.println("虚拟线程中的阻塞I/O:");
        System.out.println("  - 使用同步API");
        System.out.println("  - 错误处理简单");
        System.out.println("  - 代码可读性好");
    }
    
    private static void demonstrateRealWorldUsage() throws InterruptedException {
        System.out.println("=== 实际应用场景 ===");
        
        // 模拟HTTP服务器
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < 5; i++) {
                final int requestId = i;
                executor.submit(() -> {
                    handleHttpRequest(requestId);
                });
            }
        }
    }
    
    private static void handleHttpRequest(int requestId) {
        try {
            System.out.println("处理请求 " + requestId);
            
            // 模拟数据库查询（阻塞I/O）
            Thread.sleep(1000);
            
            // 模拟外部API调用（阻塞I/O）
            Thread.sleep(500);
            
            System.out.println("请求 " + requestId + " 处理完成");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

I/O密集型应用

public class IOIntensiveApplication {
    public static void main(String[] args) throws InterruptedException {
        // 1. 文件处理
        demonstrateFileProcessing();
        
        // 2. 网络请求
        demonstrateNetworkRequests();
        
        // 3. 数据库操作
        demonstrateDatabaseOperations();
    }
    
    private static void demonstrateFileProcessing() throws InterruptedException {
        System.out.println("=== 文件处理 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 10; i++) {
                final int fileId = i;
                Future<String> future = executor.submit(() -> {
                    // 模拟文件读取
                    try {
                        Thread.sleep(1000);
                        return "文件 " + fileId + " 处理完成";
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return "文件 " + fileId + " 处理中断";
                    }
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static void demonstrateNetworkRequests() throws InterruptedException {
        System.out.println("=== 网络请求 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 5; i++) {
                final int requestId = i;
                Future<String> future = executor.submit(() -> {
                    // 模拟HTTP请求
                    try {
                        Thread.sleep(2000);
                        return "请求 " + requestId + " 响应完成";
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return "请求 " + requestId + " 超时";
                    }
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static void demonstrateDatabaseOperations() throws InterruptedException {
        System.out.println("=== 数据库操作 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 8; i++) {
                final int queryId = i;
                Future<String> future = executor.submit(() -> {
                    // 模拟数据库查询
                    try {
                        Thread.sleep(1500);
                        return "查询 " + queryId + " 执行完成";
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return "查询 " + queryId + " 执行中断";
                    }
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

5. 与同步API的兼容性

synchronized兼容性

public class SynchronizedCompatibility {
    private static final Object lock = new Object();
    private static int counter = 0;
    
    public static void main(String[] args) throws InterruptedException {
        // 1. 虚拟线程中的synchronized
        demonstrateSynchronizedInVirtualThreads();
        
        // 2. 性能影响
        demonstratePerformanceImpact();
        
        // 3. 最佳实践
        demonstrateBestPractices();
    }
    
    private static void demonstrateSynchronizedInVirtualThreads() throws InterruptedException {
        System.out.println("=== 虚拟线程中的synchronized ===");
        
        List<Thread> threads = new ArrayList<>();
        
        for (int i = 0; i < 10; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                synchronized (lock) {
                    counter++;
                    System.out.println("虚拟线程 " + Thread.currentThread().getName() + 
                                     " 增加计数器到: " + counter);
                }
            });
            threads.add(thread);
        }
        
        for (Thread thread : threads) {
            thread.join();
        }
        
        System.out.println("最终计数器值: " + counter);
    }
    
    private static void demonstratePerformanceImpact() throws InterruptedException {
        System.out.println("=== synchronized性能影响 ===");
        
        // 测试无synchronized的虚拟线程
        long start = System.currentTimeMillis();
        List<Thread> threads1 = new ArrayList<>();
        
        for (int i = 0; i < 1000; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                // 无synchronized操作
                int temp = counter;
            });
            threads1.add(thread);
        }
        
        for (Thread thread : threads1) {
            thread.join();
        }
        
        long time1 = System.currentTimeMillis() - start;
        System.out.println("无synchronized耗时: " + time1 + "ms");
        
        // 测试有synchronized的虚拟线程
        start = System.currentTimeMillis();
        List<Thread> threads2 = new ArrayList<>();
        
        for (int i = 0; i < 1000; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                synchronized (lock) {
                    // synchronized操作
                    int temp = counter;
                }
            });
            threads2.add(thread);
        }
        
        for (Thread thread : threads2) {
            thread.join();
        }
        
        long time2 = System.currentTimeMillis() - start;
        System.out.println("有synchronized耗时: " + time2 + "ms");
        System.out.println("性能影响: " + (double) time2 / time1 + "x");
    }
    
    private static void demonstrateBestPractices() {
        System.out.println("=== 最佳实践 ===");
        
        System.out.println("1. 避免在虚拟线程中频繁使用synchronized");
        System.out.println("2. 优先使用无锁数据结构");
        System.out.println("3. 考虑使用ReentrantLock");
        System.out.println("4. 使用原子类替代synchronized");
        
        // 示例：使用原子类替代synchronized
        AtomicInteger atomicCounter = new AtomicInteger(0);
        
        Thread.startVirtualThread(() -> {
            atomicCounter.incrementAndGet();
            System.out.println("原子计数器: " + atomicCounter.get());
        });
    }
}

Lock兼容性

import java.util.concurrent.locks.ReentrantLock;

public class LockCompatibility {
    private static final ReentrantLock lock = new ReentrantLock();
    private static int counter = 0;
    
    public static void main(String[] args) throws InterruptedException {
        // 1. 虚拟线程中的Lock
        demonstrateLockInVirtualThreads();
        
        // 2. 性能对比
        demonstratePerformanceComparison();
        
        // 3. 异常处理
        demonstrateExceptionHandling();
    }
    
    private static void demonstrateLockInVirtualThreads() throws InterruptedException {
        System.out.println("=== 虚拟线程中的Lock ===");
        
        List<Thread> threads = new ArrayList<>();
        
        for (int i = 0; i < 10; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                lock.lock();
                try {
                    counter++;
                    System.out.println("虚拟线程 " + Thread.currentThread().getName() + 
                                     " 增加计数器到: " + counter);
                } finally {
                    lock.unlock();
                }
            });
            threads.add(thread);
        }
        
        for (Thread thread : threads) {
            thread.join();
        }
        
        System.out.println("最终计数器值: " + counter);
    }
    
    private static void demonstratePerformanceComparison() throws InterruptedException {
        System.out.println("=== 性能对比 ===");
        
        // 测试synchronized
        long start = System.currentTimeMillis();
        List<Thread> syncThreads = new ArrayList<>();
        
        for (int i = 0; i < 1000; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                synchronized (LockCompatibility.class) {
                    int temp = counter;
                }
            });
            syncThreads.add(thread);
        }
        
        for (Thread thread : syncThreads) {
            thread.join();
        }
        
        long syncTime = System.currentTimeMillis() - start;
        System.out.println("synchronized耗时: " + syncTime + "ms");
        
        // 测试ReentrantLock
        start = System.currentTimeMillis();
        List<Thread> lockThreads = new ArrayList<>();
        
        for (int i = 0; i < 1000; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                lock.lock();
                try {
                    int temp = counter;
                } finally {
                    lock.unlock();
                }
            });
            lockThreads.add(thread);
        }
        
        for (Thread thread : lockThreads) {
            thread.join();
        }
        
        long lockTime = System.currentTimeMillis() - start;
        System.out.println("ReentrantLock耗时: " + lockTime + "ms");
        System.out.println("性能差异: " + (double) lockTime / syncTime + "x");
    }
    
    private static void demonstrateExceptionHandling() throws InterruptedException {
        System.out.println("=== 异常处理 ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            lock.lock();
            try {
                System.out.println("获取锁成功");
                throw new RuntimeException("测试异常");
            } finally {
                lock.unlock();
                System.out.println("释放锁");
            }
        });
        
        virtualThread.join();
    }
}

6. 虚拟线程的调度、生命周期与调试

调度机制

public class VirtualThreadScheduling {
    public static void main(String[] args) throws InterruptedException {
        // 1. 调度器类型
        demonstrateSchedulerTypes();
        
        // 2. 调度策略
        demonstrateSchedulingStrategies();
        
        // 3. 调度监控
        demonstrateSchedulingMonitoring();
    }
    
    private static void demonstrateSchedulerTypes() {
        System.out.println("=== 调度器类型 ===");
        
        System.out.println("1. 默认调度器:");
        System.out.println("   - 使用ForkJoinPool");
        System.out.println("   - 工作窃取算法");
        System.out.println("   - 适合CPU密集型任务");
        
        System.out.println("2. 自定义调度器:");
        System.out.println("   - 可以实现自定义调度策略");
        System.out.println("   - 适合特殊场景");
    }
    
    private static void demonstrateSchedulingStrategies() throws InterruptedException {
        System.out.println("=== 调度策略 ===");
        
        // 创建大量虚拟线程
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            final int threadId = i;
            Thread thread = Thread.startVirtualThread(() -> {
                System.out.println("虚拟线程 " + threadId + " 开始执行");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("虚拟线程 " + threadId + " 执行完成");
            });
            threads.add(thread);
        }
        
        // 等待所有线程完成
        for (Thread thread : threads) {
            thread.join();
        }
    }
    
    private static void demonstrateSchedulingMonitoring() {
        System.out.println("=== 调度监控 ===");
        
        // 监控虚拟线程状态
        Thread virtualThread = Thread.startVirtualThread(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // 检查线程状态
        System.out.println("线程状态: " + virtualThread.getState());
        System.out.println("是否虚拟线程: " + virtualThread.isVirtual());
        System.out.println("线程名称: " + virtualThread.getName());
    }
}

生命周期管理

public class VirtualThreadLifecycle {
    public static void main(String[] args) throws InterruptedException {
        // 1. 生命周期阶段
        demonstrateLifecycleStages();
        
        // 2. 状态转换
        demonstrateStateTransitions();
        
        // 3. 生命周期监控
        demonstrateLifecycleMonitoring();
    }
    
    private static void demonstrateLifecycleStages() throws InterruptedException {
        System.out.println("=== 生命周期阶段 ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            System.out.println("1. 线程创建并启动");
            try {
                System.out.println("2. 线程运行中");
                Thread.sleep(1000);
                System.out.println("3. 线程即将结束");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("4. 线程被中断");
            }
        });
        
        virtualThread.join();
        System.out.println("5. 线程已终止");
    }
    
    private static void demonstrateStateTransitions() throws InterruptedException {
        System.out.println("=== 状态转换 ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            try {
                System.out.println("状态: " + Thread.currentThread().getState());
                Thread.sleep(1000);
                System.out.println("状态: " + Thread.currentThread().getState());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        System.out.println("启动后状态: " + virtualThread.getState());
        virtualThread.join();
        System.out.println("结束后状态: " + virtualThread.getState());
    }
    
    private static void demonstrateLifecycleMonitoring() throws InterruptedException {
        System.out.println("=== 生命周期监控 ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // 监控线程状态
        while (virtualThread.isAlive()) {
            System.out.println("线程状态: " + virtualThread.getState());
            Thread.sleep(500);
        }
        
        System.out.println("线程已结束");
    }
}

调试技巧

public class VirtualThreadDebugging {
    public static void main(String[] args) throws InterruptedException {
        // 1. 线程信息获取
        demonstrateThreadInfo();
        
        // 2. 异常处理
        demonstrateExceptionHandling();
        
        // 3. 性能分析
        demonstratePerformanceAnalysis();
    }
    
    private static void demonstrateThreadInfo() {
        System.out.println("=== 线程信息获取 ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            System.out.println("虚拟线程执行中");
        });
        
        // 获取线程信息
        System.out.println("线程名称: " + virtualThread.getName());
        System.out.println("线程ID: " + virtualThread.getId());
        System.out.println("是否虚拟线程: " + virtualThread.isVirtual());
        System.out.println("线程状态: " + virtualThread.getState());
        System.out.println("线程优先级: " + virtualThread.getPriority());
        System.out.println("是否守护线程: " + virtualThread.isDaemon());
    }
    
    private static void demonstrateExceptionHandling() throws InterruptedException {
        System.out.println("=== 异常处理 ===");
        
        Thread virtualThread = Thread.startVirtualThread(() -> {
            try {
                throw new RuntimeException("虚拟线程异常");
            } catch (Exception e) {
                System.out.println("捕获异常: " + e.getMessage());
            }
        });
        
        virtualThread.join();
    }
    
    private static void demonstratePerformanceAnalysis() throws InterruptedException {
        System.out.println("=== 性能分析 ===");
        
        long start = System.currentTimeMillis();
        
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            threads.add(thread);
        }
        
        for (Thread thread : threads) {
            thread.join();
        }
        
        long time = System.currentTimeMillis() - start;
        System.out.println("创建1000个虚拟线程耗时: " + time + "ms");
    }
}

7. 应用场景：高并发I/O密集型系统

HTTP服务器示例

public class VirtualThreadHttpServer {
    public static void main(String[] args) throws InterruptedException {
        // 1. 简单HTTP服务器
        demonstrateSimpleHttpServer();
        
        // 2. 高并发处理
        demonstrateHighConcurrency();
        
        // 3. 性能对比
        demonstratePerformanceComparison();
    }
    
    private static void demonstrateSimpleHttpServer() throws InterruptedException {
        System.out.println("=== 简单HTTP服务器 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            // 模拟处理HTTP请求
            for (int i = 0; i < 10; i++) {
                final int requestId = i;
                executor.submit(() -> {
                    handleHttpRequest(requestId);
                });
            }
        }
    }
    
    private static void handleHttpRequest(int requestId) {
        try {
            System.out.println("处理请求 " + requestId);
            
            // 模拟I/O操作
            Thread.sleep(1000);
            
            System.out.println("请求 " + requestId + " 处理完成");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    private static void demonstrateHighConcurrency() throws InterruptedException {
        System.out.println("=== 高并发处理 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            // 创建大量并发请求
            for (int i = 0; i < 1000; i++) {
                final int requestId = i;
                Future<String> future = executor.submit(() -> {
                    return processRequest(requestId);
                });
                futures.add(future);
            }
            
            // 等待所有请求完成
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static String processRequest(int requestId) {
        try {
            // 模拟复杂的I/O操作
            Thread.sleep(100);
            return "请求 " + requestId + " 处理完成";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "请求 " + requestId + " 处理中断";
        }
    }
    
    private static void demonstratePerformanceComparison() throws InterruptedException {
        System.out.println("=== 性能对比 ===");
        
        // 测试平台线程
        long platformTime = testPlatformThreads();
        
        // 测试虚拟线程
        long virtualTime = testVirtualThreads();
        
        System.out.println("平台线程耗时: " + platformTime + "ms");
        System.out.println("虚拟线程耗时: " + virtualTime + "ms");
        System.out.println("性能提升: " + (double) platformTime / virtualTime + "x");
    }
    
    private static long testPlatformThreads() throws InterruptedException {
        long start = System.currentTimeMillis();
        
        try (ExecutorService executor = Executors.newFixedThreadPool(100)) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 1000; i++) {
                final int requestId = i;
                Future<String> future = executor.submit(() -> {
                    try {
                        Thread.sleep(100);
                        return "平台线程请求 " + requestId + " 完成";
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return "平台线程请求 " + requestId + " 中断";
                    }
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                future.get();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        return System.currentTimeMillis() - start;
    }
    
    private static long testVirtualThreads() throws InterruptedException {
        long start = System.currentTimeMillis();
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 1000; i++) {
                final int requestId = i;
                Future<String> future = executor.submit(() -> {
                    try {
                        Thread.sleep(100);
                        return "虚拟线程请求 " + requestId + " 完成";
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return "虚拟线程请求 " + requestId + " 中断";
                    }
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                future.get();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        return System.currentTimeMillis() - start;
    }
}

数据库连接池示例

public class VirtualThreadDatabasePool {
    public static void main(String[] args) throws InterruptedException {
        // 1. 数据库连接池
        demonstrateDatabaseConnectionPool();
        
        // 2. 事务处理
        demonstrateTransactionProcessing();
        
        // 3. 批量操作
        demonstrateBatchOperations();
    }
    
    private static void demonstrateDatabaseConnectionPool() throws InterruptedException {
        System.out.println("=== 数据库连接池 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 100; i++) {
                final int queryId = i;
                Future<String> future = executor.submit(() -> {
                    return executeDatabaseQuery(queryId);
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static String executeDatabaseQuery(int queryId) {
        try {
            // 模拟数据库查询
            Thread.sleep(500);
            return "查询 " + queryId + " 执行完成";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "查询 " + queryId + " 执行中断";
        }
    }
    
    private static void demonstrateTransactionProcessing() throws InterruptedException {
        System.out.println("=== 事务处理 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 50; i++) {
                final int transactionId = i;
                Future<String> future = executor.submit(() -> {
                    return processTransaction(transactionId);
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static String processTransaction(int transactionId) {
        try {
            // 模拟事务开始
            System.out.println("事务 " + transactionId + " 开始");
            
            // 模拟多个数据库操作
            Thread.sleep(200);
            
            // 模拟事务提交
            System.out.println("事务 " + transactionId + " 提交");
            
            return "事务 " + transactionId + " 处理完成";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "事务 " + transactionId + " 处理中断";
        }
    }
    
    private static void demonstrateBatchOperations() throws InterruptedException {
        System.out.println("=== 批量操作 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 20; i++) {
                final int batchId = i;
                Future<String> future = executor.submit(() -> {
                    return processBatch(batchId);
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static String processBatch(int batchId) {
        try {
            // 模拟批量处理
            System.out.println("批量操作 " + batchId + " 开始");
            
            for (int i = 0; i < 10; i++) {
                Thread.sleep(50);
            }
            
            System.out.println("批量操作 " + batchId + " 完成");
            
            return "批量操作 " + batchId + " 处理完成";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "批量操作 " + batchId + " 处理中断";
        }
    }
}

实践练习

练习1：虚拟线程性能测试

public class VirtualThreadPerformanceTest {
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建性能测试
        testCreationPerformance();
        
        // 2. 执行性能测试
        testExecutionPerformance();
        
        // 3. 内存使用测试
        testMemoryUsage();
    }
    
    private static void testCreationPerformance() throws InterruptedException {
        System.out.println("=== 创建性能测试 ===");
        
        int threadCount = 10000;
        
        // 测试平台线程创建
        long start = System.currentTimeMillis();
        List<Thread> platformThreads = new ArrayList<>();
        
        for (int i = 0; i < threadCount; i++) {
            Thread thread = new Thread(() -> {
                // 空任务
            });
            platformThreads.add(thread);
            thread.start();
        }
        
        for (Thread thread : platformThreads) {
            thread.join();
        }
        
        long platformTime = System.currentTimeMillis() - start;
        System.out.println("平台线程创建耗时: " + platformTime + "ms");
        
        // 测试虚拟线程创建
        start = System.currentTimeMillis();
        List<Thread> virtualThreads = new ArrayList<>();
        
        for (int i = 0; i < threadCount; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                // 空任务
            });
            virtualThreads.add(thread);
        }
        
        for (Thread thread : virtualThreads) {
            thread.join();
        }
        
        long virtualTime = System.currentTimeMillis() - start;
        System.out.println("虚拟线程创建耗时: " + virtualTime + "ms");
        System.out.println("性能提升: " + (double) platformTime / virtualTime + "x");
    }
    
    private static void testExecutionPerformance() throws InterruptedException {
        System.out.println("=== 执行性能测试 ===");
        
        int taskCount = 1000;
        int taskDuration = 100; // 毫秒
        
        // 测试平台线程执行
        long start = System.currentTimeMillis();
        try (ExecutorService platformExecutor = Executors.newFixedThreadPool(100)) {
            List<Future<Void>> futures = new ArrayList<>();
            
            for (int i = 0; i < taskCount; i++) {
                Future<Void> future = platformExecutor.submit(() -> {
                    try {
                        Thread.sleep(taskDuration);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return null;
                });
                futures.add(future);
            }
            
            for (Future<Void> future : futures) {
                future.get();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        long platformTime = System.currentTimeMillis() - start;
        System.out.println("平台线程执行耗时: " + platformTime + "ms");
        
        // 测试虚拟线程执行
        start = System.currentTimeMillis();
        try (ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<Void>> futures = new ArrayList<>();
            
            for (int i = 0; i < taskCount; i++) {
                Future<Void> future = virtualExecutor.submit(() -> {
                    try {
                        Thread.sleep(taskDuration);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return null;
                });
                futures.add(future);
            }
            
            for (Future<Void> future : futures) {
                future.get();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        long virtualTime = System.currentTimeMillis() - start;
        System.out.println("虚拟线程执行耗时: " + virtualTime + "ms");
        System.out.println("性能提升: " + (double) platformTime / virtualTime + "x");
    }
    
    private static void testMemoryUsage() {
        System.out.println("=== 内存使用测试 ===");
        
        Runtime runtime = Runtime.getRuntime();
        
        // 测试平台线程内存使用
        long beforePlatform = runtime.totalMemory() - runtime.freeMemory();
        
        List<Thread> platformThreads = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            Thread thread = new Thread(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            platformThreads.add(thread);
            thread.start();
        }
        
        long afterPlatform = runtime.totalMemory() - runtime.freeMemory();
        long platformMemory = afterPlatform - beforePlatform;
        
        System.out.println("平台线程内存使用: " + platformMemory / 1024 / 1024 + "MB");
        
        // 测试虚拟线程内存使用
        long beforeVirtual = runtime.totalMemory() - runtime.freeMemory();
        
        List<Thread> virtualThreads = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            Thread thread = Thread.startVirtualThread(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            virtualThreads.add(thread);
        }
        
        long afterVirtual = runtime.totalMemory() - runtime.freeMemory();
        long virtualMemory = afterVirtual - beforeVirtual;
        
        System.out.println("虚拟线程内存使用: " + virtualMemory / 1024 / 1024 + "MB");
        System.out.println("内存节省: " + (double) platformMemory / virtualMemory + "x");
    }
}

练习2：虚拟线程Web应用

public class VirtualThreadWebApplication {
    public static void main(String[] args) throws InterruptedException {
        // 1. 模拟Web服务器
        demonstrateWebServer();
        
        // 2. 模拟API网关
        demonstrateApiGateway();
        
        // 3. 模拟微服务
        demonstrateMicroservices();
    }
    
    private static void demonstrateWebServer() throws InterruptedException {
        System.out.println("=== Web服务器 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            // 模拟处理HTTP请求
            for (int i = 0; i < 100; i++) {
                final int requestId = i;
                executor.submit(() -> {
                    handleHttpRequest(requestId);
                });
            }
        }
    }
    
    private static void handleHttpRequest(int requestId) {
        try {
            System.out.println("处理HTTP请求 " + requestId);
            
            // 模拟I/O操作
            Thread.sleep(100);
            
            System.out.println("HTTP请求 " + requestId + " 处理完成");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    private static void demonstrateApiGateway() throws InterruptedException {
        System.out.println("=== API网关 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 50; i++) {
                final int requestId = i;
                Future<String> future = executor.submit(() -> {
                    return processApiRequest(requestId);
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static String processApiRequest(int requestId) {
        try {
            System.out.println("处理API请求 " + requestId);
            
            // 模拟API调用
            Thread.sleep(200);
            
            return "API请求 " + requestId + " 处理完成";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "API请求 " + requestId + " 处理中断";
        }
    }
    
    private static void demonstrateMicroservices() throws InterruptedException {
        System.out.println("=== 微服务 ===");
        
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (int i = 0; i < 30; i++) {
                final int serviceId = i;
                Future<String> future = executor.submit(() -> {
                    return processMicroservice(serviceId);
                });
                futures.add(future);
            }
            
            for (Future<String> future : futures) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static String processMicroservice(int serviceId) {
        try {
            System.out.println("处理微服务 " + serviceId);
            
            // 模拟微服务处理
            Thread.sleep(300);
            
            return "微服务 " + serviceId + " 处理完成";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "微服务 " + serviceId + " 处理中断";
        }
    }
}

总结

本部分深入介绍了Java 21虚拟线程：

  1. 基本概念：理解虚拟线程的设计目标和优势
  2. 线程对比：掌握虚拟线程与平台线程的区别
  3. 创建方式：学习多种创建虚拟线程的方法
  4. I/O处理：了解虚拟线程在阻塞I/O中的优势
  5. API兼容性：掌握与同步API的兼容性
  6. 调度机制：理解虚拟线程的调度和生命周期
  7. 应用场景：学习在高并发I/O密集型系统中的应用

虚拟线程是Java并发编程的重要革新，为构建高性能、高并发的应用程序提供了新的解决方案。下一部分将学习性能优化、调试与并发设计模式。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

我测试了一个0.9B模型，它竟然能自动区分多人讲话）

avatar AI Agent技术社区
cover

从一张商品图到正式上架：实测店匠 Shoplazza Athena，AI Agent 不只是聊天框

avatar AI Agent技术社区
cover

【AI Agent】《2026年9月 AI Agent全栈开发技术选型专项面试宝典》

avatar AI Agent技术社区