两个线程问题
一条响应式流里有两个不同的「线程归属」问题：
- 订阅动作在哪条线程执行：
subscribe()调用、上游的onSubscribe、以及后续request(n)的传播。 - 信号在哪条线程到达下游：
onNext/onError/onComplete的执行线程。
默认两者都在调用 subscribe() 的线程上：
Flux.just(1, 2, 3)
.map(i -> i * 2)
.subscribe(System.out::println);
subscribeOn 管第一个，publishOn 管第二个。它们不是「把整条流搬到某个线程池」的开关，作用范围完全不同。
| 操作符 | 控制 | 影响范围 |
|---|---|---|
subscribeOn | 订阅与 request 的传播 | 整条链的上游（订阅方向） |
publishOn | 下游收到的信号 | 它之后的所有操作符 |
subscribeOn：在哪个调度器上订阅
语义
Flux.just(1, 2, 3)
.subscribeOn(Schedulers.boundedElastic())
.subscribe(System.out::println);
subscribeOn 把「向上游发起订阅」这个动作提交到指定调度器。对于同步源（Flux.just、Flux.range），上游的 onSubscribe 和 onNext 都在该调度器线程上发生；request(n) 也从下游经同一个 worker 转发给上游。
关键点是订阅方向：subscribeOn 影响的是它上游的订阅过程，而不是它下游的执行。下游操作符要不要换线程，由 publishOn 决定。
手写 SubscribeOnOperator
package com.plume.eventd.core;
import java.util.concurrent.Executor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
/**
* 把订阅动作和 request/cancel 的传播提交到 executor。
* executor 必须是串行执行的（单线程或串行化队列），否则 request 可能并发。
*/
public final class SubscribeOnOperator<T> implements Publisher<T> {
private final Publisher<T> upstream;
private final Executor executor;
public SubscribeOnOperator(Publisher<T> upstream, Executor executor) {
this.upstream = upstream;
this.executor = executor;
}
@Override
public void subscribe(Subscriber<? super T> downstream) {
executor.execute(() -> upstream.subscribe(
new SubscribeOnSubscriber<>(downstream, executor)));
}
private static final class SubscribeOnSubscriber<T> implements Subscriber<T> {
private final Subscriber<? super T> downstream;
private final Executor executor;
private Subscription upstream;
SubscribeOnSubscriber(Subscriber<? super T> downstream, Executor executor) {
this.downstream = downstream;
this.executor = executor;
}
@Override
public void onSubscribe(Subscription subscription) {
this.upstream = subscription;
downstream.onSubscribe(new Subscription() {
@Override
public void request(long n) {
// request 也回到订阅线程，与上游的发射保持串行
executor.execute(() -> subscription.request(n));
}
@Override
public void cancel() {
executor.execute(subscription::cancel);
}
});
}
@Override public void onNext(T item) { downstream.onNext(item); }
@Override public void onError(Throwable t) { downstream.onError(t); }
@Override public void onComplete() { downstream.onComplete(); }
}
}
executor 必须是串行的。如果传入线程池，两次 request 可能并发进入上游，违反 §2.7（Subscriber 对 request/cancel 的调用必须串行）。Reactor 的实现不用「提交任务」这种方式，而是用 Scheduler.Worker 的 schedule 串行执行，并把 request 累积后统一转发。
多次 subscribeOn：靠源最近的那个生效
Flux.just(1, 2, 3)
.subscribeOn(schedulerA) // 内层，靠近源
.map(i -> i * 2)
.subscribeOn(schedulerB) // 外层
.subscribe(System.out::println);
最终源的订阅发生在 schedulerA 上，不是 B。原因在调用方向：
- 最外层
subscribeOn(B)收到subscribe，把「订阅上游」提交给 B。 - 上游是
map，它继续向上游订阅，遇到subscribeOn(A)。 subscribeOn(A)把「订阅源」提交给 A。- 源最终在 A 的线程上被订阅。
所以是离源最近的 subscribeOn 决定源线程。说「第一个生效」时要注意是从源往下的第一个，不是代码里从左到右的第一个。想让不同阶段在不同线程，用 publishOn。
publishOn：在哪个调度器上发射
语义
Flux.just(1, 2, 3)
.map(i -> i * 2) // 订阅线程
.publishOn(Schedulers.boundedElastic())
.filter(i -> i > 2) // boundedElastic
.subscribe(System.out::println); // boundedElastic
publishOn 之后的 onNext/onError/onComplete 都在指定调度器上执行，之前的操作符不受影响。它只改变「信号交给下游时在哪个线程」，不会让单个流变成并行处理。
手写 PublishOnOperator
真实的 publishOn 不是「每个元素提交一个任务」——那会在多线程池上打乱顺序。它用队列 + WIP 计数器保证同一时刻只有一个 drain 在向 downstream 发信号。
package com.plume.eventd.core;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
public final class PublishOnOperator<T> implements Publisher<T> {
private final Publisher<T> upstream;
private final Executor executor;
public PublishOnOperator(Publisher<T> upstream, Executor executor) {
this.upstream = upstream;
this.executor = executor;
}
@Override
public void subscribe(Subscriber<? super T> downstream) {
upstream.subscribe(new PublishOnSubscriber<>(downstream, executor));
}
private static final class PublishOnSubscriber<T> implements Subscriber<T> {
private final Subscriber<? super T> downstream;
private final Executor executor;
private final Queue<T> queue = new ConcurrentLinkedQueue<>();
private final AtomicInteger wip = new AtomicInteger();
private volatile boolean done;
private volatile Throwable error;
private volatile boolean cancelled;
PublishOnSubscriber(Subscriber<? super T> downstream, Executor executor) {
this.downstream = downstream;
this.executor = executor;
}
@Override
public void onSubscribe(Subscription subscription) {
downstream.onSubscribe(new Subscription() {
@Override public void request(long n) { subscription.request(n); }
@Override public void cancel() {
cancelled = true;
subscription.cancel();
}
});
}
@Override
public void onNext(T item) {
if (done || cancelled) return;
queue.offer(item);
schedule();
}
@Override
public void onError(Throwable t) {
if (done) return;
error = t;
done = true;
schedule();
}
@Override
public void onComplete() {
if (done) return;
done = true;
schedule();
}
private void schedule() {
// 只有从 0 变 1 的线程负责 drain，其余线程只入队
if (wip.getAndIncrement() == 0) {
executor.execute(this::drain);
}
}
private void drain() {
int missed = 1;
for (;;) {
if (cancelled) return;
for (;;) {
T item = queue.poll();
if (item == null) break;
downstream.onNext(item);
}
if (done && queue.isEmpty()) {
Throwable e = error;
if (e != null) downstream.onError(e);
else downstream.onComplete();
return;
}
missed = wip.addAndGet(-missed);
if (missed == 0) return;
}
}
}
}
为什么不能「一个元素一个任务」
// 错误：多线程 Executor 下顺序不保证
onNext(T item) {
executor.execute(() -> downstream.onNext(item));
}
ThreadPoolExecutor 按提交顺序入队，但多个工作线程会并发取任务，元素 1 和元素 2 谁先执行没有保证。这直接违反 §1.3（信号必须串行）。改成单线程池能解决顺序，但会把所有 publishOn 串成一条线程，失去 Reactor 调度器的复用能力。
WIP 模式的做法是：所有线程都可以入队，但只有抢到 wip: 0 -> 1 的那个线程执行 drain，drain 内部循环取空队列；退出前用 missed 检查是否有新元素在 drain 期间入队。这是 Reactor、RxJava 内部反复出现的模式，和 2.2 的 emitting 是同一个思想。
三个线程归属实验
实验一：subscribeOn 只改订阅线程
AsyncRangePublisher source = new AsyncRangePublisher(
1, 4, Executors.newSingleThreadExecutor()); // 源自己的线程: pool-1-thread-1
SubscribeOnOperator<Integer> subscribeOn = new SubscribeOnOperator<>(
source, Executors.newSingleThreadExecutor()); // 订阅线程: pool-2-thread-1
subscribeOn.subscribe(subscriber);
输出（线程名以实际为准）：
onSubscribe thread: pool-2-thread-1
onNext 1 thread: pool-1-thread-1
onNext 2 thread: pool-1-thread-1
onNext 3 thread: pool-1-thread-1
onComplete thread: pool-1-thread-1
onSubscribe 在 subscribeOn 的线程；onNext 在源自己的线程。因为 AsyncRangePublisher 内部把发射提交到了自己的 executor，subscribeOn 管不到它。subscribeOn 只决定「订阅发生在哪」，不决定「源决定在哪发射」。
如果源是同步的（SyncSinglePublisher），onNext 会跟着订阅线程走，因为没有第二个线程参与。
实验二：publishOn 只改发射线程
PublishOnOperator<Integer> publishOn = new PublishOnOperator<>(
source, Executors.newSingleThreadExecutor()); // pool-2-thread-1
publishOn.subscribe(subscriber);
输出：
onSubscribe thread: main
onNext 1 thread: pool-2-thread-1
onNext 2 thread: pool-2-thread-1
onNext 3 thread: pool-2-thread-1
onComplete thread: pool-2-thread-1
onSubscribe 仍在调用 subscribe() 的主线程；onNext/onComplete 被搬到 publishOn 的线程。
实验三：组合时各管一段
PublishOnOperator<Integer> chain = new PublishOnOperator<>(
new MapOperator<>(new SubscribeOnOperator<>(source, executorA), i -> i * 2),
executorB);
- 订阅动作在
executorA上发生。 map在源的发射线程上执行（它在 publishOn 之前）。onNext到达下游时在executorB上。
常见错误是以为「publishOn 之后声明的操作符都在 B 上，所以 map 也在 B 上」。map 声明在 publishOn 之前，它在 B 的上游，不在 B 上执行。
四个常见误解
误解一：subscribeOn 放在末尾就没有效果
Flux.just(1, 2, 3)
.publishOn(Schedulers.boundedElastic())
.subscribeOn(Schedulers.single());
subscribeOn(single) 仍然把订阅动作放到 single 线程；只是下游看到的 onNext 由前面的 publishOn 决定，所以看不出 single 的痕迹。它不是「无效」，而是被 publishOn 的效果遮住了。
误解二：publishOn(parallel()) 会让计算并行
Schedulers.parallel() 有多个 worker，但一条 Flux 是顺序流。publishOn(parallel()) 只是把信号交给其中一个 worker，不会把一个 map 拆到多核。真正的并行需要 Flux.parallel() + runOn(Schedulers.parallel())，或者用 flatMap 并发订阅多个内层流。
误解三：boundedElastic 线程数无限
Reactor 3.8 的 boundedElastic 默认线程上限是 10 × CPU 核数（并有队列上限）。它适合包裹阻塞调用，但不是「随便阻塞」的许可证。把大量阻塞 map 放在 boundedElastic 上仍会耗尽线程。
正确做法是把每个阻塞调用交给独立的 Mono.fromCallable(...).subscribeOn(boundedElastic())，用 flatMap 控制并发度：
Flux.range(1, 100)
.flatMap(i -> Mono.fromCallable(() -> queryDatabase(i))
.subscribeOn(Schedulers.boundedElastic()), 8)
.subscribe();
误解四：Schedulers.from(executor) 可用
Reactor 3.8 没有 Schedulers.from(...)。要把自定义 Executor 包成调度器，用：
Scheduler scheduler = Schedulers.fromExecutor(uiExecutor);
// 或
Scheduler scheduler = Schedulers.fromExecutorService(executorService);
fromExecutor 会把任务直接交给 Executor 执行，因此它是否串行取决于 Executor 本身；UI 线程的 Executor 通常是串行的，适合这个场景。
选择指南
| 场景 | 用法 | 说明 |
|---|---|---|
| 阻塞 I/O（JDBC、文件） | subscribeOn(boundedElastic()) 或 Mono.fromCallable(...).subscribeOn(...) | 避免占用计算线程 |
| CPU 密集计算 | publishOn(parallel()) + 必要时 parallel().runOn(...) | 单靠 publishOn 不会并行 |
| 定时/延迟 | 默认 Schedulers.parallel() | Reactor 的定时器基于 parallel |
| 回到 UI 线程 | publishOn(Schedulers.fromExecutor(uiExecutor)) | 必须保证 UI Executor 串行 |
| 订阅不阻塞调用方 | subscribeOn(...) | 只影响订阅方向 |
小结
| 操作符 | 改变什么 | 不影响什么 |
|---|---|---|
subscribeOn | 订阅、onSubscribe、request 传播的线程 | 源自己决定在哪个线程发射 |
publishOn | 它下游收到的信号线程 | 它上游操作符的执行线程 |
两个实现共用同一个并发思想：所有线程都可以投递，但只有一个线程负责 drain。subscribeOn 用串行 Executor 简化了这个问题，publishOn 用队列 + WIP 计数器在任意 Executor 上保证串行。
前三章到此结束：规范、手写 Publisher/Subscriber/Processor、操作符链、线程切换。下一章进入 Project Reactor，用框架能力解决这些已经理解原理的问题。
阅读导航
下一篇：4.1 Mono 与 Flux：零个、一个与多个元素
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。