java

3.1 Processor：Subscriber 与 Publisher 的桥接

实现 MapProcessor<T,R>：同时扮演上游 Subscriber 与下游 Publisher。重点解决下游先订阅时 request 丢失的 pending demand 问题、cancel 双向上传播、终止信号只发一次，以及 mapper 抛异常时先取消上游再通知下游的顺序。附两个订阅顺序的实测与三条 Processor 专项规则。

Leeting Yan 2026-10-30 6 分钟阅读 2775 字

Processor 的定位

Processor 同时实现 Subscriber<T> 和 Publisher<R>，是管道中的「一段」：

转换：接收上游 T，映射成 R 发给下游（map 的内部形态）。
桥接：在不同实现之间转接。注意「都实现了 RS 接口」只保证类型能接上，语义差异仍需要库提供的适配器。
副作用：不改变数据，但记录日志、指标、埋点。

§4.1 要求 Processor 同时遵守 Subscriber 和 Publisher 两套契约；§4.2 规定它可以选择恢复上游的 onError，一旦选择恢复，必须把上游 Subscription 视为已取消，否则要立即向下游传播错误。

MapProcessor 的四个难点

双重订阅：上游 upstream.subscribe(processor)，下游 processor.subscribe(downstream)，两边顺序不确定。
pending demand：下游可能在上游 onSubscribe 之前就 request(n)，这时没有对象可转发，必须先记账。
cancel 传播：下游取消要向上游传播；上游终止要让下游只收到一次终止信号。
异常顺序：mapper 抛异常时，先取消上游，再向下游发 onError。

完整实现

package com.plume.eventd.core;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import org.reactivestreams.Processor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * 单下游的转换 Processor：T -> R。
 *
 * <p>支持两种订阅顺序：下游先订阅时缓存 request，上游到达后补发；
 * 上游先订阅时直接转发 request。
 */
public final class MapProcessor<T, R> implements Processor<T, R> {

    private final Function<? super T, ? extends R> mapper;
    private final Object lock = new Object();
    // 串行化对下游的信号：onNext、onError、onComplete 不能并发/重入
    private final Object emitLock = new Object();

    private Subscriber<? super R> downstream;
    private Subscription upstream;
    private long pendingDemand;          // 上游未就绪时缓存的 request
    private boolean done;

    private final AtomicBoolean cancelled = new AtomicBoolean();

    public MapProcessor(Function<? super T, ? extends R> mapper) {
        this.mapper = Objects.requireNonNull(mapper, "mapper");
    }

    // ================= Publisher 侧 =================

    @Override
    public void subscribe(Subscriber<? super R> subscriber) {
        Objects.requireNonNull(subscriber, "subscriber");

        synchronized (lock) {
            if (downstream != null) {
                // 本实现只支持一个下游；拒绝也必须走 onSubscribe + onError
                subscriber.onSubscribe(NoopSubscription.INSTANCE);
                subscriber.onError(new IllegalStateException("MapProcessor 只支持单订阅"));
                return;
            }
            downstream = subscriber;
        }
        subscriber.onSubscribe(new DownstreamSubscription());
    }

    private final class DownstreamSubscription implements Subscription {

        @Override
        public void request(long n) {
            synchronized (lock) {
                // §3.6：取消/终止之后，request 是 no-op，包括 n <= 0
                if (done || cancelled.get()) return;
            }
            if (n <= 0) {
                // fail 内部用 emitLock 串行化，不会与正在进行的 onNext 并发
                fail(new IllegalArgumentException(
                        "§3.9: non-positive request: " + n));
                return;
            }
            Subscription target;
            synchronized (lock) {
                if (done || cancelled.get()) return;
                if (upstream == null) {
                    // 上游还没订阅：缓存，等 onSubscribe 时补发
                    long updated = pendingDemand + n;
                    pendingDemand = updated < 0 ? Long.MAX_VALUE : updated;
                    return;
                }
                target = upstream;
            }
            target.request(n);
        }

