java

11.3 Subscriber 与 Processor 合规测试

分清 SubscriberWhiteboxVerification 与 SubscriberBlackboxVerification 的真实抽象方法：白盒通过 WhiteboxSubscriberProbe 注册 onSubscribe/onNext/onError/onComplete，并用 SubscriberPuppet 控制 request 与 cancel；黑盒的 triggerRequest 默认是空实现。给出可编译、14 个必测项全部通过的白盒骨架，用实测失败信息解释 spec205 要求取消的是「新收到的重复订阅」而不是旧订阅，并展示 IdentityProcessorVerification 的四个抽象方法与 PublishProcessor 的 14 项实败。

Leeting Yan 6 分钟阅读 2699 字

两套基类，差别在「谁能控制订阅」

TCK 对 Subscriber 提供两个基类：

基类订阅控制适合什么实现
SubscriberWhiteboxVerification<T>测试通过 probe 直接控制 request 和 cancel自己能管理订阅状态、想让 TCK 精确驱动信号的 Subscriber
SubscriberBlackboxVerification<T>测试只能发信号，请求依赖 triggerRequest不便暴露内部状态，或想模拟真实 Publisher 行为

两者验证的规则高度重叠，白盒多出的能力是「精确控制」：TCK 可以只请求 1 个元素，也可以在你以为订阅还活着的时候取消。代价是你要在 Subscriber 里多写几行探测回调。

白盒的真实抽象方法只有两个

不少旧文章会把 publisherExecutorService() 也算成必须实现的方法。1.0.4 的实际签名不是这样：

public abstract Subscriber<T> createSubscriber(WhiteboxSubscriberProbe<T> probe);
public abstract T createElement(int element);          // 来自 WithHelperPublisher

// 有默认实现，可以不覆写
public ExecutorService publisherExecutorService();

publisherExecutorService() 的默认实现由 setUp() 创建 4 线程池、tearDown() 关闭。只有需要指定线程模型时才覆写它，而且覆写后要自己负责关闭。

createElement(int) 负责生成第 n 个测试元素。TCK 不关心元素内容，只要求同一个 element 索引每次生成可比较的值。

probe 与 puppet 的职责

WhiteboxSubscriberProbe<T> 有四个注册方法：

方法调用时机
registerOnSubscribe(SubscriberPuppet)收到 onSubscribe 后
registerOnNext(T)处理完 onNext 后
registerOnError(Throwable)处理完 onError 后
registerOnComplete()处理完 onComplete 后

只调用自己的 onNext 而不告诉 probe，TCK 会认为信号没有到达，测试最终以超时失败。这是白盒最容易踩的坑：探测回调是契约的一部分，不是可选的日志。

registerOnSubscribe 接收的 SubscriberPuppet 有两个方法：

方法TCK 的意图
triggerRequest(long n)要求 Subscriber 向上游请求 n 个
signalCancel()要求 Subscriber 取消订阅

一个能跑通的白盒骨架

下面这段代码基于 Reactor 3.8.8 的 BaseSubscriber，只实现两个抽象方法：

import org.reactivestreams.Subscription;
import org.reactivestreams.tck.SubscriberWhiteboxVerification;
import org.reactivestreams.tck.TestEnvironment;
import org.reactivestreams.tck.SubscriberWhiteboxVerification.SubscriberPuppet;
import org.reactivestreams.tck.SubscriberWhiteboxVerification.WhiteboxSubscriberProbe;
import reactor.core.publisher.BaseSubscriber;

public class EventSubscriberWhiteboxTckTest
        extends SubscriberWhiteboxVerification<String> {

    public EventSubscriberWhiteboxTckTest() {
        super(new TestEnvironment(5_000));
    }

    @Override
    public BaseSubscriber<String> createSubscriber(WhiteboxSubscriberProbe<String> probe) {
        return new BaseSubscriber<String>() {
            @Override
            protected void hookOnSubscribe(Subscription s) {
                probe.registerOnSubscribe(new SubscriberPuppet() {
                    @Override public void triggerRequest(long n) { request(n); }
                    @Override public void signalCancel() { cancel(); }
                });
                request(Long.MAX_VALUE);
            }

            @Override protected void hookOnNext(String value) { probe.registerOnNext(value); }
            @Override protected void hookOnError(Throwable error) { probe.registerOnError(error); }
            @Override protected void hookOnComplete() { probe.registerOnComplete(); }
        };
    }

    @Override
    public String createElement(int element) {
        return "elem" + element;
    }
}

