跳到主要内容

虚拟线程（Virtual Threads）

虚拟线程是 Java 21 引入的革命性特性，它从根本上改变了 Java 并发编程的方式。虚拟线程是轻量级线程，可以大幅降低编写、维护和调试高吞吐量并发应用的难度。这是 Java 并发发展史上最重要的改进之一。

为什么需要虚拟线程？​

平台线程的局限性​

在 Java 21 之前，Java 只有一种线程——平台线程（Platform Thread）。平台线程是对操作系统线程的薄封装，每个平台线程都绑定一个操作系统线程。

平台线程存在以下问题：

资源消耗大

每个平台线程默认分配约 1MB 的栈空间，还要占用操作系统的其他资源。创建一万个平台线程就需要约 10GB 内存，这在实际应用中是不可接受的。

创建成本高

创建平台线程需要向操作系统申请资源，涉及系统调用，开销较大。频繁创建和销毁线程会带来显著的性能损耗。

数量受限

由于操作系统线程数量有限，平台线程的数量也受到限制。在高并发场景下，线程数量成为系统的瓶颈。

线程池的复杂性

为了解决线程创建成本高的问题，我们使用线程池来复用线程。但线程池引入了新的复杂性：如何确定线程池大小？如何处理任务队列？如何配置拒绝策略？

传统并发模型的困境​

为了应对平台线程的限制，开发者采用了各种异步编程模式：

// 异步回调模式
httpClient.sendAsync(request, BodyHandlers.ofString())
    .thenApply(response -> parseResponse(response))
    .thenAccept(data -> processData(data))
    .exceptionally(ex -> handleError(ex));

这种方式虽然解决了资源问题，但代码变得难以理解和维护。回调嵌套形成"回调地狱"，异常处理变得复杂，调试和追踪都很困难。

虚拟线程的解决方案​

虚拟线程的思想很简单：用大量的虚拟线程来模拟少量的操作系统线程，就像操作系统用虚拟内存来模拟有限的物理内存一样。

虚拟线程由 Java 运行时而非操作系统管理。当虚拟线程执行阻塞 I/O 操作时，Java 运行时会自动挂起该虚拟线程，释放其占用的载体线程，让载体线程去执行其他虚拟线程。当 I/O 操作完成时，虚拟线程会被恢复执行。

这意味着你可以创建数百万个虚拟线程，而不必担心资源耗尽。更重要的是，你可以用传统的同步阻塞风格编写代码，而不必使用复杂的异步模式。

虚拟线程与平台线程的对比​

特性平台线程虚拟线程
实现封装操作系统线程由 JVM 管理
栈空间约 1MB（固定）动态增长，初始很小
创建成本高（系统调用）低（普通对象）
数量限制受操作系统限制几乎无限制
调度方式操作系统调度JVM 调度
阻塞影响占用 OS 线程释放载体线程
适用场景CPU 密集型I/O 密集型

理解载体线程

虚拟线程并不直接运行在 CPU 上，它需要"挂载"到载体线程上才能执行。载体线程就是普通的平台线程。当虚拟线程执行阻塞操作时，它会从载体线程"卸载"，载体线程可以执行其他虚拟线程。

操作系统线程（载体线程）: Thread-1
    ├── 挂载虚拟线程 VT-1（执行到阻塞点）
    ├── 卸载 VT-1，挂载 VT-2
    ├── 卸载 VT-2，挂载 VT-3
    └── ...

虚拟线程队列：VT-1（等待 I/O）, VT-4, VT-5, VT-6...

创建虚拟线程​

使用 Thread 类创建​

方式一：直接创建并启动

Thread vThread = Thread.ofVirtual().start(() -> {
    System.out.println("Hello from virtual thread");
});

vThread.join();

方式二：先创建后启动

Thread.Builder builder = Thread.ofVirtual().name("my-virtual-thread");
Thread vThread = builder.unstarted(() -> {
    System.out.println("Hello from " + Thread.currentThread().getName());
});
vThread.start();
vThread.join();

方式三：创建多个命名虚拟线程

Thread.Builder builder = Thread.ofVirtual().name("worker-", 0);

Thread t1 = builder.start(() -> {
    System.out.println("Thread: " + Thread.currentThread().getName());
});

Thread t2 = builder.start(() -> {
    System.out.println("Thread: " + Thread.currentThread().getName());
});

t1.join();
t2.join();

// 输出:
// Thread: worker-0
// Thread: worker-1

使用 Executors 创建​

Java 21 在 Executors 类中新增了创建虚拟线程执行器的方法：

方式一：newVirtualThreadPerTaskExecutor

