1.3 规范规则逐条分析

先纠正一个常见误解

Reactive Streams 1.0.4 的规范正文里共有 43 条规则，按接口分四节：

网上不少二手资料把它压缩成「15 条」，并且把 §2 写成 Subscription、§3 写成 Subscriber。编号一旦错位，后面引用 TCK 测试名（required_spec2xx_* 对应 Subscriber、required_spec3xx_* 对应 Subscription）就全对不上。本节按官方编号来。

43 条规则速查表

下表是逐条摘要，用于查阅；每条的完整原文以官方文本为准。真正容易写错的几条，后面单独拆。

§1 Publisher（1.1–1.11）

§2 Subscriber（2.1–2.13）

§3 Subscription（3.1–3.17）

§4 Processor（4.1–4.2）

六个真正容易写错的点

规则表看起来只是「背下来」，但实现时翻车集中在下面六处。

1.1 是「任何时刻」的不变量，不是最终统计

规则原文强调 at all times，而且官方解释里点明了一个隐含的 happens-before：需求必须先被收到，元素才可能发出。这带来两个实践后果：

• 先 onNext 再记账，即使最终数量对得上，中途也已经违规。发布者必须「先扣额度，再发射」。
• 累加需求时不能简单 requested += n。Long.MAX_VALUE 再加就会溢出成负数，之后的 requested > 0 判断全部失效。正确做法是饱和加法：

private long requested;      // 只在串行化保护下访问

void addDemand(long n) {
    long current = requested;
    long updated = current + n;
    // 溢出或达到上限，都按「无界」处理（对应 3.17）
    requested = updated < 0 || updated == Long.MAX_VALUE
            ? Long.MAX_VALUE
            : updated;
}

1.9 / 1.10 / 2.5 / 2.12：一次订阅到底允许发生什么

这四条要连起来读，它们描述的是同一件事的不同侧面：

• subscribe 的第一个动作是 onSubscribe，并且在正常路径上必须正常返回。如果 Publisher 太忙、资源耗尽或者已经终止，唯一合法的拒绝方式是「先 onSubscribe，再 onError」，不能从 subscribe 里往外抛 IllegalStateException。
• 只有一种情况可以抛异常：传入的 Subscriber 是 null，此时抛 NullPointerException，因为没有任何对象可以承载 onError。
• 同一个 Subscriber 对象不允许被订阅两次（1.10 + 2.12）。如果它真的又收到了一个 onSubscribe，责任在 Subscriber 一侧：必须 cancel 新来的这个（2.5），避免两个 Publisher 同时向它发数据。

一个容易忽略的细节：2.12 的「同一个」按 equals 判定，不是按引用。用值对象当 Subscriber（很少见但合法）时，equals 相同就算同一个。

2.3：终止回调里什么都别碰

onComplete/onError 到达时，Subscription 在这些语义上已经结束了（2.4、1.6）。此时再调用 request 或 cancel，规范明确禁止。常见错误写法：

@Override
public void onComplete() {
    // 错误：终止回调里调用 Subscription 方法，违反 2.3
    subscription.cancel();
    latch.countDown();
}

需要释放资源时，正确的位置是在 onComplete/onError 之前由业务方自行管理，或者用 doFinally 这类「不碰 Subscription」的钩子（Reactor 的 doFinally 在终止之后执行，且不再回调 Subscription）。不要在终止回调里试图「补救性地 cancel」。

顺带说明一个常见疑问：onComplete 里发日志、countDown、关闭自己的缓冲区都没问题，被禁止的是调用 Subscription 或 Publisher 的方法。

2.8 / 2.9 / 2.10：Subscriber 必须接受「不按剧本」的信号

三条规则都在提醒 Subscriber 不要写过于乐观的假设：

• 2.8：cancel() 只是「请求停止」，不是「立即停止」。如果之前已经 request 了 100 个、只收到 40 个，cancel 之后仍可能再收到若干 onNext。所以 cancel() 之后的状态清理必须能容忍迟到的元素，不能把缓冲区直接置 null 然后 dereference。
• 2.9 / 2.10：onComplete 和 onError 可以在没有 request 过的情况下到达。空流、订阅即失败、上游提前终止都属于这种情况。Subscriber 不能把「没 request 过」当成「不会收到终止信号」。
• 反过来，2.8 也不要求 Publisher 在 cancel 后补发终止信号。规范只说「最终停止发信号」（1.8），并没有「cancel 后必须收到 onComplete/onError」这一条。把 1.8 读成「必须补一个终止信号」是二手资料里最常见的错误。

3.2 / 3.3：同步 request 合法，但不能无限递归