用 BaseSubscriber 不是为了省代码，而是因为它已经处理了两件容易写错的事：

  1. onSubscribe 用 Operators.setOnce 保证只接受一个订阅；收到第二个时取消新来的那个。
  2. onNext、onError 对 null 参数做 NPE 检查，满足 §2.13。

如果你的实现不是 Reactor，这两条要自己写，而且必须写对。下一节就是一个真实的反例。

白盒实测：27 个用例，14 个必测项全过

用 TestNG 7.3.0 运行上面的类：

===============================================
Command line suite
Total tests run: 27, Passes: 14, Failures: 0, Skips: 13
===============================================

13 个跳过全部是 untested_*。它们不是「通过」，是 TCK 1.0.4 没有实现断言。14 个通过项如下：

测试方法规则
required_exerciseWhiteboxHappyPath综合路径：订阅、请求、取消
required_spec201_mustSignalDemandViaSubscriptionRequest§2.1 必须通过 request 表达需求
required_spec203_mustNotCallMethodsOnSubscriptionOrPublisherInOnComplete§2.3 完成回调里不得再 request/cancel
required_spec203_mustNotCallMethodsOnSubscriptionOrPublisherInOnError§2.3 错误回调同样不得再操作订阅
required_spec205_mustCallSubscriptionCancelIfItAlreadyHasAnSubscriptionAndReceivesAnotherOnSubscribeSignal§2.5 重复 onSubscribe 要取消新订阅
required_spec208_mustBePreparedToReceiveOnNextSignalsAfterHavingCalledSubscriptionCancel§2.8 取消后仍可能收到已请求的 onNext
required_spec209_mustBePreparedToReceiveAnOnCompleteSignalWithPrecedingRequestCall§2.9 先请求，再收到完成
required_spec209_mustBePreparedToReceiveAnOnCompleteSignalWithoutPrecedingRequestCall§2.9 没有请求也要能处理完成
required_spec210_mustBePreparedToReceiveAnOnErrorSignalWithPrecedingRequestCall§2.10 先请求，再收到错误
required_spec210_mustBePreparedToReceiveAnOnErrorSignalWithoutPrecedingRequestCall§2.10 没有请求也要能处理错误
required_spec213_onSubscribe_mustThrowNullPointerExceptionWhenParametersAreNull§2.13 onSubscribe(null) 抛 NPE
required_spec213_onNext_mustThrowNullPointerExceptionWhenParametersAreNull§2.13 onNext(null) 抛 NPE
required_spec213_onError_mustThrowNullPointerExceptionWhenParametersAreNull§2.13 onError(null) 抛 NPE
required_spec308_requestMustRegisterGivenNumberElementsToBeProduced§3.8 请求量必须被记录

注意 §2.9、§2.10 各有两条：带前置请求和不带前置请求是两种路径。只处理其中一种，另一种会失败。

spec205：取消的是新订阅

这是白盒最反直觉的一条。规范 §2.5 的原文是：

A Subscriber MUST call Subscription.cancel() on the given Subscription after an onSubscribe signal if it already has an active Subscription.

「given Subscription」指刚收到的那个，不是手里已有的那个。很多人凭直觉写成：

// 错：取消旧订阅，保留新订阅
@Override
public void onSubscribe(Subscription s) {
    if (this.subscription != null) {
        this.subscription.cancel();
    }
    this.subscription = s;
    s.request(Long.MAX_VALUE);
}

TCK 的报错是：

java.lang.AssertionError: Expected 2nd Subscription given to subscriber to be cancelled,
but Subscription.cancel() was not called within 5000 ms

正确写法是取消新来的重复订阅，保留原来的：

// 对：拒绝重复订阅
@Override
public void onSubscribe(Subscription s) {
    if (s == null) {
        throw new NullPointerException();
    }
    if (this.subscription != null) {
        s.cancel();
        return;
    }
    this.subscription = s;
    s.request(Long.MAX_VALUE);
}

Reactor 的 Operators.setOnce 就是这么做的：已有订阅时直接 cancel() 新传入的订阅，并报告一次重复设置。所以 BaseSubscriber 能过这条，手写状态机反而容易写反。

从协议角度看也合理：一个 Subscriber 只应该有一个活跃订阅。第二个 onSubscribe 是异常信号，拒绝它比替换旧连接更安全。

黑盒：没有 probe，也没有默认请求

黑盒的抽象方法同样是两个：

public abstract Subscriber<T> createSubscriber();
public abstract T createElement(int element);

