Java 21虚拟线程深度实战：百万并发连接的高性能Web服务器实现

引言：Java并发编程的革命性突破

Java 21正式引入了虚拟线程（Virtual Threads），这是Project Loom项目的成果，彻底改变了Java处理高并发的方式。本文将深入探讨如何利用虚拟线程构建支持百万级并发连接的高性能Web服务器，并提供完整的实战案例。

一、虚拟线程核心原理与优势

1.1 虚拟线程与传统线程的对比

// 传统平台线程 - 每个线程对应一个操作系统线程
ExecutorService executor = Executors.newFixedThreadPool(200); // 最多200个并发
for (int i = 0; i  {
        // I/O操作会阻塞操作系统线程
        String response = httpClient.send(request);
        processResponse(response);
    });
}
// 问题：线程数受限于操作系统，创建成本高

// 虚拟线程 - 轻量级用户态线程
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
for (int i = 0; i  {
        // I/O操作会挂起虚拟线程，但不会阻塞平台线程
        String response = httpClient.send(request);
        processResponse(response);
    });
}
// 优势：创建成本极低，支持海量并发

1.2 虚拟线程的调度机制

虚拟线程由JVM调度，运行在少量的载体线程（Carrier Threads）上：

public class VirtualThreadScheduler {
    public static void demonstrateScheduling() throws Exception {
        // 创建虚拟线程
        Thread virtualThread = Thread.ofVirtual()
            .name("virtual-thread-", 0)
            .unstarted(() -> {
                System.out.println("虚拟线程开始执行");
                
                // 模拟I/O操作 - 虚拟线程会被挂起
                try {
                    Thread.sleep(1000); // 挂起虚拟线程，释放载体线程
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                
                System.out.println("虚拟线程恢复执行");
            });
        
        // 启动虚拟线程
        virtualThread.start();
        virtualThread.join();
        
        // 查看线程类型
        System.out.println("是否为虚拟线程: " + virtualThread.isVirtual());
        System.out.println("线程ID: " + virtualThread.threadId());
    }
}

二、实战案例：构建高性能Web服务器

2.1 基于虚拟线程的HTTP服务器架构

// server/VirtualThreadHttpServer.java
import java.io.*;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;

public class VirtualThreadHttpServer {
    private final int port;
    private final ExecutorService virtualExecutor;
    private ServerSocket serverSocket;
    private volatile boolean running;
    
    public VirtualThreadHttpServer(int port) {
        this.port = port;
        // 使用虚拟线程执行器
        this.virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
    }
    
    public void start() throws IOException {
        serverSocket = new ServerSocket(port);
        serverSocket.setReuseAddress(true);
        running = true;
        
        System.out.println("服务器启动在端口: " + port);
        System.out.println("使用虚拟线程处理连接...");
        
        // 使用虚拟线程接受连接
        Thread.ofVirtual().start(() -> {
            while (running) {
                try {
                    Socket clientSocket = serverSocket.accept();
                    // 为每个连接创建虚拟线程处理
                    virtualExecutor.submit(() -> handleConnection(clientSocket));
                } catch (IOException e) {
                    if (running) {
                        System.err.println("接受连接失败: " + e.getMessage());
                    }
                }
            }
        });
    }
    
    private void handleConnection(Socket clientSocket) {
        String threadInfo = Thread.currentThread().toString();
        
        try (clientSocket;
             BufferedReader in = new BufferedReader(
                 new InputStreamReader(clientSocket.getInputStream()));
             OutputStream out = clientSocket.getOutputStream()) {
            
            // 解析HTTP请求
            String requestLine = in.readLine();
            if (requestLine == null) return;
            
            String[] requestParts = requestLine.split(" ");
            if (requestParts.length < 2) return;
            
            String method = requestParts[0];
            String path = requestParts[1];
            
            // 读取请求头
            StringBuilder headers = new StringBuilder();
            String headerLine;
            while ((headerLine = in.readLine()) != null && !headerLine.isEmpty()) {
                headers.append(headerLine).append("n");
            }
            
            // 处理请求
            String response = processRequest(method, path, headers.toString());
            
            // 发送响应
            out.write(response.getBytes(StandardCharsets.UTF_8));
            out.flush();
            
            System.out.println("处理请求: " + method + " " + path + 
                             " [线程: " + threadInfo + "]");
            
        } catch (IOException e) {
            System.err.println("处理连接失败: " + e.getMessage());
        }
    }
    
