附2：Reactor 3 之选择合适的操作符——响应式Spring的道法术器

标签：Spring WebFlux   响应式编程   

本系列文章索引《响应式Spring的道法术器》
前情提要 Reactor Operators

本节的内容来自我翻译的Reactor 3 参考文档——如何选择操作符。由于部分朋友打开github.io网速比较慢或上不去，贴出来方便大家查阅。

如果一个操作符是专属于 Flux 或 Mono 的，那么会给它注明前缀。
公共的操作符没有前缀。如果一个具体的用例涉及多个操作符的组合，这里以方法调用的方式展现，
会以一个点（.）开头，并将参数置于圆括号内，比如： .methodCall(parameter)。

1）创建一个新序列，它...

• 发出一个 T，我已经有了：just
  • ...基于一个 Optional<T>：Mono#justOrEmpty(Optional<T>)
  • ...基于一个可能为 null 的 T：Mono#justOrEmpty(T)
• 发出一个 T，且还是由 just 方法返回
  • ...但是“懒”创建的：使用 Mono#fromSupplier 或用 defer 包装 just
• 发出许多 T，这些元素我可以明确列举出来：Flux#just(T...)
• 基于迭代数据结构:
  • 一个数组：Flux#fromArray
  • 一个集合或 iterable：Flux#fromIterable
  • 一个 Integer 的 range：Flux#range
  • 一个 Stream 提供给每一个订阅：Flux#fromStream(Supplier<Stream>)
• 基于一个参数值给出的源：
  • 一个 Supplier<T>：Mono#fromSupplier
  • 一个任务：Mono#fromCallable，Mono#fromRunnable
  • 一个 CompletableFuture<T>：Mono#fromFuture
• 直接完成：empty
• 立即生成错误：error
  • ...但是“懒”的方式生成 Throwable：error(Supplier<Throwable>)
• 什么都不做：never
• 订阅时才决定：defer
• 依赖一个可回收的资源：using
• 可编程地生成事件（可以使用状态）:
  • 同步且逐个的：Flux#generate
  • 异步（也可同步）的，每次尽可能多发出元素：Flux#create
    （Mono#create 也是异步的，只不过只能发一个）

2）对序列进行转化

• 我想转化一个序列：

  • 1对1地转化（比如字符串转化为它的长度）：map
  • ...类型转化：cast
  • ...为了获得每个元素的序号：Flux#index
  • 1对n地转化（如字符串转化为一串字符）：flatMap + 使用一个工厂方法
  • 1对n地转化可自定义转化方法和/或状态：handle
  • 对每一个元素执行一个异步操作（如对 url 执行 http 请求）：flatMap + 一个异步的返回类型为 Publisher 的方法
  • ...忽略一些数据：在 flatMap lambda 中根据条件返回一个 Mono.empty()
  • ...保留原来的序列顺序：Flux#flatMapSequential（对每个元素的异步任务会立即执行，但会将结果按照原序列顺序排序）
  • ...当 Mono 元素的异步任务会返回多个元素的序列时：Mono#flatMapMany

• 我想添加一些数据元素到一个现有的序列：

  • 在开头添加：Flux#startWith(T...)
  • 在最后添加：Flux#concatWith(T...)

• 我想将 Flux 转化为集合（一下都是针对 Flux 的）

  • 转化为 List：collectList，collectSortedList
  • 转化为 Map：collectMap，collectMultiMap
  • 转化为自定义集合：collect
  • 计数：count
  • reduce 算法（将上个元素的reduce结果与当前元素值作为输入执行reduce方法，如sum） reduce
  • ...将每次 reduce 的结果立即发出：scan
  • 转化为一个 boolean 值：
  • 对所有元素判断都为true：all
  • 对至少一个元素判断为true：any
  • 判断序列是否有元素（不为空）：hasElements
  • 判断序列中是否有匹配的元素：hasElement