        @Override
        public void cancel() {
            if (!cancelled.compareAndSet(false, true)) return;
            Subscription target;
            synchronized (lock) {
                target = upstream;
            }
            if (target != null) target.cancel();
        }
    }

    // ================= Subscriber 侧 =================

    @Override
    public void onSubscribe(Subscription subscription) {
        Objects.requireNonNull(subscription, "subscription");

        long demand;
        synchronized (lock) {
            if (upstream != null) {
                // §2.5：已有活动订阅时，取消新来的这个
                subscription.cancel();
                return;
            }
            if (cancelled.get()) {
                subscription.cancel();
                return;
            }
            upstream = subscription;
            demand = pendingDemand;
            pendingDemand = 0;
        }
        if (demand > 0) {
            subscription.request(demand);
        }
    }

    @Override
    public void onNext(T item) {
        Subscriber<? super R> target;
        synchronized (lock) {
            if (done || cancelled.get()) return;
            target = downstream;
        }
        if (target == null) {
            // 没有下游却收到数据，说明上游违反了「只在 request 后发射」
            fail(new IllegalStateException("onNext before downstream subscribe"));
            return;
        }
        R mapped;
        try {
            mapped = Objects.requireNonNull(mapper.apply(item), "mapper returned null");
        } catch (Throwable t) {
            fail(t);
            return;
        }
        synchronized (emitLock) {
            if (done || cancelled.get()) return;
            target.onNext(mapped);
        }
    }

    @Override
    public void onError(Throwable throwable) {
        Objects.requireNonNull(throwable, "throwable");
        Subscriber<? super R> target;
        synchronized (lock) {
            if (done) return;
            done = true;
            target = downstream;
        }
        if (target != null) {
            synchronized (emitLock) {
                target.onError(throwable);
            }
        }
    }

    @Override
    public void onComplete() {
        Subscriber<? super R> target;
        synchronized (lock) {
            if (done) return;
            done = true;
            target = downstream;
        }
        if (target != null) {
            synchronized (emitLock) {
                target.onComplete();
            }
        }
    }

    /** 本地失败：取消上游、终止下游，且只执行一次。 */
    private void fail(Throwable t) {
        Subscriber<? super R> target;
        Subscription up;
        synchronized (lock) {
            if (done) return;
            done = true;
            target = downstream;
            up = upstream;
        }
        if (up != null) up.cancel();
        if (target != null) {
            synchronized (emitLock) {
                target.onError(t);
            }
        }
    }

    private enum NoopSubscription implements Subscription {
        INSTANCE;
        @Override public void request(long n) { }
        @Override public void cancel() { }
    }
}

关键设计决策

1. 订阅顺序：两种都必须支持

下游先订阅是自然写法：

MapProcessor<String, Integer> processor = new MapProcessor<>(String::length);
processor.subscribe(downstream);       // 1. 下游拿到 Subscription
upstream.subscribe(processor);         // 2. 上游把 Subscription 交给 processor

问题在于第 1 步：downstream.onSubscribe 通常会立刻 request(n)，而这时 upstream 还是 null。如果直接丢弃，上游永远收不到需求，整条流挂起。必须把 request 记下来（pendingDemand），在 onSubscribe 到达时补发。

反过来，上游先订阅也合法：

upstream.subscribe(processor);
processor.subscribe(downstream);

此时 onSubscribe 先到，pendingDemand 为 0，下游的 request 直接转发。两种顺序共用同一份代码，不需要分支。

pendingDemand 同样要做饱和加法，避免多次 request 累加后溢出。

2. 为什么用 synchronized 而不是纯 AtomicLong

upstream、pendingDemand、done 之间有组合不变式：「只有当 upstream 非空时才转发 request」「补发 pendingDemand 只能发生一次」。用一把锁把 request、onSubscribe、cancel 的临界区串起来，比维护多个原子变量再推理交错简单得多。

锁内不做 I/O、不调用下游方法，所以不会长时间持有。onNext 只读取 downstream 和 done 后立即出锁，mapper 和下游调用都在锁外执行。

3. cancel 与终止的分工

