换成虚拟线程后，谁来限制下游并发？用 200 个任务验证 Semaphore 的位置

把newFixedThreadPool(8)换成newVirtualThreadPerTaskExecutor()时原来的并发边界也可能一起变了。如果任务同步调用下游并等待返回固定池的 8 个工作线程原本还限制着同时进行的调用数。同样提交 200 个虚拟线程任务本例无并发限制时调用区峰值达到 200先取得共享信号量的 8 个许可峰值就保持为 8。Java 21 的虚拟线程执行器按任务创建线程数量没有固定的池大小上限。Executors 的方法说明明确给出了这个约定。运行环境是 Windows、OpenJDK 21.0.213-58。程序只依赖 JDK用受控等待让进入调用区的任务停留在那里便于观察并发数。先把同时执行的数量留在现场任务若只执行一行println可能在其他任务开始前就结束。这里增加一道等待门任务进入“下游调用区”后先停下来主线程取完快照再统一放行。两组各提交 200 个任务并使用同一个虚拟线程执行器工厂。第一组直接进入调用区第二组先申请 8 个共享许可中的一个。计数器只记录已经进入、尚未退出调用区的任务等待许可的任务不算进去。调用区由内存中的等待夹具代替没有实际发出远程请求所以这里观察到的是并发边界。完整代码保存为src/VirtualThreadLimits.javaimportjava.util.ArrayList;importjava.util.concurrent.CountDownLatch;importjava.util.concurrent.Executors;importjava.util.concurrent.Future;importjava.util.concurrent.Semaphore;importjava.util.concurrent.TimeUnit;importjava.util.concurrent.atomic.AtomicInteger;publicclassVirtualThreadLimits{privatestaticfinalintTASKS200;privatestaticfinalintLIMIT8;privatestaticvoidawait(CountDownLatchlatch)throwsInterruptedException{if(!latch.await(10,TimeUnit.SECONDS)){thrownewIllegalStateException(fixture timed out);}}privatestaticvoidrunCase(booleanlimited)throwsException{Semaphorepermitslimited?newSemaphore(LIMIT,true):null;intexpectedFirstWavelimited?LIMIT:TASKS;varreadynewCountDownLatch(TASKS);varstartnewCountDownLatch(1);varfirstWavenewCountDownLatch(expectedFirstWave);varreleasenewCountDownLatch(1);varactivenewAtomicInteger();varpeaknewAtomicInteger();varcompletednewAtomicInteger();varvirtualTasksnewAtomicInteger();varfuturesnewArrayListFuture?();intheldActive;try(varexecutorExecutors.newVirtualThreadPerTaskExecutor()){try{for(inti0;iTASKS;i){futures.add(executor.submit(()-{ready.countDown();await(start);if(Thread.currentThread().isVirtual()){virtualTasks.incrementAndGet();}if(permits!null){permits.acquire();}try{intnowactive.incrementAndGet();peak.accumulateAndGet(now,Math::max);firstWave.countDown();try{await(release);// A held-open downstream operation.}finally{active.decrementAndGet();}completed.incrementAndGet();}finally{if(permits!null){permits.release();}}returnnull;}));}await(ready);start.countDown();await(firstWave);heldActiveactive.get();}finally{start.countDown();release.countDown();}for(varfuture:futures){future.get(10,TimeUnit.SECONDS);}}if(heldActive!expectedFirstWave||peak.get()!expectedFirstWave||completed.get()!TASKS||virtualTasks.get()!TASKS||active.get()!0||(permits!nullpermits.availablePermits()!LIMIT)){thrownewAssertionError(unexpected concurrency result);}System.out.printf({\case\:\%s\,\tasks\:%d,\virtualTasks\:%d,\heldActive\:%d,\peakActive\:%d,\completed\:%d,\finalActive\:%d,\finalPermits\:%d}%n,limited?semaphore-8:unlimited,TASKS,virtualTasks.get(),heldActive,peak.get(),completed.get(),active.get(),permitsnull?-1:permits.availablePermits());}publicstaticvoidmain(String[]args)throwsException{System.out.println(java.runtime.versionSystem.getProperty(java.runtime.version));runCase(false);runCase(true);System.out.println(PASS);}}在文章目录执行javac--release 21-encoding UTF-8-d out src/VirtualThreadLimits.java java-cpout VirtualThreadLimits--release 21指定编译目标。若找不到newVirtualThreadPerTaskExecutor先核对java -version和javac -version确保两者都支持本文目标版本。正常运行最后输出PASS夹具等待超过 10 秒会报错。两组都是 200 个虚拟线程峰值分别为 200 和 8本次以三个独立 Java 进程重复执行两组结果每次一致进程无并发限制的峰值共享 8 个许可的峰值每组完成任务数120082002200820032008200第一组的原始结果行是{case:unlimited,tasks:200,virtualTasks:200,heldActive:200,peakActive:200,completed:200,finalActive:0,finalPermits:-1}第二组为{case:semaphore-8,tasks:200,virtualTasks:200,heldActive:8,peakActive:8,completed:200,finalActive:0,finalPermits:8}virtualTasks200说明两组确实都在虚拟线程里执行。peakActive才是本文关心的同时调用数finalActive0、许可最终回到 8说明任务完成后没有把调用区计数或许可留住。第一组的-1只是“未启用信号量”的输出标记。第一波任务被主动留在调用区200 和 8 因而是受控夹具下的结果。它说明共享信号量确实限制了这一段代码的并发真实服务的自然峰值、吞吐和耗时还需要另测。许可需要围住哪一段代码先acquire()成功后才进入包含release()的try/finally。这样等待许可时若被中断就不会释放一个自己没有取得的许可。Semaphore 的 API 约定也允许等待中的acquire()响应中断。放到应用里许可通常围住真正消耗有限资源的那段调用。若把解析请求、读缓存、拼响应都围进去和下游无关的工作也会占用名额。若拿到许可后立即释放再去发请求限制则根本没有覆盖远程调用。还有一个常见的无效写法每个请求内部都new Semaphore(8)。这些对象各有自己的 8 个许可彼此没有竞争关系。本文把信号量放在这一组任务共享的位置真实应用也需要先明确它代表哪一种资源以及哪些调用路径必须共享它。加了信号量仍然要处理等待假设远程接口允许同时 8 个调用信号量可以表达这个限制但它不会自动决定第 9 个请求该等多久也不会限制每秒可以发出多少次请求。并发数、请求速率、入口积压是三个不同问题。阻塞式acquire()方便观察这次实验实际请求还需要等待预算。请求有截止时间时应把等许可、建连接和读取响应放进同一份预算等许可太久可以通过带超时的tryAcquire或上层准入规则结束。等待的虚拟线程还持有请求正文、上下文等资源这部分成本也要计算。也不必给所有调用再套一层信号量。例如数据库连接池已经把可用连接数限制在固定范围再加同样大小的一层未必带来收益。要先看实际瓶颈和其他调用入口避免双重排队。Oracle 的虚拟线程采用指南讨论了用信号量限制资源访问并指出连接池自身也承担类似作用。本文选 8 是为了让两个实验分支的差别容易观察不是给服务推荐一个通用并发值。真实值要结合下游约定、连接容量、调用时长和允许等待的请求数来确定进程里放 8 个许可也不会自动变成多个部署副本合计 8 个。迁移前先列出原线程池大小隐含的资源限制再逐项确定由连接池、共享信号量还是入口准入来承接。这个名单比直接沿用原来的数字更有用。