• 我想合并 publishers...

  • 按序连接：Flux#concat 或 .concatWith(other)
  • ...即使有错误，也会等所有的 publishers 连接完成：Flux#concatDelayError
  • ...按订阅顺序连接（这里的合并仍然可以理解成序列的连接）：Flux#mergeSequential
  • 按元素发出的顺序合并（无论哪个序列的，元素先到先合并）：Flux#merge / .mergeWith(other)
  • ...元素类型会发生变化：Flux#zip / Flux#zipWith
  • 将元素组合：
  • 2个 Monos 组成1个 Tuple2：Mono#zipWith
  • n个 Monos 的元素都发出来后组成一个 Tuple：Mono#zip
  • 在终止信号出现时“采取行动”：
  • 在 Mono 终止时转换为一个 Mono<Void>：Mono#and
  • 当 n 个 Mono 都终止时返回 Mono<Void>：Mono#when
  • 返回一个存放组合数据的类型，对于被合并的多个序列：
    • 每个序列都发出一个元素时：Flux#zip
    • 任何一个序列发出元素时：Flux#combineLatest
  • 只取各个序列的第一个元素：Flux#first，Mono#first，mono.or<br/>(otherMono).or(thirdMono)，`flux.or(otherFlux).or(thirdFlux)
  • 由一个序列触发（类似于 flatMap，不过“喜新厌旧”）：switchMap
  • 由每个新序列开始时触发（也是“喜新厌旧”风格）：switchOnNext

• 我想重复一个序列：repeat

  • ...但是以一定的间隔重复：Flux.interval(duration).flatMap(tick -> myExistingPublisher)

• 我有一个空序列，但是...

  • 我想要一个缺省值来代替：defaultIfEmpty
  • 我想要一个缺省的序列来代替：switchIfEmpty

• 我有一个序列，但是我对序列的元素值不感兴趣：ignoreElements

  • ...并且我希望用 Mono 来表示序列已经结束：then
  • ...并且我想在序列结束后等待另一个任务完成：thenEmpty
  • ...并且我想在序列结束之后返回一个 Mono：Mono#then(mono)
  • ...并且我想在序列结束之后返回一个值：Mono#thenReturn(T)
  • ...并且我想在序列结束之后返回一个 Flux：thenMany

• 我有一个 Mono 但我想延迟完成...

  • ...当有1个或N个其他 publishers 都发出（或结束）时才完成：Mono#delayUntilOther
  • ...使用一个函数式来定义如何获取“其他 publisher”：Mono#delayUntil(Function)
• 我想基于一个递归的生成序列的规则扩展每一个元素，然后合并为一个序列发出：
  • ...广度优先：expand(Function)
  • ...深度优先：expandDeep(Function)

3）“窥视”（只读）序列

• 再不对序列造成改变的情况下，我想：

  • 得到通知或执行一些操作：
  • 发出元素：doOnNext
  • 序列完成：Flux#doOnComplete，Mono#doOnSuccess
  • 因错误终止：doOnError
  • 取消：doOnCancel
  • 订阅时：doOnSubscribe
  • 请求时：doOnRequest
  • 完成或错误终止：doOnTerminate（Mono的方法可能包含有结果）
    • 但是在终止信号向下游传递 之后 ：doAfterTerminate
  • 所有类型的信号（Signal）：Flux#doOnEach
  • 所有结束的情况（完成complete、错误error、取消cancel）：doFinally
  • 记录日志：log
• 我想知道所有的事件:
  • 每一个事件都体现为一个 single 对象：
  • 执行 callback：doOnEach
  • 每个元素转化为 single 对象：materialize
    • ...在转化回元素：dematerialize
  • 转化为一行日志：log

4）过滤序列

• 我想过滤一个序列

  • 基于给定的判断条件：filter
  • ...异步地进行判断：filterWhen
  • 仅限于指定类型的对象：ofType
  • 忽略所有元素：ignoreElements
  • 去重:
  • 对于整个序列：Flux#distinct
  • 去掉连续重复的元素：Flux#distinctUntilChanged

