1.2 Reactive Streams 规范：四个接口

规范的定位

Reactive Streams 不是框架，不是库，也不是调度算法。它是一份接口契约：四个 Java 接口，加上 43 条交互规则。任何实现这些接口的库都能在类型层面互相连接——Reactor 的 Flux 可以交给 RxJava 的操作符，Akka Streams 的 Source 可以被 Reactor 消费。

规范有两个 Java 绑定：

• org.reactivestreams：独立 artifact，版本 1.0.4，Reactor、RxJava 3、Akka 都实现它。
• java.util.concurrent.Flow：JDK 9 起内置，接口形状与前者一致。

本书的手写实现统一使用 org.reactivestreams，原因很实际：第 11 章的 TCK 只接受这个包的类型，第 4 章之后对接的 Reactor subscribe 也只接受 org.reactivestreams.Subscriber。JDK Flow 在语义上等价，但类型不同，不能直接互传——需要一层很薄的适配器（本节末尾给出）。

四个接口

Publisher

package org.reactivestreams;

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> subscriber);
}

职责：数据生产者。只有一个方法，接受一个 Subscriber 的订阅。

关键点：

1. 订阅是惰性的：创建 Publisher 不产生数据，subscribe 被调用后才开始。
2. 可以被多次订阅：每次订阅是否共享数据流，由实现决定（Cold vs Hot，第 8 章展开）。
3. 不直接和 Subscriber 交互：速率协商全部通过 Subscription 进行。

Subscriber

public interface Subscriber<T> {
    void onSubscribe(Subscription subscription);
    void onNext(T item);
    void onError(Throwable throwable);
    void onComplete();
}

职责：数据消费者。四个回调构成完整的信号契约。

Subscription

public interface Subscription {
    void request(long n);
    void cancel();
}

职责：Publisher 与 Subscriber 之间的速率协商通道，也是整个规范的核心创新。

request(n) 是累加的：request(10) 之后再 request(5)，等价于总共请求 15 个。Subscriber 可以分批请求，不必在 onSubscribe 时一次性决定总量。n <= 0 是非法值，Publisher 必须以 onError(IllegalArgumentException) 结束（§3.9），而不是从 request 里抛异常（§3.16）。

Processor<T, R>

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

职责：既是 Subscriber 又是 Publisher。空接口，语义完全由两个父接口继承。它必须同时满足两套契约（§4.1）；如果选择恢复上游的 onError，必须视上游 Subscription 已取消，否则要立即向下游传播错误（§4.2）。

Processor 的典型用途是转换（map/filter 的内部实现）、桥接（不同实现之间）和副作用（日志、指标）。它不是必需品——直接写一个 Subscriber 再手动向下游发数据也能完成同样的事；它的价值在于让框架把管道中的每一段统一当作 Processor 处理。

订阅生命周期的完整时序

Subscriber                      Publisher
    |                               |
    |-------- subscribe() --------->|
    |<------- onSubscribe(sub) ----|
    |                               |
    |-------- request(3) --------->|
    |<------- onNext(A) -----------|
    |<------- onNext(B) -----------|
    |<------- onNext(C) -----------|
    |                               |
    |-------- request(2) --------->|
    |<------- onNext(D) -----------|
    |<------- onNext(E) -----------|
    |                               |
    |<------- onComplete() --------|

中途取消：

Subscriber                      Publisher
    |-------- subscribe() --------->|
    |<------- onSubscribe(sub) ----|
    |-------- request(5) --------->|
    |<------- onNext(A) -----------|
    |                               |
    |-------- cancel() ------------>|
    |                               | (最终停止发射)
    |        (可能还有已请求的 onNext 在路上)

注意最后一句。cancel() 不保证立即生效（§3.12），如果此前已经请求了 5 个而只收到 1 个，剩余 4 个仍可能在 cancel 之后到达（§2.8）。Subscriber 必须能容忍这种「迟到的元素」。

四个高频误解

误解一：onSubscribe 必须在 subscribe 返回前调用

规范 §1.9 的原文是：subscribe 必须在任何其他信号之前调用 onSubscribe，并且必须正常返回。它没有要求 onSubscribe 必须在 subscribe 返回之前完成。

也就是说，异步调用 onSubscribe 是合法的：

@Override
public void subscribe(Subscriber<? super T> s) {
    executor.execute(() -> s.onSubscribe(new MySubscription(s)));
    // 这里正常返回，onSubscribe 稍后到达
}

只要最终调用、且先于任何 onNext/onComplete/onError 即可。实践中的同步实现几乎都在 subscribe 里直接调用，但不要把「常见」当成「必须」。

误解二：同一个 Subscriber 重复订阅可以抛 IllegalStateException

§1.9 把 subscribe 的失败出口限定得很死：只有 subscriber == null 时抛 NullPointerException；其他任何拒绝（资源耗尽、已终止、太忙）都必须先 onSubscribe、再 onError，且 subscribe 本身正常返回。

