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

四种调度器解决四个不同问题

第 3 章手写的 subscribeOn/publishOn 说明了线程切换的机制，但没有回答「切到哪个线程池」。Reactor 内置了四种调度器：

默认值都可以被系统属性覆盖，后文会给出具体名字。

single()：全局单线程

Flux.range(1, 3)
    .publishOn(Schedulers.single())
    .doOnNext(i -> System.out.println(
        "item " + i + " on " + Thread.currentThread().getName()))
    .blockLast();

实测输出：

item 1 on single-1
item 2 on single-1
item 3 on single-1

Schedulers.single() 返回一个全局共享的 SingleScheduler，底层只有一个单线程 ScheduledExecutorService。所有调用 single() 的流都排队用这一条线程，任务按提交顺序执行。

这也意味着它不适合当作「每个业务一条顺序队列」来用：订单流和支付流会互相排队。需要多个独立的串行执行器时，用 Schedulers.newSingle("order")、Schedulers.newSingle("payment") 分别创建。

适用场景：轻量的状态机推进、必须严格串行的簿记操作。不适合任何可能阻塞或耗时的任务——一条线程被占住，所有使用者一起等。

parallel()：CPU 核数个线程

Flux.range(1, 10)
    .parallel()
    .runOn(Schedulers.parallel())
    .doOnNext(i -> System.out.println(
        "item " + i + " on " + Thread.currentThread().getName()))
    .sequential()
    .blockLast();

在 10 核机器上实测（打印顺序不稳定）：

item 1 on parallel-1
item 4 on parallel-4
item 3 on parallel-3
item 2 on parallel-2
item 10 on parallel-10
...

Schedulers.parallel() 默认创建 Runtime.availableProcessors() 个单线程的 ScheduledThreadPoolExecutor，每个 Worker 通过内部的 roundRobin 计数器轮询选择其中一个，没有 work-stealing。Flux.parallel() 默认也按 CPU 核数切分 rail，所以 10 个元素正好一条 rail 一个元素。

关键限制：parallel() 也承担了 Flux.interval、timeout、delayElements 等基于时间的操作符的计时任务。只要有几个 parallel 线程被阻塞，定时器就可能延迟甚至停摆。CPU 密集型任务应该短小且不阻塞；任何 I/O 都不应该放在这里。

系统属性 reactor.schedulers.defaultPoolSize 可以覆盖默认线程数。

boundedElastic()：为阻塞任务准备的有界池

Flux.range(1, 100)
    .flatMap(i -> Mono.fromCallable(() -> {
            Thread.sleep(100);          // 模拟 JDBC 查询
            return "result" + i;
        }).subscribeOn(Schedulers.boundedElastic()),
        20)                             // 同时最多 20 个阻塞任务
    .subscribe();

默认参数：

线程是按需创建的，不是启动时就开满。选择顺序是：先找空闲线程，没有空闲且未到上限就新建，已到上限则选当前挂载任务最少的线程。任务在哪个线程上执行就固定在哪个线程，不会被空闲线程偷走。

实测：10 核机器上的 20 个阻塞任务

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

实测输出：

task 0 on boundedElastic-1
task 1 on boundedElastic-2
...
task 19 on boundedElastic-20

这里出现了 20 条线程，而不是 10 条。因为默认上限是 10 × 10 = 100，20 个并发任务远未到顶，调度器直接为每个任务创建了新线程。「boundedElastic 只有 CPU 个线程」是把 parallel 的默认值记混了。

到达上限后开始排队

用构造参数把上限压到 4，跑 12 个耗时 200ms 的任务：

Scheduler io = Schedulers.newBoundedElastic(4, 100, "io");

实测（线程名与时间戳来自本次运行）：

task 0 on io-1 at +5ms
task 1 on io-2 at +5ms
task 2 on io-3 at +6ms
task 3 on io-4 at +6ms
task 4 on io-1 at +218ms
task 5 on io-2 at +218ms
task 6 on io-3 at +218ms
task 7 on io-4 at +218ms
task 8 on io-1 at +430ms
...