• 我只想要一部分序列：

  • 只要 N 个元素：
  • 从序列的第一个元素开始算：Flux#take(long)
    • ...取一段时间内发出的元素：Flux#take(Duration)
    • ...只取第一个元素放到 Mono 中返回：Flux#next()
    • ...使用 request(N) 而不是取消：Flux#limitRequest(long)
  • 从序列的最后一个元素倒数：Flux#takeLast
  • 直到满足某个条件（包含）：Flux#takeUntil（基于判断条件），Flux#takeUntilOther（基于对 publisher 的比较）
  • 直到满足某个条件（不包含）：Flux#takeWhile
  • 最多只取 1 个元素：
  • 给定序号：Flux#elementAt
  • 最后一个：.takeLast(1)
    • ...如果为序列空则发出错误信号：Flux#last()
    • ...如果序列为空则返回默认值：Flux#last(T)
  • 跳过一些元素：
  • 从序列的第一个元素开始跳过：Flux#skip(long)
    • ...跳过一段时间内发出的元素：Flux#skip(Duration)
  • 跳过最后的 n 个元素：Flux#skipLast
  • 直到满足某个条件（包含）：Flux#skipUntil（基于判断条件），Flux#skipUntilOther （基于对 publisher 的比较）
  • 直到满足某个条件（不包含）：Flux#skipWhile
  • 采样：
  • 给定采样周期：Flux#sample(Duration)
    • 取采样周期里的第一个元素而不是最后一个：sampleFirst
  • 基于另一个 publisher：Flux#sample(Publisher)
  • 基于 publisher“超时”：Flux#sampleTimeout （每一个元素会触发一个 publisher，如果这个 publisher 不被下一个元素触发的 publisher 覆盖就发出这个元素）
• 我只想要一个元素（如果多于一个就返回错误）...
  • 如果序列为空，发出错误信号：Flux#single()
  • 如果序列为空，发出一个缺省值：Flux#single(T)
  • 如果序列为空就返回一个空序列：Flux#singleOrEmpty

5）错误处理

• 我想创建一个错误序列：error...

  • ...替换一个完成的 Flux：.concat(Flux.error(e))
  • ...替换一个完成的 Mono：.then(Mono.error(e))
  • ...如果元素超时未发出：timeout
  • ...“懒”创建：error(Supplier<Throwable>)

• 我想要类似 try/catch 的表达方式：

  • 抛出异常：error
  • 捕获异常：
  • 然后返回缺省值：onErrorReturn
  • 然后返回一个 Flux 或 Mono：onErrorResume
  • 包装异常后再抛出：.onErrorMap(t -> new RuntimeException(t))
  • finally 代码块：doFinally
  • Java 7 之后的 try-with-resources 写法：using 工厂方法

• 我想从错误中恢复...

  • 返回一个缺省的：
  • 的值：onErrorReturn
  • Publisher：Flux#onErrorResume 和 Mono#onErrorResume
  • 重试：retry
  • ...由一个用于伴随 Flux 触发：retryWhen
• 我想处理回压错误（向上游发出“MAX”的 request，如果下游的 request 比较少，则应用策略）...
  • 抛出 IllegalStateException：Flux#onBackpressureError
  • 丢弃策略：Flux#onBackpressureDrop
  • ...但是不丢弃最后一个元素：Flux#onBackpressureLatest
  • 缓存策略（有限或无限）：Flux#onBackpressureBuffer
  • ...当有限的缓存空间用满则应用给定策略：Flux#onBackpressureBuffer 带有策略 BufferOverflowStrategy

6） 基于时间的操作

• 我想将元素转换为带有时间信息的 Tuple2<Long, T>...

  • 从订阅时开始：elapsed
  • 记录时间戳：timestamp

• 如果元素间延迟过长则中止序列：timeout

• 以固定的周期发出元素：Flux#interval

• 在一个给定的延迟后发出 0：static Mono.delay.

• 我想引入延迟：
  • 对每一个元素：Mono#delayElement，Flux#delayElements
  • 延迟订阅：delaySubscription

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)
    • ...触发判断条件的元素会分到下一个buffer：.bufferUntil(predicate, true)
    • ...满足条件的元素在一个buffer，直到不满足条件的元素发出开始下一buffer：bufferWhile(Predicate)
  • 通过另一个 Publisher 的每一个 onNext 信号来拆分序列：buffer(Publisher)，bufferWhen
  • 拆分到指定类型的 "collection"：buffer(int, Supplier<C>)
• 我想将 Flux<T> 中具有共同特征的元素分组到子 Flux：groupBy(Function<T,K>)（注意返回值是 Flux<GroupedFlux<K, T>>，每一个 GroupedFlux 具有相同的 key 值 K，可以通过 key() 方法获取）。

8）回到同步的世界

• 我有一个 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

附2：Reactor 3 之选择合适的操作符——响应式Spring的道法术器

标签：Spring WebFlux   响应式编程   

原文地址：http://blog.51cto.com/liukang/2094073