很多人以为「request 里直接同步 onNext」是坏味道。规范其实明确允许（3.10、3.11），并且要求 request 必须支持从 onNext/onSubscribe 里同步调用（3.2）。

危险在于互相递归：onNext -> request -> onNext -> request -> ...。3.3 要求实现给这个递归设上限，官方建议深度 1。做法是「记账 + 循环泵」，而不是在 request 里直接递归回调：

@Override
public void request(long n) {
    if (n <= 0) {
        subscriber.onError(new IllegalArgumentException("non-positive request: " + n));
        return;
    }
    addDemand(n);
    drain();                 // 单次进入，内部用循环消费需求
}

private void drain() {
    if (!wip.compareAndSet(0, 1)) return;   // 已有线程在 drain
    int missed = 1;
    for (;;) {
        long r = requested;
        long emitted = 0;
        while (emitted != r && !cancelled && hasNext()) {
            subscriber.onNext(next());
            emitted++;
        }
        if (emitted != 0) {
            // 用 CAS 扣减，避免与并发 request 相互覆盖
            long next;
            do {
                r = requested;
                next = r - emitted;
            } while (!requestedUpdater.compareAndSet(this, r, next));
        }
        missed = wip.addAndGet(-missed);
        if (missed == 0) return;
    }
}

这段「wip + missed」是 Reactor、RxJava 内部到处出现的固定模式：用一个原子计数保证同一时刻只有一个线程在发信号（满足 1.3 串行），同时不丢失并发进入的请求（满足 3.8 累加）。

3.9 / 3.17：请求的上下边界

request(0) 和 request(-1) 没有语义，3.9 要求以 onError(IllegalArgumentException) 结束。注意是 onError，不是从 request 里抛异常——3.16 明确要求 request 正常返回。

上界则是 Long.MAX_VALUE。3.17 允许 Publisher 把「≥ Long.MAX_VALUE」视为无界，但必须能接受任意次 request 的累加。实现里通常用一个 volatile long requested 加饱和加法；把所有 request 加总后判断「是否 ≥ 元素总数」是错的，因为 1.1 检查的是任意时刻。

用 TCK 验证这些规则

TCK 的方法名直接编码了规则号，命名规律是 required_specXYZ_*，其中 XYZ 是 节号 + 两位规则号。下面是 PublisherVerification 里与上面六点直接相关的真实方法（完整清单见第 11 章）：

Subscriber 一侧的规则（2.3、2.5、2.8–2.10、2.13）由 SubscriberWhiteboxVerification / SubscriberBlackboxVerification 覆盖，例如 required_spec203_mustNotCallMethodsOnSubscriptionOrPublisherInOnComplete、required_spec205_mustCallSubscriptionCancelIfItAlreadyHasAnSubscriptionAndReceivesAnotherOnSubscribeSignal、required_spec208_mustBePreparedToReceiveOnNextSignalsAfterHavingCalledSubscriptionCancel。这些类基于 TestNG，不是 JUnit 4——第 11 章会实际跑一遍。

违反规则的代价

规范里的规则不是为了考试，每一条都对应一类生产事故：

• 违反 1.1 会击穿下游的容量假设：限流的连接池、固定大小的缓冲区、按批提交的事务，都会在「多发一个」时出错。
• 违反 2.3 会在终止路径上制造环和竞态：Publisher、Subscription、Subscriber 三方在「谁先收尾」上互相回调，轻则重复释放，重则卡死。
• 违反 3.5（在 cancel 里同步做重计算）会拖住取消线程：超时线程迟迟回不来，事件循环被阻塞，取消失去了「及时止损」的意义；违反 3.7（重复 cancel 不幂等）则可能导致引用计数被减两次。
• 违反 3.9 会让 request(-1) 被当成 Long.MAX_VALUE，一次把整个数据集拉进内存。
• 违反 3.17 的溢出判断，会在累计需求超过 Long.MAX_VALUE 后变成负数，发布者认为「没有需求」而永久挂起。

小结

记忆这 43 条规则时，可以按「谁负责什么」来分层：

1. Publisher 负责不超发、不早发、终止后不再发（1.1、1.3、1.7、1.8）。
2. Subscriber 负责表达需求、串行调用、接受迟到的信号（2.1、2.7、2.8–2.10）。
3. Subscription 负责把 request/cancel 做成可累加、取消后 no-op、非法请求能正确报错的边界操作（3.6–3.9、3.17）。
4. Processor 同时扮演两边，终止传播不能两头都想要（4.1、4.2）。

下一章开始手写实现。建议边写边对照本节速查表——每写一个 request 分支，就问一句「这行代码在满足哪条规则，或者违反了哪条」。

阅读导航

上一篇：1.2 Reactive Streams 规范：四个接口

下一篇：2.1 同步 Publisher 最小实现