java

2.1 同步 Publisher 最小实现

从零实现符合 Reactive Streams 规范的最小同步 Publisher：SyncSinglePublisher。逐行剖析 subscribe 的 onSubscribe 时序（§1.9）、request 的额度与非法值处理（§1.1、§3.9）、终止后的信号禁令（§1.7）与 cancel 语义（§3.12），并给出编译运行、边界测试和常见错误写法。

Leeting Yan 6 分钟阅读 2876 字

目标：一个最小的合规闭环

本节实现 SyncSinglePublisher<T>：它只发射一个元素然后完成。代码不到 60 行，但 onSubscribe 时序、request 额度、非法参数、终止状态、取消传播五个环节一个都不能省。

先说明一个边界：单值 Publisher 只能满足 TCK 中「恰好 1 个元素」的用例。PublisherVerification 还会调用 createPublisher(3)，要求实现精确发射 3 个元素；单值实现无法通过那一项。第 11 章的 TCK 测试对象是能按 elements 参数发射的 RangePublisher。本节的目标是把规则跑通，不是宣称单值类能全绿。

完整实现

package com.plume.eventd.core;

import java.util.Objects;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * 同步单值 Publisher 的最小实现。
 *
 * <p>规范落点：
 * <ul>
 *   <li>§1.9：subscribe 先调用 onSubscribe，再发其他信号，且正常返回</li>
 *   <li>§1.1：onNext 数量不超过 request 总量</li>
 *   <li>§1.7：发出终止信号后不再发任何信号</li>
 *   <li>§3.9：request(n &lt;= 0) 以 IllegalArgumentException 发 onError</li>
 *   <li>§3.12：cancel 后最终停止发射</li>
 * </ul>
 */
public final class SyncSinglePublisher<T> implements Publisher<T> {

    private final T value;

    public SyncSinglePublisher(T value) {
        // §2.13：信号参数不允许为 null，在装配期就拒绝
        this.value = Objects.requireNonNull(value, "value");
    }

    @Override
    public void subscribe(Subscriber<? super T> subscriber) {
        Objects.requireNonNull(subscriber, "subscriber"); // §1.9：null 必须抛 NPE
        subscriber.onSubscribe(new SingleSubscription<>(subscriber, value));
    }

    private static final class SingleSubscription<T> implements Subscription {

        private final Subscriber<? super T> subscriber;
        private final T value;

        private boolean cancelled;
        private boolean done;      // 已发射或已终止

        SingleSubscription(Subscriber<? super T> subscriber, T value) {
            this.subscriber = subscriber;
            this.value = value;
        }

        @Override
        public void request(long n) {
            // §3.6：cancel 后 request 是 no-op
            if (cancelled || done) {
                return;
            }

            // §3.9：非正数请求必须 onError，且 request 本身正常返回（§3.16）
            if (n <= 0) {
                done = true;
                subscriber.onError(new IllegalArgumentException(
                        "§3.9: non-positive request: " + n));
                return;
            }

            // §1.1：request(>0) 允许发射至多 1 个；先置位再发射，防重入
            done = true;
            subscriber.onNext(value);

            // §1.7：终止信号只发一次；如果 onNext 里已 cancel，就不再补 onComplete
            if (!cancelled) {
                subscriber.onComplete();
            }
        }

        @Override
        public void cancel() {
            // §3.5：快速返回，不同步做重计算；§3.7：重复 cancel 是 no-op
            cancelled = true;
        }
    }
}

逐行分析

subscribe：顺序与返回值

Objects.requireNonNull(subscriber, "subscriber");
subscriber.onSubscribe(new SingleSubscription<>(subscriber, value));

§1.9 要求两件事：

  1. onSubscribe 必须早于任何 onNext/onComplete/onError。
  2. subscribe 必须正常返回。唯一允许抛异常的情况是 subscriber 为 null——此时没有对象可以承载 onError，只能向调用者抛 NullPointerException。

所以「Publisher 太忙」不能写成 throw new IllegalStateException(...)。合法做法是先 onSubscribe，再 onError：

@Override
public void subscribe(Subscriber<? super T> subscriber) {
    subscriber.onSubscribe(noopSubscription());
    subscriber.onError(new IllegalStateException("publisher is shutting down"));
}

subscribe 依然正常返回，错误通过信号通道传递。

request：三个检查的顺序

if (cancelled || done) return;        // 状态检查
if (n <= 0) { done = true; onError; } // 参数检查
done = true;                          // 占位
onNext(value);

顺序不能反。如果先检查 n 再检查 cancelled，一个已取消的订阅仍会收到 onError——违反 §3.6（cancel 后 request 必须是 no-op）。如果先发射再置 done，onNext 里重入调用 request 会再发一次——违反 §1.1。

done 同时承担「已发射」和「已错误」两种终止状态。§1.7 要求进入终止状态后不再有任何信号，一个标志就够。

非法 request：为什么是 onError 而不是抛异常

if (n <= 0) {
    done = true;
    subscriber.onError(new IllegalArgumentException(
            "§3.9: non-positive request: " + n));
    return;
}

§3.9 规定 request(n <= 0) 必须发 onError(IllegalArgumentException)；§3.16 同时规定 request 必须正常返回。两条合起来排除了「直接 throw」的写法：

// 错误：违反 §3.16
if (n <= 0) throw new IllegalArgumentException("bad request");

这条规则在生产里很有用：request 可能从线程池、定时器或框架内部调用，抛出的异常会沿着调用栈冒到不该处理它的地方；而 onError 会走正常的终止路径，让 Subscriber 有机会清理资源。

