附录 C：背压规则清单

规则全集：43 条，分四组

Reactive Streams 1.0.4 的规范正文一共 43 条，分组如下：

两个常见错误先纠正：

1. §2 是 Subscriber，§3 是 Subscription。旧版把两者对调，导致 request(0)、cancel 之后的行为被归到了错误的角色上。
2. 不是 15 条，也不存在「9 条与背压直接相关」这种官方口径。43 条里很多规则都在约束背压链路，硬切一个「9 条」既没有出处，也会漏掉 3.8、3.9、3.17 这些关键条款。

本附录不要求你背下 43 条。先抓住 request → demand → onNext 这条主链，再按角色查表。

背压主链

Subscriber.onSubscribe(subscription)
        │
        ├── subscription.request(n)          §2.1
        │        │
        │        └── Subscription 累加 demand  §3.8
        │                 │
        │                 └── Publisher 最多发 demand 个 onNext  §1.1
        │
        └── subscription.cancel()             §3.6 / §3.7
                 │
                 └── Publisher 最终停止发信号   §1.8 / §3.12

这条链上有五个关键点：

Flux.range 这类同步源之所以“天然有背压”，不是因为它特殊，而是因为它把 §3.8 的累加和 §1.1 的上限实现对了。手写 Publisher 出错，通常也错在这两条。

§1 Publisher：11 条

几个容易误解的点：

• 1.1 不是「发了 N 个就一定要等下一个 request」。如果一次 request(10)，Publisher 可以连续发 10 个，不必每个元素之间插入一次 request。
• 1.6 与 1.7 是两道不同的门。1.6 说终止后 Subscription 失效，1.7 说终止后不能再有信号；前者约束状态，后者约束信号。
• 1.9 的「返回正常」指不向调用方抛异常。内部错误必须转成 onError，而不是让 subscribe 抛出。

§2 Subscriber：13 条

实践中最常踩的是 2.8：cancel() 返回不代表立刻安全。如果 cancel 之前已经 request(100)，Publisher 可能还有元素在路上。Subscriber 的 onNext 必须能容忍“取消后仍被调用几次”，不能一 cancel 就把状态清空到必然 NPE。

§3 Subscription：17 条

3.8 与 3.9 是手写 Subscription 最容易错的地方。一个正确的累加应该是饱和加法：

void request(long n) {
    if (n <= 0) {
        subscriber.onError(new IllegalArgumentException("§3.9: " + n));
        return;
    }
    for (;;) {
        long current = requested.get();
        long next = current + n;
        if (next < 0) next = Long.MAX_VALUE;   // 饱和，不能溢出成负数
        if (requested.compareAndSet(current, next)) return;
    }
}

写成 requested.addAndGet(n) 会在累计值接近 Long.MAX_VALUE 时回绕成负数，下一次判断 requested > 0 就失效，表现为“订阅突然不再发元素”。

§4 Processor：2 条

4.1 的含义不是“实现两个接口就自动合规”，而是两条链都要各自满足：作为 Subscriber 要正确 request，作为 Publisher 要正确结算下游 demand。PublishProcessor 在 TCK 的 IdentityProcessorVerification 中出现大量失败，根源通常就在这里——它把上游和下游当成了同一个 demand 池。

4.2 只在 Processor 主动“吞掉错误并继续”时生效。普通转发型 Processor 不应该恢复错误，直接把 onError 往下传即可。

Reactor 的背压操作符

这些 onBackpressure* 操作符有一个共同点：它们会向上游请求 Long.MAX_VALUE，把上游变成“推”模型，然后由自己在中间结算下游需求。它们不是“把背压传回上游”的工具，而是“上游不会背压时，在中间做有损/有界处理”的工具。

两参数重载不是“溢出即丢弃”

flux.onBackpressureBuffer(50, dropped::incrementAndGet)

这里的 dropped 是溢出回调，默认策略是 BufferOverflowStrategy.ERROR。缓冲区满时它会调用一次回调，然后以 OverflowException 终止整条流，后面的元素不会再产生。要丢弃并继续，必须写三参：

flux.onBackpressureBuffer(
    50,
    dropped::incrementAndGet,
    BufferOverflowStrategy.DROP_LATEST);

实测（Flux.range(1, 1000) + publishOn(..., 1) + 每次只 request 1）：

error: OverflowException: The receiver is overrun by more signals than expected
delivered=10 overflowCallback=1

容量 10，交付 10 个后溢出，回调触发 1 次，流终止。这个例子同时说明：onBackpressureBuffer 会向上游请求无界需求，Flux.range 的 1000 个元素会一次性推进缓冲区；如果没有 publishOn 这样的异步边界，下游在 onNext 里同步补充 request，缓冲反而可能永远不满，测不出溢出。

delayElements 不是背压缓冲边界

旧版附录给过这样的例子：

Flux.interval(Duration.ofMillis(10))
    .onBackpressureDrop(i -> metrics.counter("dropped").increment())
    .delayElements(Duration.ofMillis(100))
    .subscribe(i -> dashboard.update(i));

看起来是“10ms 生产、100ms 消费，所以中间会丢弃”。但 delayElements(d) 在 Reactor 3.8 里的实现是 delayUntil(x -> Mono.delay(d))，它一次只向上游请求一个元素。丢弃确实会发生，但原因是它自己的 100ms 节流，不是下游消费者慢。实测对比：

fast consumer (no sleep)      delivered=3 drops=27
slow consumer (80ms sleep)    delivered=2 drops=28

消费者睡不睡几乎不影响结果，因为瓶颈是 delayElements 的 100ms，而不是消费者的 80ms。用这条链演示“消费者驱动背压”会得出错误结论。

要观察真正的下游需求，应该显式控制 request：

Flux.interval(Duration.ofMillis(10))
    .onBackpressureDrop(dropped::incrementAndGet)
    .publishOn(Schedulers.boundedElastic(), 1)
    .subscribe(new BaseSubscriber<Long>() {
        @Override protected void hookOnSubscribe(Subscription s) {
            request(1);
        }
        @Override protected void hookOnNext(Long v) {
            sleep(50);
            request(1);
        }
    });

publishOn(..., 1) 把预取压到 1，需求才会及时回传；默认预取 256 会把慢消费掩盖在队列里。

与 TCK 的对应关系

TCK 的方法名就是规则编号，排查失败时可以直接定位：

PublisherVerification 这个名字容易让人以为它只测 §1。实际上它同时包含 §1 的 11 条和 §3 的 17 条中可测的部分，共 22 个 required_* 方法。§2 的验证在 SubscriberWhiteboxVerification（14 个 required）和 SubscriberBlackboxVerification（11 个 required）里。§4 只由 IdentityProcessorVerification 覆盖，且源码只引用 4.1，4.2 没有对应的 required 测试。

自检清单

实现自定义 Publisher 时，按这个顺序检查：

小结

阅读导航

上一篇：附录 B 调度器选择指南

下一篇：附录 D 从 CompletableFuture 到响应式流迁移指南