操作符的本质
在 Reactor 和 RxJava 中，操作符不是「函数调用」，而是返回新 Publisher 的装饰器。当你写：
Flux.just(1, 2, 3)
.map(i -> i * 2)
.filter(i -> i > 2)
.take(2);
实际上发生了：
TakePublisher(
FilterPublisher(
MapPublisher(
JustPublisher(1, 2, 3)
)
)
)
每个操作符创建一个新的 Publisher，它包装了上游 Publisher，拦截并转换信号后再转发给下游。
共同模式
所有操作符共享以下结构：
public class XxxOperator<T, R> implements Publisher<R> {
private final Publisher<T> upstream;
private final SomeFunction<T, R> function;
@Override
public void subscribe(Subscriber<? super R> downstream) {
upstream.subscribe(new XxxSubscriber(downstream));
}
private class XxxSubscriber implements Subscriber<T> {
private final Subscriber<? super R> downstream;
private Subscription upstreamSubscription;
// onSubscribe: 保存上游 Subscription，向下游发送自己的 Subscription
// onNext: 转换/过滤/计数，再发射给 downstream
// onError/onComplete: 透传
}
}
MapOperator
package com.plume.eventd.core;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import java.util.function.Function;
public final class MapOperator<T, R> implements Publisher<R> {
private final Publisher<T> upstream;
private final Function<T, R> mapper;
public MapOperator(Publisher<T> upstream, Function<T, R> mapper) {
this.upstream = upstream;
this.mapper = mapper;
}
@Override
public void subscribe(Subscriber<? super R> downstream) {
upstream.subscribe(new MapSubscriber(downstream));
}
private class MapSubscriber implements Subscriber<T> {
private final Subscriber<? super R> downstream;
private Subscription upstreamSubscription;
MapSubscriber(Subscriber<? super R> downstream) {
this.downstream = downstream;
}
@Override
public void onSubscribe(Subscription subscription) {
this.upstreamSubscription = subscription;
// 向下游发送一个「透明」的 Subscription
downstream.onSubscribe(new Subscription() {
@Override public void request(long n) {
upstreamSubscription.request(n);
}
@Override public void cancel() {
upstreamSubscription.cancel();
}
});
}
@Override
public void onNext(T item) {
try {
R result = mapper.apply(item);
downstream.onNext(result);
} catch (Exception e) {
upstreamSubscription.cancel();
downstream.onError(e);
}
}
@Override public void onError(Throwable t) {
downstream.onError(t);
}
@Override public void onComplete() {
downstream.onComplete();
}
}
}
关键洞察：Subscription 的透明转发
MapSubscriber.onSubscribe 创建了一个新的 Subscription 对象发给下游。这个对象只是简单地转发 request 和 cancel 给上游。
为什么不直接把上游的 Subscription 发给下游？因为下游可能调用 cancel()，而 MapSubscriber 需要在 cancel 时做清理（虽然本例没有）。更关键的是，这保持了分层结构——每个操作符都有机会拦截和修改信号。
FilterOperator
package com.plume.eventd.core;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import java.util.function.Predicate;
public final class FilterOperator<T> implements Publisher<T> {
private final Publisher<T> upstream;
private final Predicate<T> predicate;
public FilterOperator(Publisher<T> upstream, Predicate<T> predicate) {
this.upstream = upstream;
this.predicate = predicate;
}
@Override
public void subscribe(Subscriber<? super T> downstream) {
upstream.subscribe(new FilterSubscriber(downstream));
}
private class FilterSubscriber implements Subscriber<T> {
private final Subscriber<? super T> downstream;
private Subscription upstreamSubscription;
FilterSubscriber(Subscriber<? super T> downstream) {
this.downstream = downstream;
}
@Override
public void onSubscribe(Subscription subscription) {
this.upstreamSubscription = subscription;
downstream.onSubscribe(new Subscription() {
@Override public void request(long n) {
upstreamSubscription.request(n);
}
@Override public void cancel() {
upstreamSubscription.cancel();
}
});
}
@Override
public void onNext(T item) {
boolean pass;
try {
pass = predicate.test(item);
} catch (Throwable e) {
upstreamSubscription.cancel();
downstream.onError(e);
return;
}
if (pass) {
downstream.onNext(item);
} else {
// 被丢弃的元素已经消耗了一份下游需求，必须补回来
upstreamSubscription.request(1);
}
}
@Override public void onError(Throwable t) {
downstream.onError(t);
}
@Override public void onComplete() {
downstream.onComplete();
}
}
}
关键洞察：过滤必须补偿 request
FilterOperator 把下游的 request(n) 转发给上游，但过滤掉的元素已经消耗了一份额度。如果只转发、不补偿，下游请求 1 个而恰好被过滤时，上游额度归零，整条流停在那里等一个永远不会到来的 request。
所以丢弃元素时要立刻补 request(1)：
} else {
upstreamSubscription.request(1);
}
Reactor 的 FluxFilter 正是这么做的——FilterSubscriber.onNext 在谓词返回 false 时调用 s.request(1)（3.8.8 字节码里可以看到这条调用）。这也是「request(10) 不保证收到 10 个、但过滤器会尽力补足」的实现基础。
即便如此，下游也不能假设「请求 n 必得 n」：上游数据耗尽时会发 onComplete，§1.2 允许少发后终止。业务侧应该以终止信号为准，而不是以请求数量为准。
TakeOperator
package com.plume.eventd.core;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
public final class TakeOperator<T> implements Publisher<T> {
private final Publisher<T> upstream;
private final long limit;
public TakeOperator(Publisher<T> upstream, long limit) {
this.upstream = upstream;
this.limit = limit;
}
@Override
public void subscribe(Subscriber<? super T> downstream) {
upstream.subscribe(new TakeSubscriber(downstream));
}
private class TakeSubscriber implements Subscriber<T> {
private final Subscriber<? super T> downstream;
private Subscription upstreamSubscription;
private final AtomicLong remaining = new AtomicLong(limit);
private final AtomicBoolean completed = new AtomicBoolean(false);
TakeSubscriber(Subscriber<? super T> downstream) {
this.downstream = downstream;
}
@Override
public void onSubscribe(Subscription subscription) {
this.upstreamSubscription = subscription;
downstream.onSubscribe(new Subscription() {
@Override public void request(long n) {
if (limit <= 0) return; // take(0)：不向上游要任何元素
upstreamSubscription.request(n);
}
@Override public void cancel() {
upstreamSubscription.cancel();
}
});
if (limit <= 0 && completed.compareAndSet(false, true)) {
subscription.cancel();
downstream.onComplete(); // 立即完成，不等上游
}
}
@Override
public void onNext(T item) {
if (completed.get()) return;
long rem = remaining.decrementAndGet();
if (rem >= 0) {
downstream.onNext(item);
if (rem == 0) {
// 达到数量限制，取消上游并通知下游完成
if (completed.compareAndSet(false, true)) {
upstreamSubscription.cancel();
downstream.onComplete();
}
}
}
// rem < 0 意味着已经取消，忽略额外元素
}
@Override public void onError(Throwable t) {
if (completed.compareAndSet(false, true)) {
downstream.onError(t);
}
}
@Override public void onComplete() {
if (completed.compareAndSet(false, true)) {
downstream.onComplete();
}
}
}
}
关键洞察：主动取消上游
TakeOperator 的关键行为是：当发射数量达到限制时，主动取消上游并向下游发出 onComplete。
if (rem == 0) {
upstreamSubscription.cancel(); // 停止上游不再发射
downstream.onComplete(); // 告诉下游「完成了」
}
为什么要在 onNext 里发 onComplete？因为 TakeSubscriber 是中间层，它决定了「什么时候算完成」。上游可能还有更多数据，但 Take 说「我只要 N 个」。
take(0) 是一个必须单独处理的边界：下游可能永远等不到 onNext 来触发完成逻辑。正确行为是在 onSubscribe 里直接 cancel 上游并 onComplete，本实现用 limit <= 0 分支做了这件事。Reactor 的 Flux.take(0) 同样在订阅阶段立即完成。
操作符链的完整示例
import com.plume.eventd.core.*;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import java.util.concurrent.Executors;
public class OperatorChainDemo {
public static void main(String[] args) {
// 上游：异步发射 1 到 10
AsyncRangePublisher upstream = new AsyncRangePublisher(
1, 11, Executors.newSingleThreadExecutor());
// 操作符链：map(x -> x * 2) -> filter(x > 5) -> take(3)
TakeOperator<Integer> chain = new TakeOperator<>(
new FilterOperator<>(
new MapOperator<>(upstream, x -> x * 2),
x -> x > 5
),
3
);
chain.subscribe(new Subscriber<>() {
@Override public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}
@Override public void onNext(Integer i) {
System.out.println("result: " + i);
}
@Override public void onError(Throwable t) {
System.out.println("error: " + t.getMessage());
}
@Override public void onComplete() {
System.out.println("done");
}
});
try { Thread.sleep(200); } catch (InterruptedException e) {}
System.exit(0);
}
}
计算过程：
| 上游 | map(x*2) | filter(x>5) | take(3) |
|---|---|---|---|
| 1 | 2 | 丢弃 | — |
| 2 | 4 | 丢弃 | — |
| 3 | 6 | 通过 | emit 6 (1/3) |
| 4 | 8 | 通过 | emit 8 (2/3) |
| 5 | 10 | 通过 | emit 10 (3/3) → cancel 上游 |
| 6..10 | — | — | 已取消，不再接收 |
输出：
result: 6
result: 8
result: 10
done
操作符链的 request 信号流
操作符链中的 request 是如何传递的？
downstream.subscriber(request 3)
→ TakeOperator: 转发 request(3) 给 FilterOperator
→ FilterOperator: 转发 request(3) 给 MapOperator
→ MapOperator: 转发 request(3) 给 AsyncRangePublisher
→ AsyncRangePublisher: 发射 3 个元素
大多数操作符把 request(n) 原样转发给上游，但有三类会修改额度：
| 类型 | request 行为 | 例子 |
|---|---|---|
| 过滤型 | 丢弃元素时补 request(1) | filter、distinct |
| 截断型 | 达到上限后 cancel，不再转发 | take(n)、takeUntil |
| 缓冲/聚合型 | 下游 request 1，上游 request N 或 Long.MAX_VALUE | onBackpressureBuffer、collectList |
图中的 take(3) 在上游发出第 3 个通过元素后立即 cancel；filter 每丢弃一个元素就向上游补一个请求，否则下游的额度会被「过滤」悄悄吃掉。
操作符的实现检查清单
| 检查项 | 说明 | 常见错误 |
|---|---|---|
| subscribe 返回新对象 | 不修改上游，返回包装 Publisher | 直接返回上游 |
| onSubscribe 发送透明 Subscription | 让下游可以 request/cancel | 不发 Subscription |
| request 转发给上游 | 大多数操作符直接转发 | request 不转发（下游收不到数据） |
| cancel 转发给上游 | 停止整条链 | cancel 不传播（上游继续发射） |
| onNext 转换后发射 | 异常时取消上游 + onError | 直接透传（不做转换） |
| onError/onComplete 透传 | 或根据逻辑修改（如 take） | 吃掉终止信号 |
与 Reactor 操作符的对比
| 手写实现 | Reactor 等价 | 差异 |
|---|---|---|
MapOperator | Flux.map | Reactor 额外提供上下文传播、算子融合与错误模式 |
FilterOperator | Flux.filter | 同样在丢弃元素时补 request(1) |
TakeOperator | Flux.take(n) | Reactor 还支持 take(Duration)、takeUntil 等变体 |
手写操作符的价值不在「性能更好」，而在「理解框架在做什么」。当你遇到 Reactor 的 bug 或需要自定义操作符时，这些知识会直接派上用场。
小结
操作符链的本质是装饰器模式的层层包装：
- 每个操作符创建新的 Publisher，包装上游。
- 订阅时，下游的
Subscriber被包装成操作符内部的Subscriber。 request和cancel大多数情况透明转发。onNext被拦截、转换、过滤或计数后，再发射给下游。
理解这个模式后，你可以：
- 自定义任何操作符（如
distinct、debounce）。 - 调试 Reactor/RxJava 操作符链时知道「信号走到哪一层了」。
- 理解为什么某些操作符组合会导致意外的背压行为（如
filter后onComplete提前到达）。
下一节，我们讲一个最关键的概念——subscribeOn 和 publishOn，它们决定了操作符链在哪个线程上执行。
阅读导航
上一篇：3.1 Processor：Subscriber 与 Publisher 的桥接
下一篇：3.3 调度切换：subscribeOn 与 publishOn
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。