books

《Java 并发编程·线程与内存模型》14.3 虚拟线程下的池、ThreadLocal 与限流

虚拟线程把每条阻塞任务降到约 1 KB，但连接池、下游 QPS 和 ThreadLocal 副本数没变。本节实测 1 000 个任务不设限时 maxInFlight=1 000、36–57 ms 打穿下游；加 Semaphore(16) 后恒定 16、耗时 2 256–2 286 ms。另测 10 万条虚拟线程各存 1 KB ThreadLocal 多占约 104 MB。

Leeting Yan 7 分钟阅读 3448 字

承接的问题

14.1 把线程成本从约 1 MB 降到约 1 KB，14.2 说明 pinning 只剩 native 和类初始化两块。于是很多人得出一个推论：既然线程便宜，就把所有请求都丢给虚拟线程，并发自然上去了。

这个推论漏掉了后半句：虚拟线程只改变了「等待」的成本，没有改变下游的容量。数据库连接池还是 20 个连接，下游短信网关还是每秒 200 条，ThreadLocal 还是每条线程一份。并发数从 200 涨到 10 万，只是把排队从线程池挪到了连接池和网关。

这一节把三件事补齐：不要池化虚拟线程、ThreadLocal 在虚拟线程下的数量变化、以及用 Semaphore 把并发压回下游能承受的范围。

不要池化虚拟线程

平台线程池的存在理由是两个：创建线程贵，且线程数必须受控。虚拟线程把第一条消掉了，第二条则应该交给真正稀缺的资源去管。Executors.newVirtualThreadPerTaskExecutor() 返回的执行器每个任务开一条新虚拟线程，任务结束线程就结束：

try (ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor()) {
    for (var task : tasks) {
        ex.submit(task);
    }
}   // close() 等待所有任务结束

不要写成 new ThreadPoolExecutor(4, 4, 60, SECONDS, queue, Thread.ofVirtual().factory())。固定大小的虚拟线程池会把「M 条虚拟线程映射到 N 条载体」退化成「N 条虚拟线程抢 N 条载体」，阻塞时没有多余的虚拟线程可以顶上来，虚拟线程的唯一优势就没了。

做法结果评价
newVirtualThreadPerTaskExecutor()每任务一条，阻塞即卸载正确默认
固定大小虚拟线程池并发被池大小锁死反模式
平台线程池回到 1:1，扩展性受栈内存限制CPU 密集时仍可用
每请求 Thread.ofVirtual().start()等价于第一行，但少了结构化关闭可用，注意生命周期

实验：并发上限到底卡在哪

下面用 1 000 个任务打一个模拟下游，下游每次耗时 30 ms。第一轮不设限，第二轮用 Semaphore(16) 包住下游调用，记录同时在途的最大任务数。

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class SemaphoreCap {
    static final int TASKS = 1_000, PERMITS = 16;
    static final AtomicInteger inFlight = new AtomicInteger();
    static final AtomicInteger maxInFlight = new AtomicInteger();
    static final AtomicInteger completed = new AtomicInteger();

    static void downstream() {
        maxInFlight.accumulateAndGet(inFlight.incrementAndGet(), Math::max);
        try { Thread.sleep(30); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        finally { inFlight.decrementAndGet(); completed.incrementAndGet(); }
    }

    static long run(Semaphore permits) throws Exception {
        inFlight.set(0); maxInFlight.set(0); completed.set(0);
        CountDownLatch go = new CountDownLatch(1);
        long t0 = System.nanoTime();
        try (ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < TASKS; i++) {
                ex.submit(() -> {
                    try {
                        go.await();
                        if (permits == null) {
                            downstream();
                        } else {
                            permits.acquire();
                            try { downstream(); } finally { permits.release(); }
                        }
                    } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                });
            }
            go.countDown();
        }
        return (System.nanoTime() - t0) / 1_000_000;
    }

    public static void main(String[] args) throws Exception {
        System.out.println("java.version = " + System.getProperty("java.version")
            + " tasks=" + TASKS + " permits=" + PERMITS);
        for (int round = 1; round <= 3; round++) {
            long a = run(null);
            System.out.printf("round %d no-limit : maxInFlight=%d completed=%d elapsed=%dms%n",
                round, maxInFlight.get(), completed.get(), a);
            long b = run(new Semaphore(PERMITS));
            System.out.printf("round %d semaphore: maxInFlight=%d completed=%d elapsed=%dms%n",
                round, maxInFlight.get(), completed.get(), b);
        }
    }
}

