Java 虚拟线程怎么用 Semaphore 限制外部资源并发

Java 虚拟线程可以为每个任务创建一个线程，但数据库连接、第三方 HTTP 接口和文件句柄仍有明确容量。正确做法不是把虚拟线程放进固定大小的线程池，而是在真正访问稀缺资源的方法外包一层 Semaphore：许可数就是允许同时进入资源区的任务数，成功获取后执行调用，并在 finally 中释放。

官方资料：https://openjdk.org/jeps/444

OpenJDK 在 JEP 444 中明确建议：虚拟线程便宜且充足，不应为了限制并发而池化；需要保护有限资源时，应使用 Semaphore 这类专门的同步结构。

一、先分清两种容量：线程很多，不等于资源无限

虚拟线程解决的是“等待 I/O 时不要长期占住昂贵的平台线程”。它提高的是大量阻塞任务的可伸缩性，并不会增加下游数据库连接数、接口限额或设备处理能力。把这两种容量混在一起，常见结果是虚拟线程创建得很顺畅，外部服务却被瞬时并发压垮。

二、把许可数设计成资源参数

最小可用设计只需要一个参数：maxConcurrent。它不应根据“机器能开多少虚拟线程”来定，而应根据下游资源的容量来定。例如接口只允许 20 个并行请求，许可数就从 20 开始；数据库连接池最多 30 条连接时，还要给其他调用路径留出余量，不能机械设成 30。

import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.Semaphore;

public final class ExternalServiceLimiter {
    private final Semaphore permits;

    public ExternalServiceLimiter(int maxConcurrent, boolean fair) {
        if (maxConcurrent  T call(Callable action) throws Exception {
        Objects.requireNonNull(action, "action");

        // acquire 可被中断；成功返回后当前任务才拥有一个许可
        permits.acquire();
        try {
            // 只有持有许可的虚拟线程才能访问受限外部资源
            return action.call();
        } finally {
            // 外部调用正常返回或抛异常时都必须归还许可
            permits.release();
        }
    }
}

fair 决定是否使用公平模式。公平模式会尽量按等待顺序分配许可，适合需要降低长时间饥饿风险的场景；非公平模式通常吞吐更高。它不是虚拟线程的必选项，应该根据调用方是否要求排队公平来决定。

三、把获取、调用和释放封装在同一个方法里

接口设计的重点是让“拥有许可”和“负责释放”落在同一个作用域。不要让调用方先 acquire()，再把许可跨多个方法传递；这种写法很容易在异常分支漏掉 release()。

上面的实现还有一个容易忽略的细节：try 块放在 acquire() 之后。若线程在等待许可时被中断，acquire() 会抛出 InterruptedException，此时任务从未拿到许可，所以也不能调用 release()。只有获取成功，才进入需要释放的作用域。

四、调用方保留每任务一个虚拟线程

限制外部服务并发后，任务执行器仍然使用“每任务一个虚拟线程”。没有许可的任务会停在 Semaphore 上；JEP 444 说明，Semaphore 所依赖的阻塞机制可以让虚拟线程正常 park，而不会为了等待许可持续占用载体线程。

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.List;
import java.util.concurrent.Executors;

var limiter = new ExternalServiceLimiter(20, false);
var client = HttpClient.newHttpClient();

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    var futures = List.of("1001", "1002", "1003").stream()
        .map(id -> executor.submit(() -> limiter.call(() -> {
            // 每个任务都有自己的虚拟线程，但外部调用最多同时进入 20 个
            var request = HttpRequest.newBuilder()
                .uri(URI.create("https://service.example/items/" + id))
                .GET()
                .build();

            // send 是阻塞式调用，代码保持顺序风格
            return client.send(request, HttpResponse.BodyHandlers.ofString());
        })))
        .toList();

    for (var future : futures) {
        // 示例在任务边界集中等待并处理结果
        var response = future.get();
        System.out.println(response.statusCode());
    }
}

示例域名只表示受限外部服务的位置，实际项目应替换为已配置的客户端地址。核心结构是：任务数量由业务负载决定，资源并发由 Semaphore 决定，两者不要绑成同一个线程池大小。

五、等待许可也要有超时和错误语义

直接调用 acquire() 会一直等到许可可用或线程被中断。面向在线请求时，通常还需要等待上限。可以用 tryAcquire 把“限流等待超时”与“下游调用失败”区分开。

import java.time.Duration;
import java.util.concurrent.Callable;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public final class TimedLimiter {
    private final Semaphore permits;

    public TimedLimiter(int maxConcurrent) {
        // 非公平模式适合优先追求整体吞吐的普通接口
        this.permits = new Semaphore(maxConcurrent);
    }

    public  T call(Duration wait, Callable action) throws Exception {
        // 等待许可超时属于本地过载，不应伪装成下游服务错误
        boolean acquired = permits.tryAcquire(wait.toMillis(), TimeUnit.MILLISECONDS);
        if (!acquired) {
            throw new ResourceBusyException("resource concurrency limit reached");
        }

        try {
            return action.call();
        } finally {
            // acquired 为 true 后，无论 action 如何结束都归还许可
            permits.release();
        }
    }
}

final class ResourceBusyException extends RuntimeException {
    ResourceBusyException(String message) {
        // 保留明确的本地限流错误信息，方便上层映射状态码
        super(message);
    }
}

建议调用方分别统计许可等待超时、InterruptedException、下游超时和下游业务错误。若在边界层捕获 InterruptedException，要么继续抛出，要么调用 Thread.currentThread().interrupt() 恢复中断标记，不能静默吞掉取消信号。

六、怎么确认限制真的生效

不要只看虚拟线程总数。验证指标应围绕受保护资源：

• 正在执行外部调用的实时数量，峰值不应超过许可数；
• 等待许可的任务数量和等待时长；
• 许可等待超时次数；
• 外部服务自身的延迟、拒绝率和连接池占用率。

availablePermits() 可以用于诊断，但它只是读取瞬时状态，不应成为并发正确性的判断条件。真正的上限由 acquire/tryAcquire 与 release 的配对保证。

常见问题

为什么不用固定大小的虚拟线程池？

虚拟线程本身不是稀缺资源。固定线程池会把任务调度和外部资源限流耦合在一起；Semaphore 直接表达“这个资源允许多少个并发访问”，意图更清楚，也能让其他不访问该资源的虚拟线程继续独立运行。

许可数应该等于数据库连接池大小吗？

不一定。如果同一个连接池还服务其他路径，应预留容量；若一个任务会占多个连接或包含事务，也要按真实占用模型调整。先从下游容量和服务等级目标出发，再通过监控修正。

Semaphore 能限制请求速率吗？

Semaphore 限制的是同时进行的数量，不是每秒请求数。需要 QPS、突发容量或时间窗口控制时，应另用速率限制器；两者可以组合，但语义不同。

总结

虚拟线程负责让大量阻塞任务以简单的顺序代码运行，Semaphore 负责保护有限的外部资源。把许可数作为资源参数，把获取和释放封装在同一方法，并明确超时、中断和下游错误，就能在保留每任务一个虚拟线程的同时，把真实并发压在下游可承受范围内。