6.2 BoundedElastic 与线程池模型

它要解决什么问题

在响应式流里执行阻塞操作（JDBC、文件、阻塞 HTTP 客户端）会占住线程。早期 Schedulers.elastic() 不限制线程数，突发流量可以把线程数一路拉高，最终拖垮进程。它在 Reactor 3.8 里已经不存在，替代品 boundedElastic 加了两道硬上限：

这些默认值可以从公开常量核对：

System.out.println(Schedulers.DEFAULT_BOUNDED_ELASTIC_SIZE);      // 10 核机器 → 100
System.out.println(Schedulers.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE); // 100000

也可以用系统属性覆盖：

-Dreactor.schedulers.defaultBoundedElasticSize=60
-Dreactor.schedulers.defaultBoundedElasticQueueSize=20000

线程生命周期

创建：按需创建，直到上限

for (int i = 0; i < 5; i++) {
    int idx = i;
    Mono.fromCallable(() -> {
        System.out.println("task " + idx + " on "
            + Thread.currentThread().getName());
        Thread.sleep(100);
        return idx;
    }).subscribeOn(Schedulers.boundedElastic()).subscribe();
}
Thread.sleep(600);

在 10 核机器上实测：

task 0 on boundedElastic-1
task 1 on boundedElastic-2
task 2 on boundedElastic-3
task 3 on boundedElastic-4
task 4 on boundedElastic-5

5 个任务同时在跑，调度器就创建 5 条线程。默认上限是 100，远未到顶。不要把「最多 10×CPU」误记成「只有 CPU 条线程」——那是 parallel()。

复用：worker 空闲后才会被再次选中

线程不是用完就销毁。任务结束后 worker 回到空闲队列，后续任务会优先复用它。下面三个任务用 block() 串行执行：

for (int i = 0; i < 3; i++) {
    int idx = i;
    Mono.fromCallable(() -> {
        System.out.println("seq " + idx + " on "
            + Thread.currentThread().getName());
        return idx;
    }).subscribeOn(ttl).block();
}

实测三个任务都在同一条线程上：

seq 0 on ttl-6
seq 1 on ttl-6
seq 2 on ttl-6

线程名后面的编号来自 Reactor 的全局线程计数器，不是「这个调度器的第几条线程」，所以编号可能从 6 开始。不要用编号推断线程总数。

回收：空闲超过 TTL

用 newBoundedElastic(2, 10, "ttl", 1, true) 把 TTL 调成 1 秒。三个串行任务结束后等待 2.5 秒，再提交一个任务：

seq 0 on ttl-6
seq 1 on ttl-6
seq 2 on ttl-6
after idle on ttl-7

旧线程在空闲 1 秒后被回收，新任务创建了 ttl-7。TTL 的代价是频繁创建/销毁线程；稳定负载下保持默认 60 秒更划算。

Worker 与队列模型

boundedElastic 不是「一个 ThreadPoolExecutor 加一个共享队列」，而是一组 worker：

• 每个 worker 是一个 core = max = 1 的 ScheduledThreadPoolExecutor，构造后设置 setMaximumPoolSize(1) 和 setRemoveOnCancelPolicy(true)；
• 任务进入该线程的 DelayedWorkQueue，同一个 worker 上的任务严格串行，不同 worker 之间才并行；
• JDK 的 DelayedWorkQueue 本身无界，Reactor 在入队前调用 ensureQueueCapacity：队列长度 + 本次任务数 > 容量 就拒绝。

选哪个 worker

分配策略不是 round-robin，而是按顺序尝试：

1. 有空闲 worker → 直接复用；
2. 没有空闲且未到线程上限 → 新建一个；
3. 已到上限 → 选当前挂载任务数最少的 worker。

任务一旦选中某个 worker，就固定在该线程上执行，不会在空闲后被其他线程「偷走」。

队列满时发生什么

用一个 worker、队列容量 2 的调度器跑 6 个耗时 300ms 的任务：

Scheduler tiny = Schedulers.newBoundedElastic(1, 2, "tiny", 1, true);

实测输出（通过 subscribeOn 提交）：

submitted 0
submitted 1
run 0 on tiny-8
submitted 2
onError 3: reactor.core.Exceptions$ReactorRejectedExecutionException:
  Task capacity of bounded elastic scheduler reached while scheduling 1 tasks (3/2)
submitted 3
onError 4: ...
submitted 4
onError 5: ...
run 1 on tiny-8
run 2 on tiny-8

前 3 个任务被接受：0 在运行，1 和 2 在队列里；从第 4 个开始被拒。(3/2) 表示「本次提交后应有 3 个排队任务，容量只有 2」。

注意拒绝的传递方式：

• 通过 subscribeOn(...) 提交时，RejectedExecutionException 通过下游的 onError 送达，subscribe() 本身不抛；
• 直接调用 Scheduler.Worker.schedule(...) 时，异常同步抛出。

Scheduler.Worker worker = Schedulers.boundedElastic().createWorker();
try {
    worker.schedule(() -> doBlockingWork());
} catch (RejectedExecutionException e) {
    // 处理拒绝
} finally {
    worker.dispose();
}

这里用的是 createWorker()，不是 createExecutor()——Scheduler 上只有 schedule、schedulePeriodically 和 createWorker()，没有 createExecutor()。Scheduler.Worker 才能承接任务；上例必须保留 worker.dispose()，否则 worker 会一直存活。

线程耗尽与队列风险

