java

2.3 Subscriber 实现与信号契约

实现三个有代表性的 Subscriber：LoggingSubscriber、逐元素请求的 CountingSubscriber 和批量消费的 BatchingSubscriber。按 §2.1（用 request 表达需求）、§2.7（request/cancel 串行）、§2.8（cancel 后可能有迟到元素）、§2.3（终止回调不得操作 Subscription）逐条落实，并指出「synchronized 不等于信号串行」和「onComplete 里再 request」两个真实 bug。

Leeting Yan 6 分钟阅读 2684 字

Subscriber 的职责

Subscriber 看起来只是「收数据」，但它承担四条独立责任：

  1. 表达需求：通过 request(n) 告诉上游能处理多少（§2.1）。
  2. 串行调用：对 request/cancel 的调用不能并发（§2.7）。
  3. 容忍边界：cancel 后可能还有迟到元素（§2.8），未 request 也可能收到终止信号（§2.9、§2.10）。
  4. 正确收尾：终止回调里不得再操作 Subscription 或 Publisher（§2.3）。

本节用三个实现分别覆盖这些点。

Subscriber用途主要规则
LoggingSubscriber调试输出§1.9、§2.1
CountingSubscriber逐元素请求、达到目标即取消§2.1、§2.7、§2.8
BatchingSubscriber批量缓冲后处理§2.3、§2.4、§2.7

LoggingSubscriber：最小可用实现

package com.plume.eventd.core;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public final class LoggingSubscriber<T> implements Subscriber<T> {

    private final String name;
    private Subscription subscription;

    public LoggingSubscriber(String name) {
        this.name = name;
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        // §2.5：已有活动订阅时，cancel 新来的这个并返回；
        // 旧订阅保持活动，不能反过来把旧的取消掉
        if (this.subscription != null) {
            subscription.cancel();
            return;
        }
        this.subscription = subscription;
        log("onSubscribe");
        subscription.request(Long.MAX_VALUE);   // §2.1：表达需求
    }

    @Override
    public void onNext(T item) {
        log("onNext: " + item);
    }

    @Override
    public void onError(Throwable throwable) {
        // §2.3：终止回调里不调用 subscription.cancel()
        log("onError: " + throwable);
    }

    @Override
    public void onComplete() {
        log("onComplete");
    }

    public void cancel() {
        if (subscription != null) {
            subscription.cancel();
        }
    }

    private void log(String message) {
        System.out.printf("[%s][%s] %s%n",
                Thread.currentThread().getName(), name, message);
    }
}

onSubscribe 里调用 request 是合法的

很多资料说「request 必须在 onSubscribe 返回后调用」，这是误解。§3.2 明确要求 Subscription 必须允许 Subscriber 从 onSubscribe 或 onNext 内部同步调用 request。同步 Publisher 正是靠这一点工作的：request 调用栈里直接发射元素。

真正需要担心的是无界递归（§3.3）：onNext -> request -> onNext -> ... 如果每一层都直接递归，栈会爆。规范建议把递归深度限制为 1，实现方式就是 2.2 里的「记账 + 循环泵」：request 只增加额度并尝试进入发射循环，如果已有线程在循环中就直接返回。

终止回调里什么不能做

§2.3 禁止在 onComplete/onError 中调用 Subscription 或 Publisher 的任何方法。下面两种写法都违规：

@Override
public void onComplete() {
    subscription.cancel();   // 违规：终止回调里操作 Subscription
}

@Override
public void onError(Throwable t) {
    publisher.subscribe(new RecoverySubscriber<>());  // 违规：操作 Publisher
}

要在终止时释放资源（关闭文件、归还连接、latch.countDown()），直接在回调里做即可，只要不碰 Subscription/Publisher。如果用的是 Reactor，doFinally 是更合适的钩子：它在终止信号之后执行，且不会再回调 Subscription。

CountingSubscriber：逐元素请求

public final class CountingSubscriber<T> implements Subscriber<T> {

    private Subscription subscription;
    private int count;
    private final int target;

    public CountingSubscriber(int target) {
        this.target = target;
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);          // 先要一个
    }

    @Override
    public void onNext(T item) {
        count++;
        if (count >= target) {
            subscription.cancel();        // 够了就取消（§2.6）
            return;
        }
        subscription.request(1);          // 处理完再要下一个
    }

    @Override
    public void onError(Throwable t) {
        System.out.println("Error: " + t);
    }

    @Override
    public void onComplete() {
        System.out.println("Complete, total: " + count);
    }
}

request(1) 放在 onNext 的哪个位置

位置效果风险
处理前提前要下一个，吞吐高处理慢时元素在内存里排队
处理后（本例）处理完才要，天然背压同步 Publisher 下有重入风险
另一个线程完全避开重入多一次调度，顺序需要额外保证

「处理后」版本的背压最直观：处理多慢，请求就多慢。重入风险由 Publisher 侧的 §3.3 兜底——合规的 Publisher 必须限制同步递归深度，不能因为 Subscriber 在 onNext 末尾调 request 就无限递归。

cancel 之后仍可能收到 onNext

cancel() 不是「立即静音」。§2.8 要求 Subscriber 准备好接收 cancel 之后、已经请求过的迟到元素；§3.12 也只要求 Publisher「最终」停止。所以 onNext 里 cancel 之后，计数器、缓冲区、状态机仍然要能安全处理后续回调，不能把字段置 null 后直接 dereference。

