我需要哪个操作符？

Doubao 中英对照 Which operator do I need?

目录

• 1. 创建新的序列…​
• 2. 转换已有序列
• 3. 窥探序列
• 4. 过滤序列
• 5. 处理错误
• 6. 与时间打交道
• 7. 拆分 Flux
• 8. 回到同步世界
• 9. 将 Flux 多播给多个 Subscriber

提示

在本节中，如果某个操作符为 Flux 或 Mono 所特有，它会相应地加上前缀并附上链接，如下所示：Flux#fromArray。通用操作符没有前缀，并同时提供两个实现的链接，例如：just (Flux|Mono)。当某个特定用例由多个操作符组合实现时，它会以方法调用的形式呈现，带前导点号，参数放在括号中，如下所示：.methodCall(parameter)。

我想要处理：

• 创建新的序列…​

• 转换已有序列

• 窥探序列

• 过滤序列

• 处理错误

• 与时间打交道

• 拆分 Flux

• 回到同步世界

• 将 Flux 多播给多个 Subscriber

1. 创建新的序列…​

• 发出一个 T，且我已经持有：just (Flux|Mono)

  • …​从一个 Optional<T>：Mono#justOrEmpty(Optional<T>)

  • …​从一个可能为 null 的 T：Mono#justOrEmpty(T)

• 发出由某个方法返回的 T：同样使用 just (Flux|Mono)

  • …​但惰性捕获：使用 Mono#fromSupplier，或将 just (Flux|Mono) 包装在 defer (Flux|Mono) 中

• 发出我可以显式枚举的若干个 T：Flux#just(T…​)

• 遍历：

  • 一个数组：Flux#fromArray

  • 一个集合或可迭代对象：Flux#fromIterable

  • 一个整数区间：Flux#range

  • 为每次 Subscription 提供的一个 Stream：Flux#fromStream(Supplier<Stream>)

• 从各种单值源发出，例如：

  • 一个 Supplier<T>：Mono#fromSupplier

  • 一个任务：Mono#fromCallable, Mono#fromRunnable

  • 一个 CompletableFuture<T>：Mono#fromFuture

• 直接完成：empty (Flux|Mono)

• 立即产生错误：error (Flux|Mono)

  • …​但惰性地构建 Throwable：error(Supplier<Throwable>) (Flux|Mono)

• 什么都不做：never (Flux|Mono)

• 在订阅时才决定：defer (Flux|Mono)

• 依赖于一个可释放资源：using (Flux|Mono)

• 以编程方式生成事件（可使用状态）：

  • 同步地、逐个地：Flux#generate

  • 异步地（也可以是同步），一次处理中可发出多个：Flux#create（Mono#create 同理，但不具备多次发出的特性）

2. 转换已有序列

• 我想转换已有数据：

  • 一对一地（例如将字符串转换为其长度）：map (Flux|Mono)

    • …​仅仅是做类型转换：cast (Flux|Mono)

    • …​为了标出每个源值的索引：Flux#index

  • 一对多地（例如将字符串转换为其字符）：flatMap (Flux|Mono) + 使用一个工厂方法

  • 一对多地，并为每个源元素和/或状态提供可编程的行为：handle (Flux|Mono)

  • 为每个源元素运行一个异步任务（例如将 url 转换为 http 请求）：flatMap (Flux|Mono) + 一个返回异步 Publisher-的方法

    • …​忽略某些数据：在 flatMap lambda 中有条件地返回 Mono.empty()

    • …​保持原始序列顺序：Flux#flatMapSequential（它会立即触发异步过程，但会对结果重新排序）

    • …​异步任务可返回多个值，源为 Mono：Mono#flatMapMany

• 我想向已有序列添加预设元素：

  • 在开头：Flux#startWith(T…​)

  • 在末尾：Flux#concatWithValues(T…​)

• 我想聚合一个 Flux：（下文省略 Flux# 前缀）

  • 聚合为一个 List：collectList, collectSortedList

  • 聚合为一个 Map：collectMap, collectMultiMap

  • 聚合为任意容器：collect

  • 聚合为序列的大小：count

  • 在每个元素之间应用一个函数（例如累加和）：reduce

    • …​但发出每个中间值：scan

  • 通过一个谓词聚合为一个布尔值：

    • 应用于所有值（AND）：all

    • 应用于至少一个值（OR）：any

    • 测试是否存在任意值：hasElements （在 hasElement 中有对应的 Mono 版本）

    • 测试是否存在某个特定值：hasElement(T)