4 条线程、3 批任务，每批约 200ms。每个线程最多再排 100000 个任务；当所有线程的队列都满时，提交会抛 RejectedExecutionException（实际类型是 Reactor 内部的包级子类 Exceptions$ReactorRejectedExecutionException，按父类型捕获即可），不会阻塞调用者，也不会把任务丢回调用线程执行。

这个「每线程 10 万队列」的上限在故障时很危险：100 条线程 × 10 万 = 1000 万个待执行任务，足以把内存和延迟一起拖垮。真实的 I/O 并发应该由 flatMap 的 concurrency 参数控制，而不是依赖队列兜底。

immediate()：不切换线程

Flux.range(1, 3)
    .publishOn(Schedulers.immediate())
    .doOnNext(i -> System.out.println(
        "item " + i + " on " + Thread.currentThread().getName()))
    .blockLast();

实测：

item 1 on main
item 2 on main
item 3 on main

immediate() 的 schedule 直接在调用线程执行，等于没有切换。它适合测试，或用于明确表达「这里不需要异步」。注意 subscribeOn(Schedulers.immediate()) 不会让订阅变成异步，它只是把订阅动作原样执行。

选择指南

陷阱一：在 parallel 上阻塞

// 错误：阻塞 parallel 线程
Flux.range(1, 100)
    .parallel()
    .runOn(Schedulers.parallel())
    .map(i -> {
        Thread.sleep(1000);
        return i;
    })
    .sequential()
    .subscribe();

parallel 只有 CPU 核数个线程。全部阻塞后，不只是这条流变慢：同进程里所有依赖 parallel 的 interval、timeout、delayElements 都会排队，超时可能迟迟不触发。

正确做法是把阻塞调用移出计算线程：

Flux.range(1, 100)
    .flatMap(i -> Mono.fromCallable(() -> blockingCall(i))
            .subscribeOn(Schedulers.boundedElastic()),
        8)
    .subscribe();

flatMap 的第二个参数限制并发，避免 100 个任务同时挤进 boundedElastic。

陷阱二：把 single 当万能串行队列

// 错误：所有任务共享同一条 single-1 线程
Flux.range(1, 1000)
    .flatMap(i -> process(i).subscribeOn(Schedulers.single()))
    .subscribe();

1000 个任务排队等一条线程。如果 process 里还有 10ms 的阻塞，总耗时至少 10 秒。需要顺序时，先问清楚「是这条流内部有序，还是不同业务之间也要有序」：前者用 concatMap 就能保证，不需要 single；后者才考虑独立的 newSingle。

陷阱三：用 boundedElastic 跑 CPU 密集任务

boundedElastic 的线程上限很高（默认 100），把 CPU 密集任务放进去不会提高吞吐，只会让 100 条线程争抢 10 个核，增加上下文切换和内存占用。CPU 密集任务用 parallel()，或者用 Flux.parallel().runOn(Schedulers.parallel()) 明确表达并行意图。

平台与容器差异

• 容器 CPU 配额：Runtime.availableProcessors() 在受支持的 JDK 上是 cgroup-aware 的，会读取 CPU quota，而不是宿主机的核数。4 vCPU 的容器里，boundedElastic 默认上限是 40，parallel 默认 4 条线程。
• JDK 21+ 虚拟线程：boundedElastic 只有在显式设置 reactor.schedulers.defaultBoundedElasticOnVirtualThreads=true 时才切换到虚拟线程实现；默认仍是平台线程。不要假设升级 JDK 就自动变成虚拟线程。
• Reactor Netty 的 native transport：Linux 上是否使用 epoll、macOS 上是否使用 kqueue，取决于 classpath 里有没有对应的 native 依赖以及运行时的可用性；没有时会回退到 NIO。性能结论要在目标平台实测，不能把 macOS 的开发数据直接外推到 Linux。
• 计时精度：不同操作系统的定时器精度不同，delayElements/timeout 的毫秒级断言在 CI 里容易偶发失败，测试应留余量或使用 VirtualTimeScheduler。

小结

下一节深入 boundedElastic 的线程上限、队列和 TTL，以及容器里怎么配置。

阅读导航

上一篇：5.3 超时与取消传播

下一篇：6.2 BoundedElastic 与线程池模型