本节的 AsyncRangePublisher 在发射循环的每个元素前检查 cancelled，所以实际观察到的是「cancel 后立刻停止」。这是实现的优化，不是规范保证——换一个异步源（网络、消息队列）就可能出现迟到元素。

BatchingSubscriber：批量消费

import java.util.ArrayList;
import java.util.List;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public final class BatchingSubscriber<T> implements Subscriber<T> {

    private final List<T> buffer = new ArrayList<>();
    private final int batchSize;
    private Subscription subscription;
    private boolean done;

    public BatchingSubscriber(int batchSize) {
        this.batchSize = batchSize;
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(batchSize);      // 只请求一批
    }

    @Override
    public void onNext(T item) {
        // synchronized 保护 buffer 状态，但不负责「信号串行」
        synchronized (this) {
            if (done) return;
            buffer.add(item);
            if (buffer.size() < batchSize) {
                return;
            }
            List<T> batch = drain();
            process(batch);
        }
        // 在本批处理完之后再请求下一批
        subscription.request(batchSize);
    }

    @Override
    public void onError(Throwable t) {
        synchronized (this) {
            done = true;
            buffer.clear();
        }
        System.out.println("Error: " + t);
    }

    @Override
    public void onComplete() {
        List<T> tail;
        synchronized (this) {
            done = true;                      // §2.4：终止后订阅视为取消
            tail = drain();
        }
        if (!tail.isEmpty()) {
            process(tail);                    // 只处理残余，不再 request
        }
        System.out.println("Complete");
    }

    private List<T> drain() {
        List<T> copy = new ArrayList<>(buffer);
        buffer.clear();
        return copy;
    }

    private void process(List<T> batch) {
        System.out.println("Processing batch of " + batch.size());
        // 真实场景在这里写库/发消息；演示中不 sleep
    }
}

为什么 onComplete 里不能再 request

终止信号到达后，Subscription 在语义上已经取消（§1.6、§2.4），§2.3 干脆禁止在 onComplete/onError 中调用它的任何方法。所以收尾逻辑必须是「把缓冲区里剩下的处理掉」，而不是「再要一批」。

早期版本常写成：

@Override
public void onComplete() {
    processBatch();          // 内部又 subscription.request(batchSize)
}

在同步 Publisher 上这会触发重入，在异步 Publisher 上会得到一个无意义的 NOP（或某些实现直接抛异常）。正确做法是把「处理」和「请求」拆开：process 只处理数据，request 只发生在 onNext 的正常路径。

synchronized 不等于信号串行

§2.7 要求 Subscriber 对 request/cancel 的调用串行；§1.3 要求 Publisher 对 Subscriber 的信号串行。这两条责任不同：

  • 如果 Publisher 并发调用 onNext，它已经违反 §1.3，Subscriber 加 synchronized 只能防止状态被破坏，不能把违规变成合规。
  • synchronized 在 onNext 里是有用的：它保证 buffer 的读写不被并发破坏，也让「判断满批 + 取出 + 清空」成为一个原子步骤。

真正的串行化应该由 Publisher 侧的 WIP 计数器（2.2 的 emitting）或调度器保证，而不是让每个 Subscriber 自己加锁。

三个边界测试

测试一：cancel 后 request 是 no-op

@Override
public void onSubscribe(Subscription s) {
    s.cancel();
    s.request(10);        // §3.6：NOP
}

预期没有任何 onNext，也没有终止信号。如果实现仍然发射，说明 request 没有检查取消状态。

测试二：非法 request 触发 onError

@Override
public void onSubscribe(Subscription s) {
    s.request(-1);
}
@Override
public void onError(Throwable t) {
    System.out.println("Expected: " + t.getMessage());
}

预期打印 Expected: §3.9: non-positive request: -1，且 request 本身没有抛异常。

测试三：迟到的 onNext

用一个「cancel 后仍会发完当前批次」的 Publisher（把 2.2 的检查从每个元素前移到每批前即可复现）：

@Override
public void onNext(Integer i) {
    System.out.println("onNext: " + i);
    if (i == 5) subscription.cancel();
}

预期看到 1–5，之后可能还会出现 6–10（取决于实现）。Subscriber 不能因为「已经 cancel」就假定状态可以随意销毁。

实现检查清单

检查项规则说明
保存 Subscription§2.12 的前提不保存就无法 request/cancel
已有订阅时 cancel 新来的§2.5保留旧订阅，拒绝第二个 Publisher 驱动
用 request 表达需求§2.1不 request 就不该收到 onNext
request/cancel 串行调用§2.7多线程时加锁或收敛到单线程
终止回调不碰 Subscription§2.3包括不调用 cancel/request
容忍迟到元素§2.8cancel 后状态仍要可用
参数 null 抛 NPE§2.13四个信号方法都适用

小结

Subscriber 的实现重点不在「收数据」，而在边界：

  1. 需求由自己表达，不要指望 Publisher 猜。
  2. 调用要串行，并发调用 request/cancel 的责任在 Subscriber。
  3. 终止回调是禁区，不能在里面操作 Subscription/Publisher。
  4. cancel 不是立即静音，状态清理要能容忍迟到信号。

下一节实现 Processor：它同时扮演 Subscriber 和 Publisher，两侧契约会在这里正面相遇。

阅读导航

上一篇：2.2 异步 Publisher 与背压信号

下一篇：3.1 Processor：Subscriber 与 Publisher 的桥接

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南