本机输出（Apple M1 Pro，10 核，load averages 约 132–183）：

java.version = 25.0.5 tasks=1000 permits=16
round 1 no-limit : maxInFlight=1000 completed=1000 elapsed=57ms
round 1 semaphore: maxInFlight=16 completed=1000 elapsed=2256ms
round 2 no-limit : maxInFlight=1000 completed=1000 elapsed=36ms
round 2 semaphore: maxInFlight=16 completed=1000 elapsed=2258ms
round 3 no-limit : maxInFlight=1000 completed=1000 elapsed=42ms
round 3 semaphore: maxInFlight=16 completed=1000 elapsed=2258ms

两组数字的含义完全不同：

  • 不设限：1 000 条虚拟线程在 36–57 ms 内全部进入下游，maxInFlight=1 000。对模拟下游没事，对真实下游就是 1 000 个并发连接或 1 000 QPS，超过容量直接超时或被打限流。
  • Semaphore(16)：maxInFlight 三轮都精确等于 16，说明限流生效；总耗时 2 256–2 286 ms。理论值是 1000 / 16 × 30 ms = 1 875 ms，多出的约 380 ms 是任务创建、许可竞争和机器高负载的排队开销。

Semaphore.acquire() 在虚拟线程里是可卸载的阻塞，不占载体——14.2 的矩阵里它产生 0 个 jdk.VirtualThreadPinned 事件。所以「10 万条虚拟线程抢 16 个许可」不会把调度器拖死，只会让 99 984 条安静地等在堆里。

ThreadLocal 与 MDC

虚拟线程不池化，意味着线程本地变量不再有「复用」这回事：每条虚拟线程一份 ThreadLocalMap。数量从「池大小份」变成「并发任务数份」。

先用小程序确认隔离性和继承行为：

static final ThreadLocal<Object> TL = new ThreadLocal<>();
static final InheritableThreadLocal<String> ITL = new InheritableThreadLocal<>();
static final ScopedValue<String> REQ = ScopedValue.newInstance();

static void isolation() throws Exception {
    TL.set("main");
    String[] seen = new String[2];
    Thread a = Thread.ofVirtual().start(() -> {
        seen[0] = String.valueOf(TL.get());
        TL.set("vt-a");
        seen[0] = seen[0] + "->" + TL.get();
    });
    a.join();
    Thread b = Thread.ofVirtual().start(() -> seen[1] = String.valueOf(TL.get()));
    b.join();
    System.out.println("isolation: vt-a saw=" + seen[0]
        + " vt-b saw=" + seen[1] + " main saw=" + TL.get());
}