try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(() -> {
        System.out.println("Task 1 running in: " + Thread.currentThread());
        return "result1";
    });
    
    executor.submit(() -> {
        System.out.println("Task 2 running in: " + Thread.currentThread());
        return "result2";
    });
}

每次提交任务时，都会创建一个新的虚拟线程来执行。这种方式不需要线程池，因为虚拟线程本身足够轻量。

方式二：结合 CompletableFuture

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(
        () -> fetchData("url1"), executor);
    
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(
        () -> fetchData("url2"), executor);
    
    String result = future1.thenCombine(future2, (r1, r2) -> r1 + " + " + r2).join();
    System.out.println(result);
}

判断是否为虚拟线程​

Thread thread = Thread.currentThread();
boolean isVirtual = thread.isVirtual();

System.out.println("当前线程是否为虚拟线程: " + isVirtual);
System.out.println("线程名称: " + thread.getName());
System.out.println("线程ID: " + thread.threadId());

实战示例​

高并发 HTTP 服务端​

这是虚拟线程最典型的应用场景。传统方式下，处理大量并发连接需要使用 NIO 或异步框架，代码复杂。使用虚拟线程，可以用简单的阻塞 I/O 模型处理海量连接。

import java.io.*;
import java.net.*;
import java.util.concurrent.*;

public class VirtualThreadServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(8080);
        System.out.println("服务器启动，监听端口 8080");
        
        while (true) {
            Socket clientSocket = serverSocket.accept();
            
            // 为每个连接创建一个虚拟线程
            Thread.ofVirtual().start(() -> handleClient(clientSocket));
        }
    }
    
    private static void handleClient(Socket socket) {
        try (socket;
             BufferedReader in = new BufferedReader(
                 new InputStreamReader(socket.getInputStream()));
             PrintWriter out = new PrintWriter(socket.getOutputStream(), true)) {
            
            String request = in.readLine();
            System.out.println("收到请求: " + request + 
                ", 处理线程: " + Thread.currentThread());
            
            // 模拟耗时操作（数据库查询、远程调用等）
            Thread.sleep(100);
            
            out.println("HTTP/1.1 200 OK\r\n\r\nHello, World!");
            
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

使用虚拟线程，即使同时处理数万个连接，也不会耗尽资源。每个连接对应一个虚拟线程，代码逻辑清晰，易于理解和维护。

并行获取多个远程资源​

import java.net.URI;
import java.net.http.*;
import java.util.concurrent.*;
import java.util.*;

public class ParallelFetchDemo {
    private static final HttpClient client = HttpClient.newHttpClient();
    
    public static void main(String[] args) throws Exception {
        List<String> urls = List.of(
            "https://httpbin.org/delay/1",
            "https://httpbin.org/delay/2",
            "https://httpbin.org/delay/3",
            "https://httpbin.org/delay/1",
            "https://httpbin.org/delay/2"
        );
        
        long start = System.currentTimeMillis();
        
        // 使用虚拟线程并行获取
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futures = new ArrayList<>();
            
            for (String url : urls) {
                futures.add(executor.submit(() -> fetchUrl(url)));
            }
            
            for (Future<String> future : futures) {
                System.out.println("结果: " + future.get().substring(0, 50) + "...");
            }
        }
        
        long elapsed = System.currentTimeMillis() - start;
        System.out.println("总耗时: " + elapsed + "ms");
    }
    
    private static String fetchUrl(String url) throws Exception {
        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create(url))
            .GET()
            .build();
        
        HttpResponse<String> response = client.send(
            request, HttpResponse.BodyHandlers.ofString());
        
        System.out.println(Thread.currentThread().getName() + 
            " 获取: " + url + " 状态: " + response.statusCode());
        
        return response.body();
    }
}

这段代码会并行发送 5 个请求，总耗时约 3 秒（最长的那个请求），而不是串行执行的 9 秒。每个请求在一个独立的虚拟线程中执行，阻塞等待响应时不会占用操作系统线程。

批量数据库操作​

import java.sql.*;
import java.util.concurrent.*;
import java.util.*;

public class DatabaseBatchDemo {
    public static void main(String[] args) throws Exception {
        List<Integer> userIds = IntStream.range(1, 101).boxed().toList();
        
        long start = System.currentTimeMillis();
        
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<User>> futures = new ArrayList<>();
            
            for (Integer userId : userIds) {
                futures.add(executor.submit(() -> fetchUser(userId)));
            }
            
            List<User> users = new ArrayList<>();
            for (Future<User> future : futures) {
                users.add(future.get());
            }
            
            System.out.println("获取了 " + users.size() + " 个用户");
        }
        
