速查表使用说明
本附录按「先分类、后对照」组织。每个操作符包含：Reactor API、RxJava API（如有）、语义一句话、典型场景。标注了差异点（⚠️）的操作符需特别注意。
一、创建操作符
创建操作符负责从零产生数据流。它们是所有响应式链的起点。
从已有数据创建
| Reactor | RxJava | 签名 | 语义 |
|---|
Flux.just(T...) | Observable.just(T...) | just(T... values) | 发射固定值序列后完成 |
Flux.fromIterable(Iterable) | Observable.fromIterable | fromIterable(Iterable<T>) | 遍历集合并逐个发射 |
Flux.fromStream(Stream) | Observable.fromStream | fromStream(Stream<T>) | 遍历 Stream |
Flux.range(int, int) | Observable.range | range(int start, int count) | 发射整数范围 |
Flux.empty() | Observable.empty() | empty() | 立即完成，不发射元素 |
Flux.never() | Observable.never() | never() | 永不终止 |
Flux.error(Throwable) | Observable.error() | error(Throwable) | 立即报错 |
从异步源创建
| Reactor | RxJava | 签名 | 语义 |
|---|
Mono.fromCallable | Single.fromCallable | fromCallable(Callable<T>) | 每次订阅时调用 |
Mono.fromFuture | Single.fromFuture | fromFuture(Future<T>) | 从 Future 创建 |
Mono.fromSupplier | Single.fromSupplier | fromSupplier(Supplier<T>) | 每次订阅时生成 |
Mono.defer | Single.defer | defer(Supplier<Mono<T>>) | 延迟创建整个流 |
Mono.create | Single.create | create(MonoSink<T>) | 异步创建单值 |
Flux.create | Observable.create | create(FluxSink<T>) | 异步创建多值 |
Flux.generate | Observable.generate | generate(Consumer<SynchronousSink<T>>) | 同步状态机生成 |
定时创建
| Reactor | RxJava | 签名 | 语义 |
|---|
Flux.interval | Observable.interval | interval(Duration) | 按固定间隔发射递增计数 |
Mono.delay | Single.timer | delay(Duration) | 延迟后发射 0 |
典型场景
- HTTP 请求结果：
Mono.fromCallable(() -> httpClient.get(url)) - 轮询数据库：
Flux.interval(Duration.ofSeconds(5)).flatMap(i -> queryDB()) - 初始化配置：
Mono.defer(() -> configService.load())
二、变换操作符
变换操作符把一个流转换成另一个流，是业务逻辑的核心载体。
一对一变换
| Reactor | RxJava | 签名 | 语义 |
|---|
map | map | map(Function<T,R>) | 同步一对一变换 |
cast | cast | cast(Class<R>) | 类型转换 |
ofType | ofType | ofType(Class<R>) | 过滤+类型转换 |
index | — | index() | 带索引发射 (Long, T)；RxJava 无此操作符 |
一对多展开
| Reactor | RxJava | 签名 | 语义 |
|---|
flatMap | flatMap | flatMap(Function<T,Publisher<R>>) | 异步展开，无序 |
concatMap | concatMap | concatMap(Function<T,Publisher<R>>) | 异步展开，有序 |
switchMap | switchMap | switchMap(Function<T,Publisher<R>>) | 只保留最新子流 |
flatMapSequential | — | 同上 | 无序展开、有序合并 |
过滤
| Reactor | RxJava | 签名 | 语义 |
|---|
filter | filter | filter(Predicate<T>) | 保留符合条件的 |
take(long) | take(long) | take(long n) | 取前 N 个 |
takeWhile | takeWhile | takeWhile(Predicate<T>) | 条件为 true 时取 |
takeUntil | takeUntil | takeUntil(Predicate<T>) | 条件为 true 时停止 |
skip(long) | skip(long) | skip(long n) | 跳过前 N 个 |
skipWhile | skipWhile | skipWhile(Predicate<T>) | 条件为 true 时跳过 |
distinct | distinct | distinct() | 去重 |
distinctUntilChanged | distinctUntilChanged | 同上 | 去除连续重复 |
分组与窗口
| Reactor | RxJava | 签名 | 语义 |
|---|
groupBy | groupBy | groupBy(Function<T,K>) | 按键分组 |
window | window | window(int size) | 按数量分窗口 |
buffer | buffer | buffer(int size) | 按数量缓冲为 List |
⚠️ 差异点
| 特性 | Reactor | RxJava |
|---|
flatMap 默认并发度 | 256 | Integer.MAX_VALUE |
concatMap 默认预取 | 1（FluxConcatMapNoPrefetch，根本不预取） | 2 |
switchMap 取消旧流 | 是 | 是 |
典型场景
- 列表查询展开：
flatMap(page -> queryPage(page)) - 去重：
distinctUntilChanged() 用于去除连续重复的价格推送 - 分页：
buffer(100) 把流转换成每页 100 条的 List
三、组合操作符
组合操作符把多个流合并成一个。
配对组合
| Reactor | RxJava | 签名 | 语义 |
|---|
zip | zip | zip(Publisher... sources) | 等所有流各一个元素后组合 |
zipWith | zipWith | zipWith(Publisher<B>) | 二元 zip |
合并
| Reactor | RxJava | 签名 | 语义 |
|---|
merge | merge | merge(Publisher... sources) | 交错，先 emit 先赢 |
Flux.mergeOrdered(...) | — | mergeOrdered(Comparator, Publisher...) | 静态方法；交错合并后按比较器排序 |
concat | concat | concat(Publisher... sources) | 顺序连接 |
combineLatest | combineLatest | combineLatest(Publisher...) | 最新值组合 |
选择
| Reactor | RxJava | 签名 | 语义 |
|---|
Flux.firstWithSignal(...) | Observable.ambArray(...) | firstWithSignal(Publisher...) | 静态方法；取最先发出任意信号（onNext/onError/onComplete）的源 |
Flux.first(...)（已废弃） | Observable.ambArray(...) | first(Publisher...) | 3.8 中标记 @Deprecated，等价于 firstWithSignal，不要在新代码里使用 |
⚠️ 差异点
first / firstWithSignal 是静态工厂，不是实例操作符。它们接受多个 Publisher，语义是“竞争源中谁先发信号就选谁”，对应 RxJava 的 ambArray。- RxJava 的
firstElement() 不是对应物。它只处理一条流，返回 Maybe<T>（第一个元素），不能用来在多个源之间竞争。 - Reactor
zip 对空流的处理：如果任一源流为空，结果为空。 - RxJava
combineLatest 初始不发射，直到所有源至少有一个元素。
典型场景
- 多字段校验：
zip(nameValid, emailValid, phoneValid, (a,b,c) -> a && b && c) - 行情汇总：
combineLatest(priceStream, volumeStream, (p,v) -> new Quote(p,v))
四、错误处理操作符
| Reactor | RxJava | 签名 | 语义 |
|---|
onErrorReturn | onErrorReturn | onErrorReturn(T fallback) | 替换为默认值；没有 onErrorReturnItem 这个名字 |
onErrorResume | onErrorResumeNext | onErrorResume(Function<Throwable,Publisher<T>>) | 切换备用流 |
onErrorMap | — | onErrorMap(Function<Throwable,Throwable>) | 转换异常类型 |
onErrorContinue | — | onErrorContinue(BiConsumer) | ⚠️ 跳过错误元素，非标准 |
retry | retry | retry(long numRetries) | 重试 N 次 |
retryWhen | retryWhen | retryWhen(Retry) | 条件重试；3.8 只有 Retry 重载，没有 Function 重载 |
timeout | timeout | timeout(Duration) / timeout(long, TimeUnit) | 元素间隔超时 |
⚠️ 陷阱
onErrorContinue 不是 RS 标准，只存在于 Reactor。它会跳过错误元素继续流，可能掩盖问题。- Reactor 没有
onErrorReturnItem。把 onErrorReturn 写成不存在的名字，编译期就会失败；需要“出错时给默认值”就用 onErrorReturn(T)。 Mono.onErrorReturn 只有三个重载：(T)、(Class<? extends Throwable>, T)、(Predicate<? super Throwable>, T)。没有 Function<Throwable, T> 重载；要根据异常计算返回值，用 onErrorResume(e -> Mono.just(f(e)))。retry 是「重新订阅」不是「重试当前元素」，整个流从头开始。
五、背压控制操作符
| Reactor | RxJava Flowable | 签名 | 语义 |
|---|
onBackpressureBuffer | onBackpressureBuffer | onBackpressureBuffer() | 缓冲 |
onBackpressureBuffer(int) | 同上 | onBackpressureBuffer(int maxSize) | 有界缓冲 |
onBackpressureDrop | onBackpressureDrop | onBackpressureDrop() | 丢弃溢出 |
onBackpressureLatest | onBackpressureLatest | onBackpressureLatest() | 保留最新 |
onBackpressureError | — | onBackpressureError() | 报错终止（RxJava 无此方法） |
limitRate | — | limitRate(int prefetch) | 批量 request（RxJava 无） |
limitRequest | — | limitRequest(long) | 限制上游总请求量（RxJava 无） |
⚠️ 背压操作符的三个坑
- 两参数
onBackpressureBuffer(int, Consumer) 不是“溢出即丢弃”。第二个参数只是溢出回调，默认策略仍是 BufferOverflowStrategy.ERROR，缓冲区满后流会以 OverflowException 终止。要丢弃并继续，必须用三参数重载显式指定 DROP_LATEST 或 DROP_OLDEST。 - 这些操作符会向上游请求
Long.MAX_VALUE。它们不是把背压传回上游，而是在中间接管需求；把它接在 Flux.range 这类本来支持背压的源后面，会把源变成“推”模型。 onBackpressureLatest() 的丢弃要走 doOnDiscard 统计，它没有溢出回调参数。
Reactor 独有
| Reactor | 语义 |
|---|
onBackpressureBuffer(int, BufferOverflowStrategy) | 有界缓冲 + 溢出策略 |
onBackpressureBuffer(int, Consumer, BufferOverflowStrategy) | 有界缓冲 + 溢出回调 + 策略 |
六、实用工具操作符
副作用（调试与监控）
| Reactor | RxJava | 触发时机 |
|---|
doOnNext | doOnNext | 每个 onNext |
doOnError | doOnError | onError |
doOnComplete | doOnComplete | onComplete |
doOnCancel | Flowable.doOnCancel / Observable.doOnDispose | 取消 |
doOnSubscribe | doOnSubscribe | onSubscribe |
doOnRequest | — | request |
doOnTerminate | doOnTerminate | onError 或 onComplete |
doAfterTerminate | doAfterTerminate | 终止之后 |
流量控制
| Reactor | RxJava | 语义 |
|---|
delayElements | — | 每个元素延迟 |
delaySequence | — | 序列整体延迟 |
sample | sample | 按时间采样 |
| — | throttleFirst | 时间窗口内取第一个；Reactor 没有 throttleFirst |
| — | throttleLatest | 时间窗口内取最新；Reactor 没有 throttleLatest |
日志与调试
| Reactor | RxJava | 语义 |
|---|
log() | — | 自动打印所有信号 |
checkpoint() | — | 检查点追踪 |
metrics() | — | Micrometer 指标 |
操作符选型速查
| 需求 | Reactor | RxJava |
|---|
| 同步变换 | map | map |
| 并发 HTTP 请求 | flatMap | flatMap |
| 有序数据库查询 | concatMap | concatMap |
| 取最新请求结果 | switchMap | switchMap |
| 合并两个 REST 结果 | zip | zip |
| 实时行情合并 | combineLatest | combineLatest |
| 失败重试 3 次 | retry(3) | retry(3) |
| 指数退避重试 | Retry.backoff() | retryWhen 手写 |
| 3 秒超时 | timeout(Duration.ofSeconds(3)) | timeout |
| 缓冲到内存 | onBackpressureBuffer | onBackpressureBuffer |
| 允许丢数据 | onBackpressureDrop | onBackpressureDrop |
| 只保留最新 | onBackpressureLatest | onBackpressureLatest |
常见陷阱速查
| 操作符 | 陷阱 | 解决方案 |
|---|
flatMap | 默认并发 256，可能压垮下游 | 指定 flatMap(func, concurrency) |
concatMap | 串行吞吐量低 | 确认是否需要顺序，否则用 flatMap |
retry | 无限重试雪崩 | 始终限制重试次数 |
timeout | 不中断阻塞操作 | 配合 subscribeOn(boundedElastic()) |
onBackpressureBuffer | 无界 OOM | 指定上限和溢出策略 |
cache() | 永远缓存，内存增长 | 用 replay().limit() 替代 |
distinct | 需要 Set 存储，内存增长 | 确认数据集大小 |
groupBy | 每个分组是一个流，流过多 OOM | 限制分组数量 |
操作符命名对照
Reactor 与 RxJava 的操作符命名基本一致，但存在少数差异：
| Reactor | RxJava | 语义 |
|---|
doOnCancel | Flowable.doOnCancel / Observable.doOnDispose | 取消回调 |
onErrorResume | onErrorResumeNext | 错误恢复 |
limitRate | — | 批量 request（Reactor 独有） |
limitRequest | — | 限制总请求（Reactor 独有） |
switchIfEmpty | switchIfEmpty | 空流切换 |
defaultIfEmpty | defaultIfEmpty | 空流默认值 |
Mono 与 Flux 的专属操作符
Mono 独有
| 操作符 | 语义 |
|---|
defaultIfEmpty | 空时提供默认值 |
switchIfEmpty(Mono) | 空时切换备用 Mono |
hasElement | 是否有元素 |
ignoreElement | 忽略值，只关心终止 |
then | 完成后执行下一个 Mono |
zipWhen | 与另一个 Mono 组合 |
Flux 独有
| 操作符 | 语义 |
|---|
collectList | 收集为 List |
collectMap | 收集为 Map |
reduce | 归约 |
scan | 累积扫描 |
window | 分窗口 |
buffer | 缓冲为 List |
groupBy | 分组 |
sort | 排序 |
distinct | 去重 |
sample | 按时间采样 |
sampleTimeout | 按时间窗口采样 |
速查卡片：我最常用的 10 个操作符
map — 数据变换（90% 的链都有）flatMap — 并发异步展开filter — 过滤zip — 多流配对timeout — 超时控制retryWhen — 退避重试onErrorResume — 降级doOnNext — 调试/监控subscribeOn — 线程切换onBackpressureBuffer — 背压控制
阅读导航
上一篇：14.3 v1 验收：RS 合规、背压可观测、文档
下一篇：附录 B 调度器选择指南
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。