    private String processRequest(String method, String path, String headers) {
        // 模拟业务处理
        try {
            // 模拟I/O操作 - 虚拟线程在这里会被挂起
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        String body = """
            {
                "status": "success",
                "message": "请求处理完成",
                "path": "%s",
                "method": "%s",
                "thread": "%s",
                "timestamp": %d
            }
            """.formatted(path, method, Thread.currentThread(), System.currentTimeMillis());
        
        return """
            HTTP/1.1 200 OKr
            Content-Type: application/jsonr
            Content-Length: %dr
            Connection: closer
            r
            %s""".formatted(body.length(), body);
    }
    
    public void stop() {
        running = false;
        try {
            if (serverSocket != null && !serverSocket.isClosed()) {
                serverSocket.close();
            }
        } catch (IOException e) {
            System.err.println("关闭服务器失败: " + e.getMessage());
        }
        
        virtualExecutor.shutdown();
        try {
            if (!virtualExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                virtualExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            virtualExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        System.out.println("服务器已停止");
    }
    
    public static void main(String[] args) throws IOException {
        VirtualThreadHttpServer server = new VirtualThreadHttpServer(8080);
        server.start();
        
        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
    }
}

2.2 性能测试客户端

// client/ConcurrentLoadTester.java
import java.net.http.*;
import java.net.URI;
import java.time.Duration;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class ConcurrentLoadTester {
    private final HttpClient httpClient;
    private final String targetUrl;
    private final AtomicInteger successCount = new AtomicInteger();
    private final AtomicInteger failureCount = new AtomicInteger();
    private final AtomicLong totalResponseTime = new AtomicLong();
    
    public ConcurrentLoadTester(String targetUrl) {
        this.targetUrl = targetUrl;
        this.httpClient = HttpClient.newBuilder()
            .connectTimeout(Duration.ofSeconds(10))
            .version(HttpClient.Version.HTTP_1_1)
            .build();
    }
    
    public TestResult runTest(int concurrentUsers, int requestsPerUser, Duration testDuration) 
            throws InterruptedException {
        
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        CountDownLatch startLatch = new CountDownLatch(1);
        CountDownLatch completionLatch = new CountDownLatch(concurrentUsers);
        
        // 重置计数器
        successCount.set(0);
        failureCount.set(0);
        totalResponseTime.set(0);
        
        // 创建虚拟线程执行测试
        for (int i = 0; i  {
                try {
                    startLatch.await(); // 等待所有线程就绪
                    
                    for (int j = 0; j < requestsPerUser; j++) {
                        long startTime = System.nanoTime();
                        
                        try {
                            HttpRequest request = HttpRequest.newBuilder()
                                .uri(URI.create(targetUrl + "?user=" + userId + "&request=" + j))
                                .GET()
                                .timeout(Duration.ofSeconds(5))
                                .build();
                            
                            HttpResponse response = httpClient.send(
                                request, HttpResponse.BodyHandlers.ofString());
                            
                            if (response.statusCode() == 200) {
                                successCount.incrementAndGet();
                            } else {
                                failureCount.incrementAndGet();
                            }
                            
                        } catch (Exception e) {
                            failureCount.incrementAndGet();
                        }
                        
                        long endTime = System.nanoTime();
                        totalResponseTime.addAndGet(endTime - startTime);
                        
                        // 控制请求速率
                        Thread.sleep(10);
                    }
                    
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    completionLatch.countDown();
                }
            });
        }
        
        // 开始测试
        System.out.println("开始压力测试: " + concurrentUsers + " 并发用户");
        startLatch.countDown();
        
        // 等待测试完成或超时
        boolean completed = completionLatch.await(
            testDuration.toMillis(), TimeUnit.MILLISECONDS);
        
        executor.shutdown();
        
        return new TestResult(
            successCount.get(),
            failureCount.get(),
            totalResponseTime.get(),
            completed
        );
    }
    
    public static record TestResult(
        int successCount,
        int failureCount,
        long totalResponseTimeNanos,
        boolean completed
    ) {
        public double getSuccessRate() {
            int total = successCount + failureCount;
            return total > 0 ? (double) successCount / total * 100 : 0;
        }
        
        public double getAverageResponseTimeMillis() {
            int totalRequests = successCount + failureCount;
            return totalRequests > 0 ? 
                (totalResponseTimeNanos / 1_000_000.0) / totalRequests : 0;
        }
        
        public int getRequestsPerSecond() {
            long totalTimeSeconds = totalResponseTimeNanos / 1_000_000_000;
            int totalRequests = successCount + failureCount;
            return totalTimeSeconds > 0 ? totalRequests / (int)totalTimeSeconds : 0;
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        ConcurrentLoadTester tester = new ConcurrentLoadTester("http://localhost:8080/api/test");
        
        // 测试不同并发级别
        int[] concurrencyLevels = {100, 1000, 10000, 50000};
        
        for (int concurrency : concurrencyLevels) {
            System.out.println("n=== 测试并发级别: " + concurrency + " ===");
            
            TestResult result = tester.runTest(
                concurrency, 
                100, // 每个用户100个请求
                Duration.ofSeconds(30)
            );
            
            System.out.printf("成功请求: %,d%n", result.successCount());
            System.out.printf("失败请求: %,d%n", result.failureCount());
            System.out.printf("成功率: %.2f%%%n", result.getSuccessRate());
            System.out.printf("平均响应时间: %.2f ms%n", result.getAverageResponseTimeMillis());
            System.out.printf("QPS: %,d%n", result.getRequestsPerSecond());
            System.out.println("测试完成: " + (result.completed() ? "是" : "超时"));
        }
    }
}

三、高级特性：结构化并发

3.1 使用StructuredTaskScope管理并发任务

// concurrent/StructuredConcurrencyExample.java
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

public class StructuredConcurrencyExample {
    
    public record UserData(String userInfo, String orders, String payments) {}
    
    public UserData fetchUserDataConcurrently(String userId) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            // 并行获取用户数据
            Future userInfoFuture = scope.fork(() -> fetchUserInfo(userId));
            Future ordersFuture = scope.fork(() -> fetchUserOrders(userId));
            Future paymentsFuture = scope.fork(() -> fetchUserPayments(userId));
            
            // 等待所有任务完成或失败
            scope.join();
            scope.throwIfFailed(); // 如果有任务失败，抛出异常
            
            // 获取结果
            return new UserData(
                userInfoFuture.resultNow(),
                ordersFuture.resultNow(),
                paymentsFuture.resultNow()
            );
        }
    }
    
    public UserData fetchUserDataWithTimeout(String userId, Duration timeout) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Future userInfoFuture = scope.fork(() -> fetchUserInfo(userId));
            Future ordersFuture = scope.fork(() -> fetchUserOrders(userId));
            
            // 设置超时
            scope.joinUntil(Instant.now().plus(timeout));
            
            if (userInfoFuture.state() == Future.State.SUCCESS && 
                ordersFuture.state() == Future.State.SUCCESS) {
                return new UserData(
                    userInfoFuture.resultNow(),
                    ordersFuture.resultNow(),
                    "超时未获取" // 支付信息超时
                );
            } else {
                throw new TimeoutException("获取用户数据超时");
            }
        }
    }
    