同一个 Subscriber 被再次订阅时，§1.10 要求每次 subscribe 使用不同的 Subscriber；§2.12 要求同一个 Subscriber 的 onSubscribe 最多一次。真正的处理责任在 Subscriber 侧：如果它已经有一个活动 Subscription，又收到新的 onSubscribe，必须 cancel 新来的这个（§2.5）。Publisher 不能靠抛 IllegalStateException 把问题推回去。

误解三：null 元素会被 Publisher 转成 onError 信号

规范没有「Publisher 检测到 null 就发 onError」这样的条款。实际约束在 §2.13：调用 onSubscribe/onNext/onError/onComplete 时，任何参数为 null，被调用方必须向调用者抛 NullPointerException。也就是说，onNext(null) 会让 Subscriber 抛 NPE，该订阅随后视为取消。

因此实现 Publisher 时，正确做法是在发射前拒绝 null（通常直接 Objects.requireNonNull），而不是把 null 塞给下游。Reactor 的 Flux 在装配期就禁止 null，报错信息是 Spec.REQUIRE_NON_NULL，本质就是这条规则。

误解四：request/cancel 可以从任意线程随意调用

§3.1 规定 request/cancel 只能在该 Subscriber 的上下文中调用；§2.7 进一步要求同一个 Subscriber 对这些方法的调用必须串行。多线程调用并非绝对禁止，但调用方必须自己保证不并发——典型做法是加锁，或把调用收敛到单个线程。

Subscription 的实现则必须做到：request、cancel 及时返回，不在调用线程同步做重计算（§3.4、§3.5）；cancel 后 request 与再次 cancel 都是 no-op（§3.6、§3.7）；cancel 不能把异常抛给调用者（§3.15）。这两侧的责任不要混淆：Subscriber 负责「按 §2.7 串行地调用」，Subscription 负责「把这串调用安全地落到状态机上」。

org.reactivestreams 与 JDK Flow 的互操作

两个包的接口方法签名看起来一样，但它们是不同的类型，没有继承关系。下面这段代码无法编译：

Flow.Publisher<String> flowPublisher = ...;
org.reactivestreams.Publisher<String> rsPublisher = flowPublisher; // 编译错误

JDK 本身没有提供适配器，但 org.reactivestreams:reactive-streams:1.0.4 自带 org.reactivestreams.FlowAdapters，双向转换都是一行：

import java.util.concurrent.Flow;
import org.reactivestreams.FlowAdapters;

Flow.Publisher<String> flowPublisher = ...;
org.reactivestreams.Publisher<String> rsPublisher =
        FlowAdapters.toPublisher(flowPublisher);       // Flow → RS

org.reactivestreams.Publisher<String> rsPublisher2 = ...;
Flow.Publisher<String> flowPublisher2 =
        FlowAdapters.toFlowPublisher(rsPublisher2);    // RS → Flow

Subscriber 和 Processor 也有对应的 toFlowSubscriber/toSubscriber、toFlowProcessor/toProcessor。Subscription 没有公开的转换方法：适配器在转发 onSubscribe 时会在内部包装，外部拿到的始终是目标包里的类型。生产代码优先用这些现成适配器；下面手写一遍 Flow → RS 的转发，是为了说明它们为什么只能逐层包装、无法靠强制类型转换完成：

import java.util.concurrent.Flow;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public final class FlowToRsPublisher<T> implements Publisher<T> {

    private final Flow.Publisher<T> source;

    public FlowToRsPublisher(Flow.Publisher<T> source) {
        this.source = source;
    }

    @Override
    public void subscribe(Subscriber<? super T> rsSubscriber) {
        source.subscribe(new Flow.Subscriber<T>() {
            @Override
            public void onSubscribe(Flow.Subscription fs) {
                rsSubscriber.onSubscribe(new Subscription() {
                    @Override public void request(long n) { fs.request(n); }
                    @Override public void cancel() { fs.cancel(); }
                });
            }
            @Override public void onNext(T item) { rsSubscriber.onNext(item); }
            @Override public void onError(Throwable t) { rsSubscriber.onError(t); }
            @Override public void onComplete() { rsSubscriber.onComplete(); }
        });
    }
}

反方向（RS → Flow）完全对称，把两个接口的角色对调即可。适配器不改变语义，只是把调用原样转发；因此上游是否合规，决定了适配后是否合规。

为什么 request(n) 是核心

传统回调 API（例如 CompletableFuture.thenAccept）是纯推模型：数据产生就推给下游，下游无法表达「我暂时处理不过来」。Reactive Streams 用 request(n) 把「消费能力」变成一条显式信号，生产者只能按额度发射。

这带来一个设计问题：Subscriber 应该怎么请求？

Reactor 的 limitRate(n) 就是批量策略的封装：它把下游的请求切成不超过 n 的批次，并在消耗到 75% 时补发下一批。第 7 章会结合源码拆解这个行为。

小结

本卷的实现统一用 org.reactivestreams；需要和 JDK Flow 互通时用本节的适配器。下一节把 43 条规则逐条展开——先记住四个接口的形状，规则才有落点。

阅读导航

上一篇：1.1 异步编程的演进

下一篇：1.3 规范规则逐条分析