cancel() 由下游发起：用 CAS 保证只向上游传播一次，然后置 cancelled。上游之后发来的信号一律丢弃。
fail() 由 Processor 自己发起：mapper 抛异常或收到非法 request 时，先置 done、取消上游，再通知下游。终止信号只发一次。
上游的 onError/onComplete：置 done 后原样转发。

注意 done 和 cancelled 是两种不同的终止原因，但对外效果一致：不再有 onNext。

4. 异常顺序：先取消上游，再通知下游

if (up != null) up.cancel();
if (target != null) target.onError(t);

顺序反了会有一个窗口：下游收到 onError 后释放资源，而上游还在继续发射，Processor 只能丢弃。先取消再通知，能最快让上游停下来。

另一个容易忽略的点是 mapper.apply 返回 null。RS 流不允许 null 元素（§2.13），Objects.requireNonNull 会把它转成 onError(NullPointerException)，而不是把 null 透传给下游。

运行验证

下游先订阅

MapProcessor<String, Integer> processor = new MapProcessor<>(String::length);

processor.subscribe(new Subscriber<>() {
    @Override public void onSubscribe(Subscription s) {
        s.request(Long.MAX_VALUE);       // 此时 upstream 还没到
    }
    @Override public void onNext(Integer i) { System.out.println("length: " + i); }
    @Override public void onError(Throwable t) { t.printStackTrace(); }
    @Override public void onComplete() { System.out.println("done"); }
});

new SyncSinglePublisher<>("hello").subscribe(processor);

输出：

length: 5
done

如果去掉 pendingDemand，这段代码会一行都不打印：request 在 upstream == null 时被丢弃，上游从未收到需求，SyncSinglePublisher 不会发射。

链式组合

MapProcessor<String, Integer> length = new MapProcessor<>(String::length);
MapProcessor<Integer, String> format = new MapProcessor<>(i -> "length=" + i);

format.subscribe(downstream);
length.subscribe(format);
upstream.subscribe(length);

从下游往上游逐个订阅，这正是 Reactor 装配期 Flux.map(...).map(...) 构造链的方向。真正开始流动发生在最后一个 subscribe 触发的 onSubscribe 回传。

异常路径

MapProcessor<String, Integer> bad = new MapProcessor<>(s -> {
    if (s.isEmpty()) throw new IllegalArgumentException("empty");
    return s.length();
});
bad.subscribe(errorLoggingSubscriber);
new SyncSinglePublisher<>("").subscribe(bad);

下游收到 onError(IllegalArgumentException("empty"))，上游被 cancel，不会再有 onNext。

Processor 的三条专项规则

规则	要求	本实现的落点
§4.1	同时遵守两侧契约	上游侧实现四个回调，下游侧实现 Publisher + Subscription
§4.2	恢复 onError 时必须视上游已取消	`fail()` 先 `up.cancel()` 再向下游发错误
§2.5	已有订阅时取消新来的	`onSubscribe` 中 `upstream != null` 时 cancel 新 Subscription

什么时候不该手写 Processor

教学之外，手写 Processor 的性价比很低。Reactor 3.8 的对应工具是 Sinks：

Sinks.Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();
Flux<String> flux = sink.asFlux().map(String::length);

Sinks.many() 提供 unicast / multicast / replay 三类，内部处理了串行化、终止状态和背压策略。DirectProcessor、UnicastProcessor 这些旧类在 Reactor 3.5 前后已被移除或弃用，新代码不应再引用。第 9 章会专门讲 Sinks 的发射语义。

小结

Processor 的难点不在「转换」，而在两侧契约的接缝：

pending demand：下游可能先于上游请求，必须缓存。
终止唯一：done 保证 onComplete/onError 只发一次。
取消传播：下游 cancel 要向上游传播，且只传播一次。
失败顺序：先取消上游，再通知下游。

下一节把单个 Processor 扩展成完整的操作符链，手写 map、filter、take。

阅读导航

上一篇：2.3 Subscriber 实现与信号契约

下一篇：3.2 操作符链：map、filter、take

继续阅读

探索更多技术文章

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

全部文章返回首页