        System.out.println("耗时: " + (System.currentTimeMillis() - start) + "ms");
    }
    
    private static User fetchUser(int id) throws SQLException {
        // 模拟数据库查询
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(
                 "SELECT * FROM users WHERE id = ?")) {
            
            stmt.setInt(1, id);
            ResultSet rs = stmt.executeQuery();
            
            if (rs.next()) {
                return new User(rs.getInt("id"), rs.getString("name"));
            }
            return null;
        }
    }
    
    private static Connection getConnection() throws SQLException {
        return DriverManager.getConnection("jdbc:mysql://localhost/mydb");
    }
    
    record User(int id, String name) {}
}

最佳实践​

为每个任务创建一个虚拟线程​

不要像使用平台线程那样复用虚拟线程。虚拟线程很轻量，应该为每个任务创建新的虚拟线程。

// 错误做法：池化虚拟线程
ExecutorService pool = Executors.newFixedThreadPool(100); // 不要这样做！

// 正确做法：每个任务一个虚拟线程
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(task1);
    executor.submit(task2);
}

为什么不应该池化虚拟线程？因为虚拟线程的设计初衷就是替代线程池模式。池化虚拟线程只会增加复杂性，而不会带来任何好处。

使用同步阻塞风格​

虚拟线程的存在就是为了让我们回归简单的同步阻塞编程模型。

// 推荐：同步阻塞风格
public User getUser(Long id) throws Exception {
    String json = httpClient.send(
        HttpRequest.newBuilder()
            .uri(URI.create("https://api.example.com/users/" + id))
            .build(),
        HttpResponse.BodyHandlers.ofString()
    ).body();
    
    return parseUser(json);
}

// 不推荐：异步回调风格（虚拟线程场景下没有优势）
public CompletableFuture<User> getUserAsync(Long id) {
    return httpClient.sendAsync(
        HttpRequest.newBuilder()
            .uri(URI.create("https://api.example.com/users/" + id))
            .build(),
        HttpResponse.BodyHandlers.ofString()
    ).thenApply(response -> parseUser(response.body()));
}

使用 Semaphore 限制并发​

如果需要限制对某个资源的并发访问，不要使用固定大小的线程池，而是使用 Semaphore。

import java.util.concurrent.*;

public class RateLimitedApi {
    // 限制同时最多 10 个请求
    private final Semaphore semaphore = new Semaphore(10);
    
    public String callExternalApi(String path) throws Exception {
        semaphore.acquire();
        try {
            return doApiCall(path);
        } finally {
            semaphore.release();
        }
    }
    
    private String doApiCall(String path) throws Exception {
        // 实际的 API 调用
        Thread.sleep(100); // 模拟网络延迟
        return "Response from " + path;
    }
}

避免在 ThreadLocal 中缓存昂贵对象​

传统的做法是在 ThreadLocal 中缓存线程不安全的昂贵对象，如 SimpleDateFormat。但对于虚拟线程，这种做法会适得其反，因为虚拟线程数量巨大，每个线程都会创建一份缓存对象。

// 错误做法：在 ThreadLocal 中缓存对象
static final ThreadLocal<SimpleDateFormat> cachedFormatter = 
    ThreadLocal.withInitial(SimpleDateFormat::new);

void process(Date date) {
    String formatted = cachedFormatter.get().format(date);
    // ...
}

// 正确做法：使用线程安全的替代品
static final DateTimeFormatter formatter = 
    DateTimeFormatter.ofPattern("yyyy-MM-dd");

void process(LocalDate date) {
    String formatted = formatter.format(date);
    // ...
}

Pinning 问题​

什么是 Pinning？​

虚拟线程在执行阻塞操作时，通常会从载体线程上卸载，释放载体线程去执行其他虚拟线程。但在某些情况下，虚拟线程会被"钉住"（Pinned），无法卸载，导致载体线程被阻塞。

Java 版本差异​

Java 21-23：虚拟线程在以下情况下会被钉住：

  1. 在 synchronized 块或方法内执行阻塞操作
  2. 调用本地方法（native method）或外部函数
  3. 在 Object.wait() 等待期间

Java 24+（JEP 491）：JVM 实现了"无 Pinning 的 synchronized"，彻底解决了 synchronized 导致的 pinning 问题。虚拟线程现在可以在 synchronized 块或方法内正常卸载，不再钉住载体线程。这是通过改进 JVM 对对象监视器（monitor）的实现来实现的。

JEP 491 的重要影响​

Java 24 引入的 JEP 491（Synchronize Virtual Threads without Pinning）是一个重大改进：

之前（Java 21-23）：

// 在 Java 21-23 中，这段代码会导致 pinning
synchronized byte[] getData() {
    byte[] buf = ...;
    int nread = socket.getInputStream().read(buf);  // 阻塞时会 pinning
    return buf;
}