线程上限是 100（10 核）时，第 101 个并发任务不会创建新线程，而是进入某个 worker 的队列。100 个 worker × 10 万队列 = 最多 1000 万个待执行任务——这个上限足够大，大到能在拒绝发生之前先把内存和延迟拖垮。

因此不要依赖调度器的队列做背压。真正的并发控制在 flatMap：

Flux.range(1, 1000)
    .flatMap(i -> Mono.fromCallable(() -> jdbcQuery(i))
        .subscribeOn(Schedulers.boundedElastic()), 16)
    .subscribe();

第二个参数 16 表示同时最多 16 个内层订阅，boundedElastic 最多创建 16 条线程，队列里不会堆 1000 个任务。

与 ExecutorService 的差异

把它硬套成 ThreadPoolExecutor 会得出错误结论，例如以为队列是 SynchronousQueue、以为拒绝策略是 CallerRuns——都不是。

自定义 boundedElastic

Scheduler custom = Schedulers.newBoundedElastic(
    20,            // 线程上限
    200,           // 每个 worker 的队列容量
    "my-io-pool",  // 线程名前缀
    30,            // 空闲 TTL（秒）
    true           // 是否守护线程
);

注意第二个参数是每个 worker 的容量，不是全局总容量。20 个 worker × 200 = 最多 4000 个排队任务。

自定义调度器用完要 dispose()，否则线程会一直存活到进程结束：

try {
    // 使用 custom
} finally {
    custom.dispose();
}

虚拟线程模式（JDK 21+）

Reactor 3.6 起为 boundedElastic 提供了虚拟线程实现，默认关闭。开启方式：

java -Dreactor.schedulers.defaultBoundedElasticOnVirtualThreads=true ...

开启后（且运行在 JDK 21+），Schedulers.boundedElastic() 使用 BoundedElasticThreadPerTaskScheduler：

• 每个任务交给一条新的虚拟线程执行；
• maxThreads 的含义变成「同时运行的虚拟线程数上限」，默认仍是 10 × CPU；
• 排队任务不占用平台线程；
• 线程名前缀是 loomBoundedElastic。

在低于 JDK 21 的运行时设置该属性不会生效，Reactor 会打印警告并回退到平台线程实现。

代价与限制：

• 虚拟线程适合大量阻塞等待，但 CPU 密集任务不会因此变快；
• JDK 21–23 上，阻塞在 synchronized 块中的虚拟线程会 pin 住载体线程；JDK 24 的 JEP 491 解除了这一限制。老 JDBC 驱动里大量 synchronized 时，先压测再开；
• 队列上限仍然存在，maxConcurrency 与 queuedTaskCap 仍要按业务配置。

不想依赖全局属性时，Schedulers.Factory 接口里有 newThreadPerTaskBoundedElastic(int, int, ThreadFactory)，可以通过 Schedulers.setFactory(...) 替换默认工厂。注意这个接口没有默认方法，四个工厂方法（newBoundedElastic、newThreadPerTaskBoundedElastic、newParallel、newSingle）都必须实现，通常只在框架集成时使用；大多数应用直接用全局属性即可。

最佳实践

1. 阻塞操作才用 boundedElastic

// 正确：JDBC 是阻塞的
Flux.range(1, 100)
    .flatMap(i -> Mono.fromCallable(() -> database.query(i))
        .subscribeOn(Schedulers.boundedElastic()), 8)
    .subscribe();

// 错误：阻塞 parallel 线程，还会拖住 interval/timeout 的计时器
Flux.range(1, 100)
    .flatMap(i -> Mono.fromCallable(() -> database.query(i))
        .subscribeOn(Schedulers.parallel()), 8)
    .subscribe();

2. 合并相邻的阻塞步骤

// 两次线程切换
Mono.fromCallable(() -> step1())
    .subscribeOn(Schedulers.boundedElastic())
    .flatMap(r1 -> Mono.fromCallable(() -> step2(r1))
        .subscribeOn(Schedulers.boundedElastic()))
    .subscribe();

// 一次切换：两个步骤都在同一个阻塞调用里
Mono.fromCallable(() -> step2(step1()))
    .subscribeOn(Schedulers.boundedElastic())
    .subscribe();

前提是 step1/step2 都是阻塞代码。如果其中一步本身是异步客户端，就不要为了省切换把它改成阻塞等待。

3. 线程耗尽时的降级要算清代价

Mono.fromCallable(() -> database.query())
    .subscribeOn(Schedulers.boundedElastic())
    .timeout(Duration.ofSeconds(5))
    .onErrorResume(TimeoutException.class, e -> cache.getFallback())
    .subscribe();

超时能让调用方尽快拿到降级结果，但不会释放已经占用的线程：正在执行的 JDBC 查询不会被中断，排队中的任务也可能在之后继续执行。降级减少的是用户等待时间，不是后端压力。要真正减少压力，需要限制并发（flatMap 的 concurrency）、缩短查询超时，或在入口做限流。

诊断

# 统计当前 boundedElastic 线程数
jcmd <pid> Thread.print | grep -c '"boundedElastic'

# 查看这些线程在做什么
jcmd <pid> Thread.print | grep -A 15 '"boundedElastic'

典型症状：

• 请求延迟上升，但 CPU 利用率不高——线程都在等 I/O；
• 队列持续增长，GC 压力变大——并发控制失效；
• 出现 ReactorRejectedExecutionException——线程和队列都到顶了，说明入口流量远超处理能力。

小结

下一节：自定义调度器与 Worker 模型。

阅读导航

上一篇：6.1 调度器：单线程、弹性、并行与立即

下一篇：6.3 自定义调度器与 Worker 模型