static void inheritance() throws Exception {
    ITL.set("itl-main");
    String[] itlSeen = new String[1];
    Thread c = Thread.ofVirtual().start(() -> itlSeen[0] = ITL.get());
    c.join();

    String[] scopedSeen = new String[1];
    ScopedValue.where(REQ, "req-42").run(() -> {
        try {
            Thread d = Thread.ofVirtual().start(() ->
                scopedSeen[0] = REQ.orElse("<unbound>"));
            d.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
    System.out.println("inheritance: InheritableThreadLocal=" + itlSeen[0]
        + " ScopedValue=" + scopedSeen[0] + " main-bound=" + REQ.isBound());
}

本机输出：

isolation: vt-a saw=null->vt-a vt-b saw=null main saw=main
inheritance: InheritableThreadLocal=itl-main ScopedValue=<unbound> main-bound=false

三个结论都和直觉有出入，值得逐条说：

  1. ThreadLocal 天然隔离。主线程设的值，子虚拟线程看不到；子线程设的值也不会回流。这是好事，但也意味着任何依赖「线程池预热 ThreadLocal」的优化都失效了。
  2. InheritableThreadLocal 会继承，在虚拟线程创建时复制一份。它能解决传递问题，但复制发生在 start() 时刻，之后父线程再改就不同步；而且每个任务一份副本，上下文对象大时内存会放大。
  3. ScopedValue 不会被 Thread.ofVirtual().start() 继承。JDK 25 的 ScopedValue 文档写得很清楚：绑定只由 StructuredTaskScope 创建时捕获，并由 fork() 的子任务继承。普通 Thread / Thread.Builder 启动的线程读不到父线程的绑定。要用它做请求上下文，子任务必须放在 StructuredTaskScope 里（JDK 25 仍是 preview，15.1 展开）。

那 MDC 怎么办？日志框架的 MDC 本质是 ThreadLocal<Map>，上面三条规则原样适用。可选方案：

方案传递方式代价
显式传 context 参数方法签名携带最可靠，侵入性大
InheritableThreadLocal创建时复制之后修改不同步，副本多
在任务内重新绑定 ScopedValue父线程捕获值，子任务 where(...).run(...)需要改任务包装层
StructuredTaskScope + ScopedValue自动继承preview API，15.1 展开

数量上也要算账：10 万条虚拟线程各存 1 KB 的上下文，就是 100 MB 量级的堆。本机实测（-Xmx512m，两轮）：

round 1 baseline heapDelta=123.3MB
round 1 +1KB TL heapDelta=228.0MB
round 2 baseline heapDelta=108.8MB
round 2 +1KB TL heapDelta=213.1MB

基线包含 10 万条阻塞线程的栈片段，约 109–123 MB；每条再挂 1 KB 数组后增加约 104 MB，正好对应 100 000 × 1 024 B。所以「虚拟线程很轻」的前提是不要给每条线程挂大对象。缓存、大 buffer、完整用户对象都不该进 ThreadLocal。

限流应该放在哪一层

Semaphore 放在虚拟线程里只是手段，真正要决定的是「许可数从哪来」。三条原则：

许可数对齐真实稀缺资源，而不是拍脑袋。 连接池 20 个连接，许可就给 20；下游文档写 200 QPS、单次 30 ms，按 200 × 0.03 = 6 个并发就够，给 16 是留余量。许可大于资源，等于没限。

每个下游一个信号量，不要全局一把。 三路通知里短信慢，不应该把邮件和推送一起拖住。按渠道拆分，慢渠道自己排队。

拿不到许可要快速失败或降级，不要无限等。 acquire() 会一直等，上游请求可能先超时。用 tryAcquire(timeout) 把等待时间纳入预算：

if (!permits.tryAcquire(200, TimeUnit.MILLISECONDS)) {
    throw new LoanException(ErrorCode.CONFLICT, "downstream saturated: " + channel);
}
try {
    client.send(reservation);
} finally {
    permits.release();
}

等待时间的账也要算：16 个许可、每个 30 ms，1 000 个任务的第 1 000 个要等约 (1000/16 - 1) × 30 ≈ 1.8 s。如果上游 SLA 是 500 ms，许可数就不该是 16，而应该让请求在排队前失败，或者把扇出改成异步补偿。

在 libraryd-conc 里落地

VirtualLoanService 给三个下游各配一个信号量，并把请求上下文做成 record 显式传进任务；任务内重新绑定 ScopedValue，因为普通虚拟线程不会继承它。

public final class VirtualLoanService implements LoanService, AutoCloseable {

    public record RequestContext(String requestId, String memberId) {}
    private static final ScopedValue<RequestContext> CTX = ScopedValue.newInstance();

    private final BookRepository books;
    private final LoanRepository loans;
    private final ExecutorService fanout = Executors.newVirtualThreadPerTaskExecutor();
    private final Map<String, Semaphore> permits = Map.of(
            "mail", new Semaphore(8),
            "sms", new Semaphore(16),
            "push", new Semaphore(16));
    private final Map<String, DownstreamClient> clients;

    public VirtualLoanService(BookRepository books, LoanRepository loans,
                              Map<String, DownstreamClient> clients) {
        this.books = books;
        this.loans = loans;
        this.clients = clients;
    }

    @Override
    public Reservation reserve(String isbn, String memberId) {
        Reservation r = reserveLocked(isbn, memberId);
        RequestContext ctx = new RequestContext("req-" + r.loanId(), memberId);
        for (String channel : List.of("mail", "sms", "push")) {
            fanout.submit(() -> ScopedValue.where(CTX, ctx).run(() -> send(channel, r)));
        }
        return r;
    }

    private void send(String channel, Reservation r) {
        Semaphore sem = permits.get(channel);
        try {
            if (!sem.tryAcquire(200, TimeUnit.MILLISECONDS)) {
                throw new LoanException(ErrorCode.CONFLICT, "saturated: " + channel);
            }
            try {
                clients.get(channel).send(r, CTX.get());
            } finally {
                sem.release();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new LoanException(ErrorCode.INTERRUPTED, "notify " + channel);
        }
    }

    private synchronized Reservation reserveLocked(String isbn, String memberId) {
        Book book = books.findByIsbn(isbn)
                .orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, isbn));
        if (book.stock() <= 0) {
            throw new LoanException(ErrorCode.OUT_OF_STOCK, isbn);
        }
        books.save(new Book(book.isbn(), book.title(), book.stock() - 1));
        Loan loan = new Loan("L-" + isbn + "-" + memberId, isbn, memberId,
                Instant.now().plusSeconds(86_400));
        loans.save(loan);
        return new Reservation(loan.loanId(), isbn, memberId, loan.dueAt());
    }

    @Override
    public boolean returnBook(String loanId) {
        return returnLocked(loanId);
    }

    private synchronized boolean returnLocked(String loanId) {
        Loan loan = loans.findById(loanId)
                .orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, loanId));
        Book book = books.findByIsbn(loan.isbn())
                .orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, loan.isbn()));
        books.save(new Book(book.isbn(), book.title(), book.stock() + 1));
        return true;
    }

    @Override
    public void close() throws InterruptedException {
        fanout.shutdown();
        fanout.awaitTermination(5, TimeUnit.SECONDS);
    }
}

