决策树
你的操作会阻塞线程吗？
├── 会（JDBC、阻塞 HTTP、文件 I/O、sleep）
│ └── boundedElastic()，且并发度受下游连接池约束
├── 不会，但吃 CPU（计算、序列化、压缩）
│ └── parallel()
├── 需要严格串行（状态机、命令队列、单写者）
│ └── single() 或 newSingle("name")
└── 只是测试/不需要切换
└── 当前线程；时间相关测试用 VirtualTimeScheduler
先记住一句话：调度器决定“在哪个线程执行”，不自动决定“是否并发”。很多“定时任务并发执行”的问题，根因不在调度器，而在 flatMap。
场景一：CPU 密集型
Flux.range(1, 1_000_000)
.parallel() // 默认并行度 = CPU 核数
.runOn(Schedulers.parallel())
.map(i -> expensiveComputation(i))
.sequential()
.subscribe();
Schedulers.parallel() 的 worker 数量默认等于 Runtime.availableProcessors()，适合不阻塞的 CPU 任务。parallel(n) 里的 n 是并行度，不是“线程池大小建议值”；在 4 核机器上写 parallel(8) 只会增加争抢。
反例：在 parallel 上阻塞
Flux.range(1, 100)
.parallel()
.runOn(Schedulers.parallel())
.map(i -> {
Thread.sleep(1000); // 阻塞
return i;
})
.sequential()
.subscribe();
parallel() 的 worker 数量固定。所有 worker 都被 sleep 占住后，后续任务只能排队，整条链的吞吐降到接近零。阻塞操作应放到 boundedElastic()，并且并发度不要超过下游资源的上限。
场景二：阻塞 I/O
Flux.range(1, 100)
.flatMap(i -> Mono.fromCallable(() -> database.query(i))
.subscribeOn(Schedulers.boundedElastic()),
10) // 同时在途的查询数
.subscribe();
boundedElastic() 的关键参数不是“固定线程数”，而是：
| 参数 | 默认值 | 说明 |
|---|---|---|
| 最大线程数 | CPU × 10 | 按需创建，不是启动时就建满 |
| 每线程排队上限 | 100,000 | 超过后抛 RejectedExecutionException |
| 空闲回收 | 60 秒 | 空闲线程会被回收 |
flatMap 的第二个参数（concurrency）决定同时在途的内层订阅数。它应该由下游容量决定，而不是拍脑袋写大：
| 下游 | 并发上限由什么决定 |
|---|---|
| 数据库 | 连接池最大连接数；并发 > 池大小只会排队 |
| HTTP 服务 | 客户端连接池与目标服务限流 |
| 文件系统 | 磁盘吞吐与文件句柄数 |
| 混合调用 | 取最紧的那个下游，再从保守值开始压测 |
反例：不必要的多次线程切换
Mono.fromCallable(() -> step1())
.subscribeOn(Schedulers.boundedElastic())
.flatMap(r -> Mono.fromCallable(() -> step2(r))
.subscribeOn(Schedulers.boundedElastic()))
.subscribe();
如果 step1 和 step2 都是阻塞调用，而且必须串行，直接在一个 fromCallable 里写完更省一次调度：
Mono.fromCallable(() -> {
var r1 = step1();
return step2(r1);
}).subscribeOn(Schedulers.boundedElastic()).subscribe();
场景三：定时任务
Flux.interval(Duration.ofSeconds(5), Schedulers.single())
.concatMap(i -> healthCheck()) // 上一次没完成就不开始下一次
.subscribe();
Flux.interval 本身是串行发射的：无论用 single() 还是 parallel()，一个 interval 实例的 tick 都由同一个 worker 按顺序发出。把调度器换成 parallel() 不会让 map 并发执行。真正会引入并发的是 flatMap：
// 危险：多个 tick 的内层任务可能重叠
Flux.interval(Duration.ofSeconds(5))
.flatMap(i -> Mono.fromCallable(this::updateState)
.subscribeOn(Schedulers.parallel()))
.subscribe();
如果业务要求“上一次执行完才能执行下一次”，用 concatMap，或者把内层任务固定到同一个单线程调度器：
Scheduler serial = Schedulers.newSingle("state-writer");
Flux.interval(Duration.ofSeconds(5))
.concatMap(i -> Mono.fromCallable(this::updateState)
.subscribeOn(serial))
.subscribe();
不要长期占用全局的 Schedulers.single()：它只有一条线程，一个慢任务会拖住所有使用它的组件。需要串行时，创建自己的 newSingle("name")，并在应用关闭时 dispose()。
场景四：测试与虚拟时间
同步逻辑直接测，不需要调度器：
@Test
void testSync() {
Flux.just(1, 2, 3)
.map(i -> i * 2)
.as(StepVerifier::create)
.expectNext(2, 4, 6)
.verifyComplete();
}
时间相关逻辑用 StepVerifier.withVirtualTime。注意：必须在 lambda 内部创建操作符，虚拟时间才会注入到 Schedulers 工厂：
@Test
void testWithVirtualTime() {
StepVerifier.withVirtualTime(() ->
Flux.interval(Duration.ofMinutes(1)).take(3))
.expectSubscription()
.expectNoEvent(Duration.ofMinutes(1))
.expectNext(0L)
.thenAwait(Duration.ofMinutes(1))
.expectNext(1L)
.thenAwait(Duration.ofMinutes(1))
.expectNext(2L)
.verifyComplete();
}
旧版示例手动 VirtualTimeScheduler.create() 并给 withVirtualTime 传 1 作为初始请求量。第三个参数是 initialRequest，传 1 意味着只请求一个元素；后面的 expectNext(1L)、expectNext(2L) 没有新的请求，测试会卡住或失败。要么不传这个参数（默认无界），要么按预期元素数显式给足。
自定义执行器
ExecutorService pool = Executors.newFixedThreadPool(10, r -> {
Thread t = new Thread(r, "biz-pool");
t.setDaemon(true);
return t;
});
Scheduler scheduler = Schedulers.fromExecutorService(pool);
Flux.range(1, 100)
.publishOn(scheduler)
.doOnNext(this::process)
.subscribe();
Schedulers.fromExecutor(Executor, boolean) 的第二个参数是 trampoline，不是“是否立即启动”：true 表示把一批任务排队后作为一个任务提交给 executor，避免重排序，但也失去多线程并行。给 publishOn/subscribeOn 用时通常传 false，或者直接用 fromExecutorService(...)。
需要按 key 路由到固定线程时，Reactor 没有公开的 PartitionedScheduler。可以按分片创建一组 newSingle("shard-N")，用 hash(key) % N 选择：
Scheduler[] shards = IntStream.range(0, 4)
.mapToObj(i -> Schedulers.newSingle("shard-" + i))
.toArray(Scheduler[]::new);
Scheduler shardFor(String key) {
return shards[Math.floorMod(key.hashCode(), shards.length)];
}
关闭应用时逐个 dispose()。
线程模型速查
| 调度器 | 线程数 | 队列 | 空闲回收 | 适用 |
|---|---|---|---|---|
immediate() | 调用线程 | 无 | N/A | 测试、已确认非阻塞 |
single() | 1 | 无界 | 不回收 | 串行任务、状态机 |
parallel() | CPU 核数 | 无 | 不回收 | 非阻塞 CPU 任务 |
boundedElastic() | 按需，最大 CPU×10 | 每线程 100,000 | 60 秒 | 阻塞 I/O |
这张表是线程模型，不是性能排名。boundedElastic() 的上下文切换比 parallel() 多，但对阻塞调用是唯一合理的选择；把阻塞调用放到 parallel() 上，吞吐不是“低一点”，而是会被彻底拖死。
容器里的 CPU 核数
旧版说“Runtime.availableProcessors() 在容器里不可靠”，这个结论已经过时。JDK 10 引入了容器感知，后续版本补齐了 cgroup v2 支持；在本文使用的 JDK 25 上，它默认按 cgroup 的 CPU 配额计算，可以直接使用。需要覆盖时用：
java -XX:ActiveProcessorCount=4 -jar app.jar
如果怀疑运行时读到的核数不对，再手工核对 cgroup。cgroup v2 的配额在 /sys/fs/cgroup/cpu.max，格式是 quota period（例如 200000 100000 表示 2 核）；v1 在 /sys/fs/cgroup/cpu/cpu.cfs_quota_us 与 cpu.cfs_period_us：
static int cgroupCpuLimit() throws IOException {
Path v2 = Path.of("/sys/fs/cgroup/cpu.max");
if (Files.exists(v2)) {
String[] parts = Files.readString(v2).trim().split("\\s+");
if ("max".equals(parts[0])) return -1;
return (int) Math.max(1,
Long.parseLong(parts[0]) / Long.parseLong(parts[1]));
}
Path quotaPath = Path.of("/sys/fs/cgroup/cpu/cpu.cfs_quota_us");
Path periodPath = Path.of("/sys/fs/cgroup/cpu/cpu.cfs_period_us");
if (Files.exists(quotaPath)) {
long quota = Long.parseLong(Files.readString(quotaPath).trim());
if (quota < 0) return -1;
long period = Long.parseLong(Files.readString(periodPath).trim());
return (int) Math.max(1, quota / period);
}
return -1;
}
返回 -1 表示没有限制。注意容器内的 cgroup 挂载路径可能不同，脚本只适合常见 Docker/containerd 布局；生产环境优先相信 JVM 的自动探测，手工读取只用于诊断。
常见错误速查
| 错误 | 现象 | 修复 |
|---|---|---|
阻塞调用放在 parallel() | worker 全被占满，吞吐接近零 | 改用 boundedElastic() |
flatMap 不设并发 | 同时打开过多连接/文件 | 显式传 concurrency |
用 flatMap 做“串行定时任务” | 多次执行重叠 | 改用 concatMap |
长期占用全局 single() | 一个慢任务拖住所有使用者 | newSingle("name") 专用 |
| 自定义调度器不关闭 | 线程泄漏 | 关闭时 dispose() |
用 totalMemory 判断调度开销 | 数字随 GC 波动，无法复现 | 用 JFR/GC 日志或固定基准 |
阅读导航
下一篇：附录 C 背压规则清单
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。