onNext 之后为什么再检查一次 cancelled

subscriber.onNext(value);
if (!cancelled) {
    subscriber.onComplete();
}

Subscriber 完全可以在 onNext 里调用 cancel()：

@Override
public void onNext(String s) {
    subscription.cancel();
}

此时 cancelled 已为 true。再发 onComplete 不违反规范的字面条款（终止信号本身是允许的），但没有意义，而且会让「cancel 后不再收到信号」的直觉失效。跳过它更干净。

反过来说，不要在 onComplete 里再调用 Subscription。§2.3 禁止 Subscriber 在 onComplete/onError 中对 Subscription 或 Publisher 调用任何方法。常见错误是「在 onComplete 里补一个 cancel 做清理」，这会制造重入和竞态。

cancel：只置位，不做重活

public void cancel() {
    cancelled = true;
}

§3.5 要求 cancel 不得在调用线程同步执行重计算，要及时返回；§3.7 要求重复 cancel 是 no-op。这里只写一个布尔字段，两次 cancel 结果相同；如果实现涉及资源释放（关闭文件、断开连接），要把释放动作交给发射循环在观察到 cancelled 后完成，而不是在 cancel() 里同步做 I/O——否则调用 cancel 的线程会被阻塞。

编译与运行

javac --release 25 \
  -cp ~/.m2/repository/org/reactivestreams/reactive-streams/1.0.4/reactive-streams-1.0.4.jar \
  -d /tmp/eventd-classes \
  eventd-core/src/main/java/com/plume/eventd/core/SyncSinglePublisher.java

一个最小 Subscriber 验证正常路径：

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import com.plume.eventd.core.SyncSinglePublisher;

public class SyncSinglePublisherTest {
    public static void main(String[] args) {
        new SyncSinglePublisher<>("hello").subscribe(new Subscriber<>() {
            @Override
            public void onSubscribe(Subscription s) {
                System.out.println("onSubscribe");
                s.request(1);
            }
            @Override public void onNext(String s) { System.out.println("onNext: " + s); }
            @Override public void onError(Throwable t) { System.out.println("onError: " + t); }
            @Override public void onComplete() { System.out.println("onComplete"); }
        });
    }
}

输出：

onSubscribe
onNext: hello
onComplete

三个边界测试

cancel 后 request 无效

@Override
public void onSubscribe(Subscription s) {
    s.cancel();
    s.request(1);          // §3.6：no-op
}

预期：没有任何输出。request 看到 cancelled 直接返回，既不发射也不补终止信号。

非法 request

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

输出：

Expected error: §3.9: non-positive request: -1

注意 request(-1) 本身没有抛异常，是 onError 被回调。

重复 request 只发射一次

@Override
public void onSubscribe(Subscription s) {
    s.request(1);
    s.request(1);          // 第二次应被 done 拦住
}

预期只打印一次 onNext: hello 和一次 onComplete。如果没有 done 检查，第二次 request 会重复发射，直接违反 §1.1 和 §1.7。

常见错误写法

错误一：在 subscribe 里新起线程延迟调用 onSubscribe

public void subscribe(Subscriber<? super T> subscriber) {
    new Thread(() -> {
        try { Thread.sleep(100); } catch (InterruptedException ignored) {}
        subscriber.onSubscribe(new SingleSubscription<>(subscriber, value));
    }).start();
}

这段代码不违反 §1.9——异步调用 onSubscribe 是允许的，只要它先于其他信号。但它引入了不必要的线程和时序不确定性。除非数据源本身就是异步的，否则不要这样做。

错误二：用抛异常拒绝订阅

public void subscribe(Subscriber<? super T> subscriber) {
    if (closed) {
        throw new IllegalStateException("closed"); // 违反 §1.9
    }
    subscriber.onSubscribe(...);
}

§1.9 只允许 null subscriber 触发 NPE。其他拒绝必须先 onSubscribe 再 onError。

错误三：把 null 留给下游

this.value = value;                 // value 可能为 null
...
subscriber.onNext(value);            // §2.13：下游会抛 NPE

在构造器里 Objects.requireNonNull 更早暴露问题，也避免 Publisher 把一个非法信号送进管道。

从单值到数组：下一个模式

把单值扩展成数组，核心变化只有三处：

private int index;

@Override
public void request(long n) {
    if (cancelled || done) return;
    if (n <= 0) { done = true; onError(...); return; }

    long emitted = 0;
    while (emitted < n && index < values.length && !cancelled) {
        subscriber.onNext(values[index++]);
        emitted++;
    }
    if (index >= values.length && !cancelled) {
        done = true;
        subscriber.onComplete();
    }
}
  1. 用 index 记录进度，而不是一个 done 布尔。
  2. request(n) 发射 min(n, 剩余元素数) 个。
  3. 元素耗尽后终止。

这个模式是所有 Cold Publisher 的骨架：维护一个游标（数组下标、文件位置、数据库游标），在 request(n) 时推进游标。区别只在于「下一个元素从哪来」以及「取元素会不会阻塞」。

小结

规则代码落点
§1.9 onSubscribe 先行、subscribe 正常返回subscribe 第一行调用，null 单独抛 NPE
§1.1 onNext ≤ requestrequest(>0) 才发射，且 done 防重入
§3.9 / §3.16 非法请求n <= 0 → onError(IllegalArgumentException)，request 正常返回
§1.7 终止后无信号done 标志拦截后续调用
§3.12 cancel 最终停止cancelled 标志，request 检查

下一节把同步改成异步：在独立线程上生产，同时保持 request 累加、信号串行与取消传播。

阅读导航

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

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

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

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