publisherExecutorService() 也有默认实现。真正的区别是 triggerRequest(Subscriber)：

public void triggerRequest(Subscriber<? super T> subscriber) {
    // 默认什么都不做
}

默认实现是空的。这意味着如果你的 Subscriber 不在 onSubscribe 里主动请求，而是在外部调用 triggerRequest 时才请求，就必须覆写这个方法；否则 TCK 发不出元素，测试会超时。反过来，像 BaseSubscriber 这样在 hookOnSubscribe 里请求 Long.MAX_VALUE 的实现，不需要覆写。

用同一个 BaseSubscriber 跑黑盒：

Total tests run: 26, Passes: 11, Failures: 0, Skips: 15

11 个必测项是：spec201、spec203（两条）、spec205、spec209（两条）、spec210（两条）、spec213（三条）。15 个跳过全部是 untested_*。

黑盒的 required_spec205_... 名字里带 blackbox，验证的行为与白盒一致：第二个 onSubscribe 到达时，新订阅必须被取消。BaseSubscriber 在两个基类下都能过，说明它实现的是协议本身，不是针对某个测试的特例。

IdentityProcessor：能编译不等于合规

IdentityProcessorVerification<T> 同时验证 Publisher 侧和 Subscriber 侧，必须实现四个方法：

public abstract Processor<T, T> createIdentityProcessor(int bufferSize);
public abstract Publisher<T> createFailedPublisher();
public abstract T createElement(int element);                    // 来自 WithHelperPublisher
public abstract ExecutorService publisherExecutorService();      // 来自 WithHelperPublisher

createPublisher(long) 有默认实现，内部用 createHelperPublisher(long) 生成元素再喂给被测 Processor，不需要自己写。

「Identity」的含义是：喂进去什么元素，下游就应该按请求量收到什么元素。bufferSize 是 TCK 期望处理器内部能缓冲的容量。一个只做转发、不缓冲的 Processor 不满足这个模型。

拿 RxJava 的 PublishProcessor 试：

@Override
public Processor<String, String> createIdentityProcessor(int bufferSize) {
    return PublishProcessor.<String>create().toSerialized();
}

实测结果是：

Total tests run: 68, Passes: 30, Failures: 14, Skips: 24

失败项包括：

失败方法现象
required_exerciseWhiteboxHappyPathDid not receive expected cancelling of upstream subscription within 5000 ms
required_spec101_subscriptionRequestMustResultInTheCorrectNumberOfProducedElementsproduced no element after first request within 5000 ms
required_mustRequestFromUpstreamForElementsThatHaveBeenRequestedLongAgo早先请求的元素没有被补发
required_spec312_cancelMustMakeThePublisherToEventuallyStopSignaling取消后没有停止向上游请求
required_spec313_cancelMustMakeThePublisherEventuallyDropAllReferencesToTheSubscriber取消后仍持有订阅者引用

这些失败不是「RxJava 不合规」。PublishProcessor 是热的多播处理器：没有下游时元素直接丢弃，它本来就不承诺「先收下、等下游请求再发」。把它当作 identity processor 用，语义就不匹配。需要的是带缓冲、按请求量排空的处理器，而不是热转发器。

这也解释了为什么 Processor 的 TCK 最难通过：它要求同一个对象在上游是守规矩的 Subscriber，在下游是守规矩的 Publisher，中间还要把请求量正确传递。只测一侧的实现，往往在另一侧暴露问题。

选择建议

场景建议
自己实现 Subscriber，想验证请求/取消细节白盒，probe 能精确驱动
Subscriber 只是转发，不想暴露内部状态黑盒，成本更低
用了 Reactor 的 BaseSubscriber两个基类都可以直接跑
实现 Processor必须跑 IdentityProcessorVerification，不能只跑 Publisher 侧
看到 13/15 个跳过那是 untested_*，不是通过

小结

关注点结论
白盒抽象方法createSubscriber(probe)、createElement(int)；执行器有默认实现
probe 回调四个都要调用，否则测试超时
spec205取消新收到的重复订阅，不是旧订阅
黑盒triggerRequest 默认空实现，惰性请求的实现必须覆写
白盒实测27 个用例，14 通过、13 个 untested_*
黑盒实测26 个用例，11 通过、15 个 untested_*
Processor四个抽象方法；热处理器实测 14 个必测项失败

阅读导航

上一篇：11.2 Publisher 合规测试

下一篇：12.1 生命周期钩子：doOn* 系列

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

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