注意 CTX.get() 只在这个子任务自己 where 出来的作用域内有效。父线程捕获 ctx 对象、在任务内重新绑定，是普通虚拟线程传递上下文的可靠写法；直接指望子线程继承，本机实测拿到的是 <unbound>。

什么时候不要用

场景问题替代
CPU 密集任务没有阻塞可卸载，信号量也救不了固定平台线程池 / ForkJoin
给每条虚拟线程挂大 ThreadLocal10 万 × 1 KB 就是 100 MB显式传参、放请求对象
许可数大于连接池限流形同虚设，压力照旧打到数据库许可 ≤ 池上限，按下游容量算
全局一把信号量慢下游拖住快下游按渠道/租户拆分
用固定大小虚拟线程池并发被池大小锁死newVirtualThreadPerTaskExecutor
以为 ScopedValue 会被 Thread.start 继承子线程读到未绑定，运行期才炸任务内重新绑定或用 15.1 的作用域

最后一条尤其容易踩：单元测试里在父线程绑定、在子线程读，看起来能跑，是因为测试框架可能复用了同一条线程；换成虚拟线程就变成 <unbound>。

小结

虚拟线程把线程成本降下来，但没有增加连接池、下游 QPS 和 ThreadLocal 副本的预算。三件事必须一起做：

  • 用 newVirtualThreadPerTaskExecutor()，不要池化虚拟线程；
  • 用 Semaphore 把并发压到下游容量，许可按渠道拆分、按真实资源计算，拿不到许可要快速失败；
  • 认清 ThreadLocal 是每条虚拟线程一份，10 万条各 1 KB 就是约 104 MB 额外堆；MDC 不会自动传，ScopedValue 也只被 StructuredTaskScope 的 fork() 继承。

本机实测：1 000 个任务不设限时 maxInFlight=1 000、36–57 ms 打穿下游；Semaphore(16) 后恒定 16、2 256–2 286 ms。下一章进入结构化并发，把这些散落的生命周期和限流收进一个作用域。

阅读导航：上一节 14.2 pinning：JDK 25 还剩什么 ｜ 下一节 15.1 StructuredTaskScope 的作用域

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收