Java 与虚拟线程

ExecutorService pool = Executors.newFixedThreadPool(200);

// 方式一：直接启动（最顺手，适合一次性任务）
Thread vt = Thread.startVirtualThread(() -> {
    System.out.println("运行于: " + Thread.currentThread());
    System.out.println("是虚拟线程吗: " + Thread.currentThread().isVirtual()); // true
});

// 方式二：Builder，可命名（排查问题必备）
Thread named = Thread.ofVirtual().name("agg-worker-", 0).start(() -> doWork());

// 方式三：ThreadFactory，接老 API 用
ThreadFactory factory = Thread.ofVirtual().name("vt-", 0).factory();

// try-with-resources：close() 会等所有任务跑完，还能传播异常
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    IntStream.range(0, 10_000).forEach(i ->
        executor.submit(() -> {
            Thread.sleep(Duration.ofSeconds(1));   // 模拟 I/O 等待
            return "done-" + i;
        }));
}
// 1 万个"并发 sleep 1 秒"的任务，总耗时约 1 秒出点零头
// 换成 200 线程的平台池要跑 50 秒

import java.net.URI;
import java.net.http.*;
import java.time.Duration;
import java.util.List;

public class FetchDemo {
    private static final HttpClient CLIENT = HttpClient.newBuilder()
            .connectTimeout(Duration.ofSeconds(5)).build();

    public static void main(String[] args) throws Exception {
        List<String> urls = java.util.stream.IntStream.range(0, 100)
                .mapToObj(i -> "https://httpbin.org/delay/" + (i % 3 + 1))
                .toList();

        long start = System.currentTimeMillis();
        try (var executor = java.util.concurrent.Executors.newVirtualThreadPerTaskExecutor()) {
            var futures = urls.stream()
                    .map(url -> executor.submit(() -> fetch(url)))
                    .toList();
            for (var f : futures) f.get();
        }
        System.out.println("100 个并发请求耗时: " + (System.currentTimeMillis() - start) + "ms");
    }

    static String fetch(String url) throws Exception {
        HttpRequest req = HttpRequest.newBuilder(URI.create(url))
                .timeout(Duration.ofSeconds(10)).GET().build();
        // send() 是阻塞 API，但在虚拟线程里阻塞时会自动卸载，不占载体线程
        HttpResponse<String> resp = CLIENT.send(req, HttpResponse.BodyHandlers.ofString());
        return url + " -> " + resp.statusCode();
    }
}

// 反模式（JDK 21-23 上钉死载体线程，退化回平台线程性能）
public synchronized User getUser(Long id) {
    return userClient.fetch(id);        // 阻塞 I/O 在 synchronized 里
}

// 修复：需要"锁住临界区 + 里面有阻塞"时，一律 ReentrantLock
private final ReentrantLock lock = new ReentrantLock();

public User getUser(Long id) {
    lock.lock();                        // 等锁时虚拟线程可正常卸载
    try {
        return userClient.fetch(id);
    } finally {
        lock.unlock();
    }
}

# JDK 21-23：打印 Pinning 堆栈（JDK 24 起该属性已移除，monitor 类 Pinning 也不存在了）
java -Djdk.tracePinnedThreads=short -jar app.jar

# 所有版本通用：JFR 事件（生产推荐）
java -XX:StartFlightRecording=filename=vt.jfr -jar app.jar
jfr print --events jdk.VirtualThreadPinned vt.jfr

# 抓含虚拟线程的线程转储（JSON，便于程序分析）
jcmd <pid> Thread.dump_to_file -format=json dump.json

// 反模式：每请求缓存 1MB 上下文
private static final ThreadLocal<byte[]> CTX = ThreadLocal.withInitial(() -> new byte[1024 * 1024]);

// 百万虚拟线程 × 1MB = 1TB，必然 OOM；且"每任务一线程"模式下线程不复用，
// 经典的"线程池复用所以最后 remove"纪律也失去了着力点

private static final ScopedValue<String> USER_ID = ScopedValue.newInstance();

ScopedValue.where(USER_ID, "u-1001").run(() -> {
    handleRequest();                    // 作用域内随处可读 USER_ID.get()
    // 结构化并发 fork 出的子任务自动继承绑定
});

// 反模式：池化虚拟线程——既没必要也没好处
new ThreadPoolExecutor(100, 100, 0, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>(), Thread.ofVirtual().factory());

// 用信号量限流，而不是线程池
Semaphore dbPermits = new Semaphore(50);   // 最多 50 个并发查询打到 DB
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(() -> {
        dbPermits.acquire();
        try { return queryDb(); }
        finally { dbPermits.release(); }
    });
}

// 截至 JDK 26 仍是预览特性（第 6 次预览，JEP 525），运行需 --enable-preview
// JDK 25 起 API 为 open() + Joiner；JDK 21-24 的 ShutdownOnFailure 写法已废弃
public OrderDetail fetchOrderDetail(long orderId) throws InterruptedException {
    try (var scope = StructuredTaskScope.open()) {
        // 默认策略：任一子任务失败，其余自动取消，join() 抛 FailedException
        var order   = scope.fork(() -> fetchOrder(orderId));
        var user    = scope.fork(() -> fetchUser(orderId));
        var payment = scope.fork(() -> fetchPayment(orderId));
        scope.join();
        return new OrderDetail(order.get(), user.get(), payment.get());
    }
}

spring:
  threads:
    virtual:
      enabled: true

@RestController
public class AggregateController {
    private final RestClient restClient = RestClient.create();

    @GetMapping("/api/aggregate")
    public String aggregate() {
        // 三个串行阻塞调用——在虚拟线程下不占载体线程，吞吐随并发数扩展
        String user    = restClient.get().uri("/users/1").retrieve().bodyToMono(String.class).block();
        String order   = restClient.get().uri("/orders/1").retrieve().bodyToMono(String.class).block();
        String payment = restClient.get().uri("/payments/1").retrieve().bodyToMono(String.class).block();
        return user + "|" + order + "|" + payment;
    }
}