• 我想组合多个发布者…​

  • 按顺序：Flux#concat 或 .concatWith(other) (Flux|Mono)

    • …​但将任何错误延迟到其余发布者都发出之后：Flux#concatDelayError

    • …​但急切地订阅后续发布者：Flux#mergeSequential

  • 按发出顺序（组合的各项随到随发）：Flux#merge / .mergeWith(other) (Flux|Mono)

    • …​带不同类型（转换式合并）：Flux#zip / Flux#zipWith

  • 通过配对值：

    • 将 2 个 Mono 配对为一个 Tuple2：Mono#zipWith

    • 当 n 个 Mono 全部完成时：Mono#zip

  • 通过协调它们的终止：

    • 将 1 个 Mono 与任意源组合为一个 Mono<Void>：Mono#and

    • 当 n 个源全部完成时：Mono#when

    • 聚合为任意容器类型：

      • 每次各方都发出一个值时：Flux#zip（以最小基数为准）

      • 任一侧每到达一个新值时：Flux#combineLatest

  • 选择第一个…​

    • 产生_值_（onNext）的发布者：firstWithValue (Flux|Mono)

    • 产生_任意信号_的发布者：firstWithSignal (Flux|Mono)

  • 由源序列中的元素触发：switchMap（每个源元素被映射为一个 Publisher）

  • 由一系列发布者中下一个发布者的开始触发：switchOnNext

• 我想重复一个已有序列：repeat (Flux|Mono)

  • …​但按固定时间间隔：Flux.interval(duration).flatMap(tick → myExistingPublisher)

• 我有一个空序列，但…​

  • 我想得到一个值代替：defaultIfEmpty (Flux|Mono)

  • 我想改用另一个序列：switchIfEmpty (Flux|Mono)

• 我有一个序列，但我对值不感兴趣：ignoreElements (Flux.ignoreElements()|Mono.ignoreElement())

  • …​并且我想把完成表示为一个 Mono<Void>：then (Flux|Mono)

  • …​并且我想在末尾等待另一个任务完成：thenEmpty (Flux|Mono)

  • …​并且我想在末尾切换到另一个 Mono：Mono#then(mono)

  • …​并且我想在末尾发出单个值：Mono#thenReturn(T)

  • …​并且我想在末尾切换到一个 Flux：thenMany (Flux|Mono)

• 我有一个 Mono，想推迟它的完成…​

  • …​直到由该值派生的另一个发布者完成：Mono#delayUntil(Function)

• 我想把元素递归地展开为一个序列图，并发出其组合…​

  • …​按广度优先展开图：expand(Function) (Flux|Mono)

  • …​按深度优先展开图：expandDeep(Function) (Flux|Mono)

3. 窥探序列

• 在不修改最终序列的前提下，我想：

  • 在以下时机获得通知 / 执行额外行为（有时称为"副作用"）：

    • 发出元素时：doOnNext (Flux|Mono)

    • 完成时：Flux#doOnComplete, Mono#doOnSuccess（如有结果，会包含结果）

    • 错误终止时：doOnError (Flux|Mono)

    • 取消时：doOnCancel (Flux|Mono)

    • 序列的"开始"：doFirst (Flux|Mono)

      • 它绑定到 Publisher#subscribe(Subscriber)

    • 订阅之后：doOnSubscribe (Flux|Mono)

      • subscribe 之后的 Subscription 确认

      • 它绑定到 Subscriber#onSubscribe(Subscription)

    • 请求时：doOnRequest (Flux|Mono)

    • 完成或出错时：doOnTerminate (Flux|Mono)

      • 但在其向下游传播_之后_：doAfterTerminate (Flux|Mono)

    • 任意类型的信号，表示为一个 Signal：doOnEach (Flux|Mono)

    • 任意终止条件（完成、错误、取消）：doFinally (Flux|Mono)

  • 记录内部发生的情况：log (Flux|Mono)

• 我想获知所有事件：

  • 每个事件表示为 Signal 对象：

    • 在序列之外的回调中：doOnEach (Flux|Mono)

    • 代替原始的 onNext 发出：materialize (Flux|Mono)

      • …​再还原回 onNext：dematerialize (Flux|Mono)

  • 作为日志中的一行：log (Flux|Mono)

4. 过滤序列

• 我想过滤一个序列：

  • 基于任意条件：filter (Flux|Mono)

    • …​该条件是异步计算的：filterWhen (Flux|Mono)

  • 按发出对象的类型进行限制：ofType (Flux|Mono)

  • 完全忽略值：ignoreElements (Flux.ignoreElements()|Mono.ignoreElement())

  • 忽略重复项：

    • 在整个序列中（逻辑集合）：Flux#distinct

    • 在相继发出的项之间（去重）：Flux#distinctUntilChanged

