先划清范围
本节的「重写」指的是：把第 2、3、13 章为了讲解而手写的示例，换成 Reactor 中语义等价的写法。
仓库里真实存在的只有 SyncSinglePublisher 和 LoggingSubscriber；AsyncRangePublisher、SensorSimulatorPublisher、BackpressureDashboard 都是章节中的示例代码。因此下面的对比不是「同一个仓库里两套生产代码」，而是「教学实现 vs 框架实现」。
SyncSinglePublisher → Mono.just
手写版的核心是 71 行里的这段：
public void request(long n) {
if (cancelled || completed) return;
if (n <= 0) {
cancelled = true;
subscriber.onError(new IllegalArgumentException("§3.9: " + n));
return;
}
completed = true;
if (value != null) subscriber.onNext(value);
if (!cancelled) subscriber.onComplete();
}
Reactor 写法：
Mono<Integer> mono = Mono.just(1);
语义并不完全一样：
| 场景 | SyncSinglePublisher | Mono.just |
|---|---|---|
value == null | 不发射元素，直接 onComplete | 构造时抛 NullPointerException |
| 需要空流 | 只能传 null | 用 Mono.empty() 表达意图 |
request(0)（raw Subscription） | onError(IllegalArgumentException) | onError(IllegalArgumentException) |
多次 request | 第二次空操作 | 单值发完后同样不会重复发射 |
| 取消后请求 | 无信号 | 无信号 |
| 接口类型 | java.util.concurrent.Flow | org.reactivestreams |
Mono.just(null) 的行为实测：
Mono.just(null): NullPointerException
这是一个容易被「等价」二字掩盖的差异。手写版用 if (value != null) 把 null 变成了空流；Mono.just 选择在装配阶段就失败，迫使调用方显式写 Mono.empty() 或 Mono.justOrEmpty(...)。后者更好，因为「空」是业务语义，不该靠 null 碰巧实现。
request(0) 的两个层面
用原始 Subscriber 调用 subscription.request(0)，Mono.just 和 Flux.range 都会报 §3.9：
Mono.just(1).subscribe(new Subscriber<Integer>() {
public void onSubscribe(Subscription s) { s.request(0); }
// ...
});
实测：
mono onSubscribe
mono error java.lang.IllegalArgumentException: §3.9 violated: positive request amount required but it was 0
但如果你用的是 BaseSubscriber：
new BaseSubscriber<Integer>() {
@Override protected void hookOnSubscribe(Subscription s) {
request(0);
}
};
则什么都不会发生——没有元素，也没有错误。原因是 BaseSubscriber.request(long) 先调用 Operators.validate(n)，非正数直接被过滤，不会转发给上游。排查「为什么没报错」时要先确认自己用的是哪一层 API。
AsyncRangePublisher → Flux.range + subscribeOn
手写版要自己维护 AtomicLong requested、AtomicBoolean cancelled、AtomicBoolean emitting 和一个 emitLoop()。Reactor 写法：
Flux<Integer> flux = Flux.range(from, count)
.subscribeOn(Schedulers.boundedElastic());
两个参数的含义要先对齐：Flux.range(start, count) 的第二个参数是元素个数，不是结束值。手写版如果约定 [from, to)，对应的写法是 Flux.range(from, to - from)；写成 Flux.range(from, to) 会多发射 to - from 个元素。
subscribeOn 只改变订阅动作和上游发射发生的线程，不改变下游处理线程。下面这条链：
Flux.range(1, 3)
.subscribeOn(Schedulers.boundedElastic())
.map(i -> {
System.out.println("map on " + Thread.currentThread().getName());
return i;
})
.subscribe(i -> System.out.println("next on " + Thread.currentThread().getName()));
map 和 subscribe 都在 boundedElastic 线程执行，因为中间没有 publishOn。如果下游处理很慢，它会阻塞这个调度器线程；想让「上游生产」和「下游处理」分开，需要再加 publishOn。
另一个差异是并发模型：Flux.range 的发射是串行的，内部已经处理了 request 累加和取消竞争。手写版如果没有 wip 或锁，多个线程同时 request 可能并发射出，直接违反 §1.3。这正是「框架经过 TCK 验证」的实际含义。
SensorSimulatorPublisher → Flux.create + sink.onRequest
先看一个错误的等价写法：
Flux.create(sink -> {
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(
() -> sink.next(20 + ThreadLocalRandom.current().nextGaussian() * 5),
0, 100, TimeUnit.MILLISECONDS);
sink.onCancel(scheduler::shutdown);
}, FluxSink.OverflowStrategy.BUFFER);
它确实每 100ms 产生一个值，但完全忽略下游请求量。下游只请求 1 个时，其余元素进入 Flux.create 的缓冲区；请求 Long.MAX_VALUE 时看起来正常，请求很慢时则变成无界缓冲。这和 13.2 手写版「严格按 request 发射」的语义不同。
正确的映射是 sink.onRequest：
double lambda = 10;
Flux<Double> sensor = Flux.create(sink -> {
ScheduledExecutorService scheduler =
Executors.newSingleThreadScheduledExecutor();
sink.onRequest(n -> {
for (long i = 0; i < n && !sink.isCancelled(); i++) {
double u = ThreadLocalRandom.current().nextDouble();
long delay = Math.max(0L,
(long) (-Math.log(u == 0 ? Double.MIN_VALUE : u) / lambda * 1000));
scheduler.schedule(
() -> sink.next(20 + ThreadLocalRandom.current().nextGaussian() * 5),
delay, TimeUnit.MILLISECONDS);
}
});
sink.onDispose(() -> scheduler.shutdownNow());
}, FluxSink.OverflowStrategy.ERROR);
它和手写版的差别：
| 关注点 | 手写 SensorSimulatorPublisher | Flux.create + onRequest |
|---|---|---|
| 需求累加 | 自己用 AtomicLong 饱和相加 | Reactor 内部记录 request |
| 并发 request | 自己用 wip 串行化 | FluxSink 保证信号串行 |
| 取消 | 置位并中断 worker | sink.isCancelled() + onDispose |
| 延迟调度 | 在 drain 线程里 sleep | 每个元素一个定时任务 |
| 无界请求 | 循环受 Long.MAX_VALUE 限制，实际永远跑 | 会尝试调度海量任务，这是缺陷 |
最后一行是关键：onRequest(Long.MAX_VALUE) 时，for (long i = 0; i < n; ...) 等价于无限循环。真实实现要把「请求量」转成「速率预算」或分批调度，不能为每个元素创建一个 ScheduledFuture。这段代码只适合有界请求的演示，13.2 已经说明了这个边界。
BackpressureDashboard 不需要「重写」
13.3 的面板本身就是 Reactor 链：
Flux.interval(Duration.ofMillis(10))
.onBackpressureBuffer(10, v -> dropped.incrementAndGet(),
BufferOverflowStrategy.DROP_LATEST)
.publishOn(Schedulers.boundedElastic(), 1)
.subscribe(new BaseSubscriber<Long>() {
@Override protected void hookOnSubscribe(Subscription s) {
request(1);
}
@Override protected void hookOnNext(Long v) {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
request(1);
}
});
它没有「手写 RS 版本」可对照。手写的是 SensorSimulatorPublisher 那种源，而 onBackpressureBuffer、publishOn 属于操作符与调度，本来就应该交给框架。把两者混在一张「重写前后」表里，只会制造不存在的对比。
行数不是衡量标准
旧版这一节曾列过「45 行 → 1 行、85 行 → 2 行」的缩减比例。这类数字不可靠：
- 手写版的行数取决于你处理了多少边界（§3.9、并发 request、取消、异常），不是固定值。
Mono.just(1)是一行，但背后是 Reactor 的MonoJust与Operators，出了问题仍要读源码。- 行数少不等于维护成本低：框架升级、调度器配置、丢弃策略、上下文传播都会带来新的学习成本。
更值得比较的是「谁负责哪条规则」：
| 责任 | 手写实现 | Reactor |
|---|---|---|
onSubscribe 只调用一次 | 自己保证 | 操作符保证 |
| request 累加与饱和 | 自己写 | 内部处理 |
| 信号串行 | 自己写 wip/锁 | 操作符保证 |
| 取消传播 | 自己检查标志 | 内置 |
| 终止后不再发信号 | 自己用 CAS | 内置 |
| 通过 TCK | 自己跑、自己修 | 官方已验证 |
手写的价值在于让你知道这些责任存在；生产的价值在于把这些责任交给经过验证的实现。两者不是替代关系。
何时仍然需要手写
| 场景 | 建议 |
|---|---|
| 学习 RS 规范 | 手写，但只写到能跑 TCK 的程度 |
标准源（just、range、interval） | 直接用 Reactor |
| 自定义异步源 | 优先 Flux.create / Mono.create，并处理 onRequest |
| 桥接非 RS 协议 | 写 Publisher 适配器，边界处必须做背压映射 |
| 框架行为不符合需求 | 先确认是不是用错了操作符，再考虑手写 |
| 生产环境 | 默认用框架；手写部分要单独跑 TCK |
小结
| 手写 | Reactor | 关键差异 |
|---|---|---|
SyncSinglePublisher | Mono.just | null 处理不同；接口类型不同 |
AsyncRangePublisher | Flux.range + subscribeOn | 第二个参数是数量；subscribeOn 不切下游 |
SensorSimulatorPublisher | Flux.create + sink.onRequest | 不能用 scheduleAtFixedRate 假装背压 |
手写 BackpressureDashboard | 不存在 | 面板本来就是 Reactor 链 |
| 手写 TCK 验证 | 官方操作符已验证 | 手写部分仍需自己跑 |
阅读导航
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。