附2：Reactor 3 之选择合适的操做符——响应式Spring的道法术器
本系列文章索引《响应式Spring的道法术器》
前情提要 Reactor Operatorshtml
本节的内容来自我翻译的Reactor 3 参考文档——如何选择操做符。因为部分朋友打开github.io网速比较慢或上不去，贴出来方便你们查阅。react
若是一个操做符是专属于
Flux或Mono的，那么会给它注明前缀。
公共的操做符没有前缀。若是一个具体的用例涉及多个操做符的组合，这里以方法调用的方式展示，
会以一个点（.）开头，并将参数置于圆括号内，好比：.methodCall(parameter)。git
1）建立一个新序列，它...github
- 发出一个
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
- 1对1地转化（好比字符串转化为它的长度）：
-
我想添加一些数据元素到一个现有的序列：数组
- 在开头添加：
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
- 转化为 List：
-
我想合并 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)
- ...当有1个或N个其余 publishers 都发出（或结束）时才完成：
- 我想基于一个递归的生成序列的规则扩展每个元素，而后合并为一个序列发出：
- ...广度优先：
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：staticMono.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)
- ...触发判断条件的元素会分到下一个buffer：
- 经过另外一个 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
- 在拿到元素前阻塞：
相关文章
- 1. （20）操做符熔合——响应式Spring的道法术器
- 2. （4）Reactor 3快速上手——响应式Spring的道法术器
- 3. （15）Reactor 3 Operators——响应式Spring的道法术器
- 4. （13）Reactor的backpressure策略——响应式Spring的道法术器
- 5. （16）Reactor的测试——响应式Spring的道法术器
- 6. （17）Reactor的调试——响应式Spring的道法术器
- 7. （19）Reactor Processors——响应式Spring的道法术器
- 8. 响应式Spring的道法术器（Spring WebFlux 教程）
- 9. （14）Reactor调度器与线程模型——响应式Spring的道法术器
- 10. （5）Spring WebFlux快速上手——响应式Spring的道法术器
- 更多相关文章...
- • QBC的2种检索方式 - Hibernate教程
- • Mybatis实现映射器的2种方式 - MyBatis教程
- • 互联网组织的未来：剖析GitHub员工的任性之源
- • Docker容器实战(六) - 容器的隔离与限制
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. （20）操做符熔合——响应式Spring的道法术器
- 2. （4）Reactor 3快速上手——响应式Spring的道法术器
- 3. （15）Reactor 3 Operators——响应式Spring的道法术器
- 4. （13）Reactor的backpressure策略——响应式Spring的道法术器
- 5. （16）Reactor的测试——响应式Spring的道法术器
- 6. （17）Reactor的调试——响应式Spring的道法术器
- 7. （19）Reactor Processors——响应式Spring的道法术器
- 8. 响应式Spring的道法术器（Spring WebFlux 教程）
- 9. （14）Reactor调度器与线程模型——响应式Spring的道法术器
- 10. （5）Spring WebFlux快速上手——响应式Spring的道法术器