• 我想只保留序列的一个子集：

  • 取 N 个元素：

    • 从序列开头：Flux#take(long)

      • …​向上游请求无界量：Flux#take(long, false)

      • …​基于时长：Flux#take(Duration)

      • …​只取第一个元素，作为一个 Mono：Flux#next()

    • 从序列末尾：Flux#takeLast

    • 直到满足某个条件（含该元素）：Flux#takeUntil（基于谓词）, Flux#takeUntilOther（基于伴随发布者）

    • 在满足条件期间持续（不含该元素）：Flux#takeWhile

  • 最多取 1 个元素：

    • 在特定位置：Flux#elementAt

    • 在末尾：.takeLast(1)

      • …​如果为空则发出错误：Flux#last()

      • …​如果为空则发出默认值：Flux#last(T)

  • 跳过元素：

    • 从序列开头：Flux#skip(long)

      • …​基于时长：Flux#skip(Duration)

    • 从序列末尾：Flux#skipLast

    • 直到满足某个条件（含该元素）：Flux#skipUntil（基于谓词）, Flux#skipUntilOther（基于伴随发布者）

    • 在满足条件期间持续（不含该元素）：Flux#skipWhile

  • 对元素采样：

    • 按时长：Flux#sample(Duration)

      • 但保留采样窗口中的第一个元素而非最后一个：sampleFirst

    • 按基于发布者的窗口：Flux#sample(Publisher)

    • 基于发布者"超时"：Flux#sampleTimeout（每个元素触发一个发布者，如果该发布者与下一个不重叠则发出该元素）

• 我预期最多 1 个元素（多于一个则报错）…​

  • 并且如果序列为空我希望报错：Flux#single()

  • 并且如果序列为空我希望得到默认值：Flux#single(T)

  • 并且我也接受空序列：Flux#singleOrEmpty

5. 处理错误

• 我想创建一个出错序列：error (Flux|Mono)…​

  • …​用来替换一个成功的 Flux 的完成：.concat(Flux.error(e))

  • …​用来替换一个成功的 Mono 的_发出_：.then(Mono.error(e))

  • …​如果 onNext 之间耗时过长：timeout (Flux|Mono)

  • …​惰性地：error(Supplier<Throwable>) (Flux|Mono)

• 我想要以下 try/catch 的等价物：

  • 抛出：error (Flux|Mono)

  • 捕获异常：

    • 并回退到一个默认值：onErrorReturn (Flux|Mono)

    • 并吞掉错误（即完成）：onErrorComplete (Flux|Mono)

    • 并回退到另一个 Flux 或 Mono：onErrorResume (Flux|Mono)

    • 并包装后重新抛出：.onErrorMap(t → new RuntimeException(t)) (Flux|Mono)

  • finally 块：doFinally (Flux|Mono)

  • Java 7 的 using 模式：using (Flux|Mono) 工厂方法

• 我想从错误中恢复…​

  • 通过回退：

    • 到一个值：onErrorReturn (Flux|Mono)

    • 到完成（"吞掉"错误）：onErrorComplete (Flux|Mono)

    • 到一个 Publisher 或 Mono，可根据错误使用不同的实现：Flux#onErrorResume 和 Mono#onErrorResume

  • 通过重试…​

    • …​使用简单策略（最大尝试次数）：retry() (Flux|Mono), retry(long) (Flux|Mono)

    • …​由一个伴随控制 Flux 触发：retryWhen (Flux|Mono)

    • …​使用标准退避策略（带抖动的指数退避）：retryWhen(Retry.backoff(…​)) (Flux|Mono)（另见 Retry 中的其他工厂方法）

• 我想处理背压"错误"（向上游请求最大量，并在下游产生的请求不足时应用策略）…​

  • 通过抛出一个特殊的 IllegalStateException：Flux#onBackpressureError

  • 通过丢弃多余的值：Flux#onBackpressureDrop

    • …​但保留看到的最后一个：Flux#onBackpressureLatest

  • 通过缓冲多余的值（有界或无界）：Flux#onBackpressureBuffer

    • …​并在有界缓冲区也溢出时应用策略：带 BufferOverflowStrategy 的 Flux#onBackpressureBuffer

6. 与时间打交道

• 我想把发出与测得的时间关联起来…​

  • …​以最佳可用精度并提供多功能数据：timed (Flux|Mono)

    • Timed<T>#elapsed() 获取距上次 onNext 的 Duration

    • Timed<T>#timestamp() 获取纪元时间戳（毫秒精度）的 Instant 表示

    • Timed<T>#elapsedSinceSubcription() 获取距订阅（而非上次 onNext）的 Duration

    • 对于经过的 Duration 可具有纳秒精度

  • …​作为一个（旧式的）Tuple2<Long, T>…​

    • 距上次 onNext：elapsed (Flux|Mono)

    • 自时间之初（准确地说，计算机时间）起：timestamp (Flux|Mono)