当虚拟线程在 synchronized 块内执行阻塞 I/O 操作时，无法卸载，载体线程被阻塞，无法执行其他虚拟线程。

现在（Java 24+）：

// 在 Java 24+ 中，这段代码可以正常工作，不会 pinning
synchronized byte[] getData() {
    byte[] buf = ...;
    int nread = socket.getInputStream().read(buf);  // 阻塞时可以正常卸载
    return buf;
}

虚拟线程在阻塞时可以正常卸载，载体线程可以执行其他虚拟线程。这意味着大多数遗留代码无需修改即可享受虚拟线程的扩展性优势。

选择 synchronized 还是 ReentrantLock​

Java 24 之后，选择锁的标准变得更加简单：

根据《Java Concurrency in Practice》的建议：

  • 优先使用 synchronized：更简单、更不易出错，Java 24+ 中不会导致 pinning
  • 使用 ReentrantLock：当需要更多灵活性时（公平锁、可中断锁、超时获取、读写锁等）

无论使用哪种锁，都应遵循以下原则：

  • 缩小锁的范围
  • 尽量避免在持有锁时执行 I/O 或其他阻塞操作

剩余的 Pinning 场景​

Java 24+ 中，仍有少数情况会导致 pinning：

  1. 调用本地方法（native method）：通过 JNI 或 Foreign Function API 调用的本地代码
  2. 类初始化阻塞：等待另一个线程完成类初始化
  3. 类加载阻塞：在加载类时阻塞

这些场景在实际应用中较少见，通常不会造成问题。

检测 Pinning（Java 21-23）​

如果你使用的是 Java 21-23，可以通过 JDK Flight Recorder 检测 pinning：

# 启用 pinning 追踪（Java 21-23）
java -Djdk.tracePinnedThreads=full MyApplication

注意：Java 24+ 中，jdk.tracePinnedThreads 系统属性已被移除，因为 synchronized 不再导致 pinning。但 jdk.VirtualThreadPinned JFR 事件仍然存在，用于检测本地方法调用等剩余的 pinning 场景。

JDK Flight Recorder 监控​

# 启动时启用 JFR
java -XX:StartFlightRecording=filename=recording.jfr MyApplication

# 查看 pinning 事件
jfr print --events jdk.VirtualThreadPinned recording.jfr

jdk.VirtualThreadPinned 事件包含以下信息：

  • blockingOperation：导致阻塞的操作（如 "Native or VM frame on stack"）
  • pinnedReason：pinning 的原因
  • duration：持续时长
  • carrierThread：载体线程信息

调试虚拟线程​

使用 jcmd 查看线程转储​

# 文本格式
jcmd <pid> Thread.dump_to_file -format=text threads.txt

# JSON 格式
jcmd <pid> Thread.dump_to_file -format=json threads.json

JSON 格式的线程转储包含了所有虚拟线程的信息，便于分析。

使用 JDK Flight Recorder​

# 启动时启用 JFR
java -XX:StartFlightRecording=filename=recording.jfr MyApplication

# 打印虚拟线程相关事件
jfr print --events jdk.VirtualThreadStart,jdk.VirtualThreadEnd,jdk.VirtualThreadPinned recording.jfr

虚拟线程适用场景​

适合使用虚拟线程​

  • 高并发的 I/O 密集型应用（Web 服务器、API 网关）
  • 需要同时处理大量阻塞操作的场景
  • 希望用简单的同步代码替代复杂的异步代码
  • 任务执行时间短、数量多的场景

不适合使用虚拟线程​

  • CPU 密集型任务（虚拟线程不会提高计算速度）
  • 长时间运行的任务
  • 需要精细控制线程优先级的场景
  • 大量使用 synchronized 进行长时间阻塞的遗留代码（需先重构）

小结​

虚拟线程是 Java 并发编程的重大突破：

  1. 轻量级：可以创建数百万个虚拟线程，不受操作系统限制
  2. 简单：回归同步阻塞编程模型，代码更易读易维护
  3. 高效：阻塞 I/O 操作时自动释放载体线程，资源利用率高
  4. 兼容：与现有线程 API 完全兼容，迁移成本低

虚拟线程的目标是提供规模（scale，更高的吞吐量），而不是速度（speed，更低的延迟）。如果你的应用是 I/O 密集型，并且希望能够同时处理大量并发请求，虚拟线程是一个理想的选择。

在实际开发中，应该为每个并发任务创建一个虚拟线程，使用同步阻塞的编程风格，避免池化虚拟线程，注意 pinning 问题。随着虚拟线程的成熟，它将成为 Java 并发编程的标准方式。