    public String fetchFastestServiceResponse(List serviceUrls) throws Exception {
        try (var scope = new StructuredTaskScope()) {
            List<Future> futures = new ArrayList();
            
            // 并行调用所有服务
            for (String url : serviceUrls) {
                futures.add(scope.fork(() -> callService(url)));
            }
            
            // 等待第一个成功响应
            scope.join();
            
            for (Future future : futures) {
                if (future.state() == Future.State.SUCCESS) {
                    return future.resultNow();
                }
            }
            
            throw new RuntimeException("所有服务调用都失败了");
        }
    }
    
    private String fetchUserInfo(String userId) {
        // 模拟数据库查询
        try {
            Thread.sleep(100);
            return "用户信息: " + userId;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
    
    private String fetchUserOrders(String userId) {
        // 模拟API调用
        try {
            Thread.sleep(150);
            return "订单列表: 10个订单";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
    
    private String fetchUserPayments(String userId) {
        // 模拟外部服务调用
        try {
            Thread.sleep(200);
            return "支付记录: 5条记录";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
    
    private String callService(String url) {
        // 模拟服务调用
        try {
            Thread.sleep(50 + new Random().nextInt(100));
            return "响应来自: " + url;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
}

四、虚拟线程池优化策略

4.1 自定义虚拟线程工厂

// pool/AdvancedVirtualThreadPool.java
import java.lang.Thread.Builder.OfVirtual;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class AdvancedVirtualThreadPool {
    
    public static ExecutorService createOptimizedVirtualThreadPool() {
        ThreadFactory virtualThreadFactory = createVirtualThreadFactory();
        
        return new ThreadPoolExecutor(
            0, // 核心线程数
            Integer.MAX_VALUE, // 最大线程数
            60L, TimeUnit.SECONDS, // 空闲线程存活时间
            new SynchronousQueue(), // 直接传递队列
            virtualThreadFactory,
            new ThreadPoolExecutor.AbortPolicy()
        ) {
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                // 监控线程开始执行
                System.debug("虚拟线程开始执行任务: " + t.getName());
            }
            
            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                // 监控线程执行完成
                if (t != null) {
                    System.debug("虚拟线程执行失败: " + t.getMessage());
                }
            }
        };
    }
    
    private static ThreadFactory createVirtualThreadFactory() {
        AtomicInteger counter = new AtomicInteger(0);
        
        return r -> {
            OfVirtual builder = Thread.ofVirtual();
            
            // 配置虚拟线程属性
            return builder
                .name("virtual-worker-", counter.incrementAndGet())
                .inheritInheritableThreadLocals(false) // 不继承线程局部变量
                .unstarted(r);
        };
    }
    
    public static ExecutorService createVirtualThreadPoolWithMetrics() {
        ThreadFactory factory = createVirtualThreadFactory();
        ExecutorService executor = Executors.newThreadPerTaskExecutor(factory);
        
        // 添加监控
        return new MonitoringExecutorService(executor);
    }
    
    static class MonitoringExecutorService implements ExecutorService {
        private final ExecutorService delegate;
        private final AtomicInteger activeTasks = new AtomicInteger();
        private final AtomicLong completedTasks = new AtomicLong();
        private final AtomicLong failedTasks = new AtomicLong();
        
        public MonitoringExecutorService(ExecutorService delegate) {
            this.delegate = delegate;
        }
        
        @Override
        public Future submit(Runnable task) {
            activeTasks.incrementAndGet();
            return delegate.submit(() -> {
                try {
                    task.run();
                    completedTasks.incrementAndGet();
                } catch (Exception e) {
                    failedTasks.incrementAndGet();
                    throw e;
                } finally {
                    activeTasks.decrementAndGet();
                }
            });
        }
        
        public Metrics getMetrics() {
            return new Metrics(
                activeTasks.get(),
                completedTasks.get(),
                failedTasks.get()
            );
        }
        
        // 其他ExecutorService方法的委托实现...
        @Override
        public void execute(Runnable command) {
            delegate.execute(command);
        }
        
        @Override
        public void shutdown() {
            delegate.shutdown();
        }
        
        @Override
        public List shutdownNow() {
            return delegate.shutdownNow();
        }
        
        @Override
        public boolean isShutdown() {
            return delegate.isShutdown();
        }
        
        @Override
        public boolean isTerminated() {
            return delegate.isTerminated();
        }
        
        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit) 
                throws InterruptedException {
            return delegate.awaitTermination(timeout, unit);
        }
        
        // 其他submit和invoke方法的实现...
    }
    
    public record Metrics(int activeTasks, long completedTasks, long failedTasks) {
        public long getTotalTasks() {
            return completedTasks + failedTasks;
        }
        
        public double getSuccessRate() {
            long total = getTotalTasks();
            return total > 0 ? (double) completedTasks / total * 100 : 0;
        }
    }
}

五、性能监控与调试

5.1 虚拟线程监控工具

// monitor/VirtualThreadMonitor.java
import java.lang.management.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class VirtualThreadMonitor {
    private final ThreadMXBean threadMXBean;
    private final ScheduledExecutorService scheduler;
    private final Map threadStats;
    
    public VirtualThreadMonitor() {
        this.threadMXBean = ManagementFactory.getThreadMXBean();
        this.scheduler = Executors.newSingleThreadScheduledExecutor();
        this.threadStats = new ConcurrentHashMap();
    }
    
    public void startMonitoring() {
        // 每5秒收集一次指标
        scheduler.scheduleAtFixedRate(() -> {
            try {
                collectMetrics();
            } catch (Exception e) {
                System.err.println("收集指标失败: " + e.getMessage());
            }
        }, 0, 5, TimeUnit.SECONDS);
    }
    
    private void collectMetrics() {
        long[] threadIds = threadMXBean.getAllThreadIds();
        
        int virtualThreadCount = 0;
        int platformThreadCount = 0;
        int runningThreads = 0;
        int blockedThreads = 0;
        int waitingThreads = 0;
        
        for (long threadId : threadIds) {
            ThreadInfo threadInfo = threadMXBean.getThreadInfo(threadId);
            if (threadInfo == null) continue;
            
            // 判断线程类型
            if (isVirtualThread(threadInfo)) {
                virtualThreadCount++;
            } else {
                platformThreadCount++;
            }
            
            // 统计线程状态
            switch (threadInfo.getThreadState()) {
                case RUNNABLE -> runningThreads++;
                case BLOCKED, WAITING, TIMED_WAITING -> {
                    if (isVirtualThread(threadInfo)) {
                        waitingThreads++; // 虚拟线程的挂起状态
                    } else {
                        blockedThreads++;
                    }
                }
            }
            
            // 记录线程名称统计
            String threadName = threadInfo.getThreadName();
            threadStats.computeIfAbsent(threadName, k -> new AtomicInteger())
                      .incrementAndGet();
        }
        
        // 输出监控信息
        System.out.println("n=== 虚拟线程监控报告 ===");
        System.out.printf("虚拟线程数: %,d%n", virtualThreadCount);
        System.out.printf("平台线程数: %,d%n", platformThreadCount);
        System.out.printf("运行中线程: %,d%n", runningThreads);
        System.out.printf("挂起虚拟线程: %,d%n", waitingThreads);
        System.out.printf("阻塞平台线程: %,d%n", blockedThreads);
        
        // 输出线程名称分布
        System.out.println("n线程名称分布:");
        threadStats.entrySet().stream()
            .sorted((a, b) -> Integer.compare(b.getValue().get(), a.getValue().get()))
            .limit(10)
            .forEach(entry -> 
                System.out.printf("  %s: %,d%n", entry.getKey(), entry.getValue().get())
            );
        
        // 清理统计数据
        threadStats.clear();
    }
    
    private boolean isVirtualThread(ThreadInfo threadInfo) {
        String threadName = threadInfo.getThreadName();
        return threadName != null && 
               (threadName.startsWith("VirtualThread") || 
                threadName.contains("/virtual"));
    }
    
    public void dumpThreadDetails() {
        System.out.println("n=== 线程详细信息 ===");
        
        Thread.getAllStackTraces().forEach((thread, stackTrace) -> {
            System.out.printf("n线程: %s (ID: %d)%n", 
                thread.getName(), thread.getId());
            System.out.printf("是否为虚拟线程: %s%n", thread.isVirtual());
            System.out.printf("状态: %s%n", thread.getState());
            System.out.printf("优先级: %d%n", thread.getPriority());
            
            if (stackTrace.length > 0) {
                System.out.println("堆栈跟踪:");
                for (StackTraceElement element : stackTrace) {
                    System.out.println("  " + element);
                }
            }
        });
    }
    
    public void stop() {
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(3, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
    
    public static void main(String[] args) throws Exception {
        VirtualThreadMonitor monitor = new VirtualThreadMonitor();
        monitor.startMonitoring();
        
        // 运行测试程序
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        for (int i = 0; i  {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        // 运行10分钟后停止
        Thread.sleep(600000);
        monitor.stop();
        executor.shutdown();
    }
}

六、实战项目：实时消息推送系统

// messaging/RealTimeMessagingSystem.java
import java.io.*;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class RealTimeMessagingSystem {
    private final int port;
    private final ExecutorService virtualExecutor;
    private final Map<String, List> roomSubscriptions;
    private final AtomicInteger connectionCounter;
    
    public RealTimeMessagingSystem(int port) {
        this.port = port;
        this.virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
        this.roomSubscriptions = new ConcurrentHashMap();
        this.connectionCounter = new AtomicInteger(0);
    }
    
    public void start() throws IOException {
        ServerSocket serverSocket = new ServerSocket(port);
        serverSocket.setReuseAddress(true);
        
        System.out.println("消息推送系统启动在端口: " + port);
        
        // 使用虚拟线程接受连接
        Thread.ofVirtual().start(() -> {
            while (true) {
                try {
                    Socket clientSocket = serverSocket.accept();
                    int connectionId = connectionCounter.incrementAndGet();
                    
                    virtualExecutor.submit(() -> 
                        handleClientConnection(connectionId, clientSocket)
                    );
                    
                } catch (IOException e) {
                    System.err.println("接受连接失败: " + e.getMessage());
                    break;
                }
            }
        });
    }
    
    private void handleClientConnection(int connectionId, Socket clientSocket) {
        String clientAddress = clientSocket.getInetAddress().getHostAddress();
        
        try (clientSocket;
             BufferedReader in = new BufferedReader(
                 new InputStreamReader(clientSocket.getInputStream()));
             PrintWriter out = new PrintWriter(
                 new OutputStreamWriter(clientSocket.getOutputStream()), true)) {
            
            out.println("欢迎连接到实时消息系统! 连接ID: " + connectionId);
            
            String clientId = "client-" + connectionId;
            List subscribedRooms = new ArrayList();
            
            String inputLine;
            while ((inputLine = in.readLine()) != null) {
                if (inputLine.startsWith("SUBSCRIBE ")) {
                    String room = inputLine.substring(10).trim();
                    subscribeToRoom(room, new ClientConnection(clientId, out));
                    subscribedRooms.add(room);
                    out.println("已订阅房间: " + room);
                    
                } else if (inputLine.startsWith("UNSUBSCRIBE ")) {
                    String room = inputLine.substring(12).trim();
                    unsubscribeFromRoom(room, clientId);
                    subscribedRooms.remove(room);
                    out.println("已取消订阅房间: " + room);
                    
                } else if (inputLine.startsWith("PUBLISH ")) {
                    String[] parts = inputLine.substring(8).split(" ", 2);
                    if (parts.length == 2) {
                        String room = parts[0];
                        String message = parts[1];
                        publishToRoom(room, clientId, message);
                        out.println("消息已发布到房间: " + room);
                    }
                    
                } else if (inputLine.equals("LIST_ROOMS")) {
                    out.println("可用房间: " + String.join(", ", roomSubscriptions.keySet()));
                    
                } else if (inputLine.equals("STATS")) {
                    out.println(getSystemStats());
                    
                } else if (inputLine.equals("QUIT")) {
                    break;
                    
                } else {
                    out.println("未知命令");
                }
            }
            
            // 清理订阅
            for (String room : subscribedRooms) {
                unsubscribeFromRoom(room, clientId);
            }
            
            System.out.println("客户端断开连接: " + clientId + " (" + clientAddress + ")");
            
        } catch (IOException e) {
            System.err.println("客户端连接错误: " + e.getMessage());
        }
    }
    
    private void subscribeToRoom(String room, ClientConnection connection) {
        roomSubscriptions.computeIfAbsent(room, k -> 
            Collections.synchronizedList(new ArrayList())
        ).add(connection);
    }
    
    private void unsubscribeFromRoom(String room, String clientId) {
        List connections = roomSubscriptions.get(room);
        if (connections != null) {
            connections.removeIf(conn -> conn.clientId().equals(clientId));
            if (connections.isEmpty()) {
                roomSubscriptions.remove(room);
            }
        }
    }
    
    private void publishToRoom(String room, String senderId, String message) {
        List connections = roomSubscriptions.get(room);
        if (connections != null) {
            String formattedMessage = String.format(
                "[%s] %s: %s", 
                new Date(), 
                senderId, 
                message
            );
            
            // 并行推送消息到所有订阅者
            connections.parallelStream().forEach(connection -> {
                try {
                    connection.out().println(formattedMessage);
                } catch (Exception e) {
                    System.err.println("推送消息失败: " + e.getMessage());
                }
            });
        }
    }
    
    private String getSystemStats() {
        int totalConnections = connectionCounter.get();
        int totalRooms = roomSubscriptions.size();
        int totalSubscriptions = roomSubscriptions.values().stream()
            .mapToInt(List::size)
            .sum();
        
        return String.format(
            "系统统计:n" +
            "总连接数: %dn" +
            "房间数量: %dn" +
            "总订阅数: %dn" +
            "虚拟线程数: 动态变化",
            totalConnections, totalRooms, totalSubscriptions
        );
    }
    
    record ClientConnection(String clientId, PrintWriter out) {}
    
    public static void main(String[] args) throws IOException {
        RealTimeMessagingSystem system = new RealTimeMessagingSystem(9090);
        system.start();
        
        System.out.println("实时消息推送系统已启动");
        System.out.println("支持命令: SUBSCRIBE, UNSUBSCRIBE, PUBLISH, LIST_ROOMS, STATS, QUIT");
    }
}

七、最佳实践与注意事项

• 避免线程局部变量：虚拟线程应避免使用ThreadLocal，使用ScopedValue替代
• 合理使用同步：虚拟线程中的同步操作会阻塞载体线程
• 监控线程创建：虽然虚拟线程创建成本低，但也要避免无限制创建
• I/O密集型优势：虚拟线程特别适合I/O密集型应用
• CPU密集型注意：CPU密集型任务仍需使用平台线程池
• 调试支持：使用JDK工具监控虚拟线程状态

八、性能对比数据