• 我希望在两次发出之间延迟过长时中断序列：timeout (Flux|Mono)

• 我想从时钟获取节拍，即固定的时间间隔：Flux#interval

• 我想在初始延迟后发出单个 0：静态的 Mono.delay。

• 我想引入延迟：

  • 在每个 onNext 信号之间：Mono#delayElement, Flux#delayElements

  • 在订阅发生之前：delaySubscription (Flux|Mono)

7. 拆分 Flux

• 我想把一个 Flux<T> 拆分为 Flux<Flux<T>>，按边界条件：

  • 按大小：window(int)

    • …​带重叠或丢弃的窗口：window(int, int)

  • 按时间 window(Duration)

    • …​带重叠或丢弃的窗口：window(Duration, Duration)

  • 按大小_或_时间（达到计数或经过超时时间即关闭窗口）：windowTimeout(int, Duration)

  • 基于元素上的谓词：windowUntil

    • …​…把触发边界的元素发到下一个窗口（cutBefore 变体）：.windowUntil(predicate, true)

    • …​在元素匹配谓词期间保持窗口开启：windowWhile（不匹配的元素不发出）

  • 由任意边界驱动，该边界表示为控制 Publisher 中的 onNext：window(Publisher), windowWhen

• 我想拆分一个 Flux<T>，并把边界内的元素缓冲在一起…​

  • 聚合为 List：

    • 按大小边界：buffer(int)

      • …​带重叠或丢弃的缓冲：buffer(int, int)

    • 按时长边界：buffer(Duration)

      • …​带重叠或丢弃的缓冲：buffer(Duration, Duration)

    • 按大小_或_时长边界：bufferTimeout(int, Duration)

    • 按任意条件边界：bufferUntil(Predicate)

      • …​把触发边界的元素放入下一个缓冲区：.bufferUntil(predicate, true)

      • …​在谓词匹配期间缓冲，并丢弃触发边界的元素：bufferWhile(Predicate)

    • 由任意边界驱动，该边界表示为控制 Publisher 中的 onNext：buffer(Publisher), bufferWhen

  • 聚合为任意"集合"类型 C：使用诸如 buffer(int, Supplier<C>) 的变体

• 我想拆分一个 Flux<T>，使具有相同特征的元素归入同一个子 Flux：groupBy(Function<T,K>) 提示：注意它返回一个 Flux<GroupedFlux<K, T>>，每个内层 GroupedFlux 共享相同的 K 键，可通过 key() 访问。

8. 回到同步世界

注意：除 Mono#toFuture 之外，所有这些方法如果在标记为"仅非阻塞"的 Scheduler（默认即 parallel() 和 single()）中调用，都会抛出 UnsupportedOperatorException。

• 我有一个 Flux<T>，我想：

  • 阻塞直到能拿到第一个元素：Flux#blockFirst

    • …​带超时：Flux#blockFirst(Duration)

  • 阻塞直到能拿到最后一个元素（若为空则为 null）：Flux#blockLast

    • …​带超时：Flux#blockLast(Duration)

  • 同步地切换为一个 Iterable<T>：Flux#toIterable

  • 同步地切换为一个 Java 8 Stream<T>：Flux#toStream

• 我有一个 Mono<T>，我想：

  • 阻塞直到能拿到值：Mono#block

    • …​带超时：Mono#block(Duration)

  • 得到一个 CompletableFuture<T>：Mono#toFuture

9. 将 Flux 多播给多个 Subscriber

• 我想把多个 Subscriber 连接到一个 Flux：

  • 并通过 connect() 决定何时触发源：publish()（返回一个 ConnectableFlux）

  • 并立即触发源（迟到的订阅者看到稍后的数据）：share() (Flux|Mono)

  • 并在足够多的订阅者注册后永久连接源：.publish().autoConnect(n)

  • 并在订阅者数量高于/低于阈值时自动连接和取消源：.publish().refCount(n)

    • …​但在取消前给新订阅者一个加入的机会：.publish().refCount(n, Duration)

• 我想缓存来自某个 Publisher 的数据，并把它重放给后来的订阅者：

  • 最多 n 个元素：cache(int)

  • 缓存 Duration 内看到的最新元素（存活时间）：cache(Duration) (Flux|Mono)

    • …​但保留的元素不超过 n 个：cache(int, Duration)

  • 但不立即触发源：Flux#replay（返回一个 ConnectableFlux）