Reactor之发射器（Flux、Mono）

Reactor之发射器（Flux、Mono）转换操作函数

https://www.jianshu.com/p/5172c48cb877

Reactor 发射器转换操作函数

• concatWithValues

@SafeVarargs
public final Flux<T> concatWithValues(T... values)

将值连接到Flux的末尾。

• concat

public static <T> Flux<T> concat(Iterable<? extends Publisher<? extends T>> sources)

连接两个Flux， 连接由源下游发射的迭代转发元素提供的所有源。
通过顺序订阅第一个源，然后在订阅下一个源之前等待它完成，等等，直到最后一个源完成，从而实现连接。任何错误立即中断序列并被转发到下游。

• concatDelayError

public static <T> Flux<T> concatDelayError(Publisher<? extends Publisher<? extends T>> sources)

将从父Publisher发出的ONNEXT信号连接到所有源，转发由下游源发出的元素。
通过顺序订阅第一个源，然后在订阅下一个源之前等待它完成，等等，直到最后一个源完成，从而实现连接。错误不会中断主序列，但是在其余的源有机会被连接之后被传播。

此操作符在取消时丢弃内部排队的元素以产生背压。

• first

@SafeVarargs
public static <I> Flux<I> first(Publisher<? extends I>... sources)

选择第一个Publisher发出任何信号（onNext/onError/onComplete）并重放来自该Publisher的所有信号，有效地表现得像这些竞争源中最快的一个。

返回一个新的Flux，其性能最快。

• merge

public static <T> Flux<T> merge(Publisher<? extends Publisher<? extends T>> source)

将由Publisher发出的Publisher序列的数据合并为交织合并序列。与concat不同的是，内部source踊跃竞争。

返回一个合并的Flux

• mergeOrdered

@SafeVarargs
public static <I extends Comparable<? super I>> Flux<I> mergeOrdered(Publisher<? extends I>... sources)

通过从每个源（由它们的自然顺序定义）中选择最小值，将来自提供的Publisher序列的数据合并成有序的合并序列。这不是一个SORT（），因为它不考虑整个序列。
相反，该操作符只考虑来自每个源的一个值，并选择所有这些值中最小的值，然后为选择的源补充槽。

返回 一个合并Flux，但保持原始排序的合并Flux

• switchOnNext

public static <T> Flux<T> switchOnNext(Publisher<? extends Publisher<? extends T>> mergedPublishers)

创建一个Flux，该镜像反映最近发布的Publisher，转发其数据直到源代码中的新Publisher进入。
一旦源中没有新的Publisher（源已完成），并且最后一个镜像Publisher也已完成，则生成的Flux将完成。

• zip

public static <T1,T2,O> Flux<O> zip(Publisher<? extends T1> source1,
                                    Publisher<? extends T2> source2,
                                    BiFunction<? super T1,? super T2,? extends O> combinator)

将两个源压缩在一起，也就是说，等待所有源发出一个元素，并将这些元素组合成一个输出值（由提供的组合器构造）。操作将继续这样做，直到任何来源完成。错误将立即被转发。这种“Step-Merge”处理在分散聚集场景中特别有用。

• all

public final Mono<Boolean> all(Predicate<? super T> predicate)

如果这个序列的所有值与谓词匹配，则发出一个布尔布尔值true 。
该实现使用短路逻辑，如果谓词与值不匹配，则用FALSE完成。

• any

public final Mono<Boolean> any(Predicate<? super T> predicate)

如果这个Flux序列的任何值与谓词匹配，则发出一个布尔布尔值true。
该实现使用短路逻辑，如果任何值与谓词不匹配，则完成FALSE。

• as

public final <P> P as(Function<? super Flux<T>,P> transformer)

将此Flux转换为目标类型。

• buffer

public final Flux<List<T>> buffer()

将所有传入的值收集到一个列表缓冲器中，一旦该Flux完成，将由返回的Flux发出。

该操作在数据信号触发的取消或错误时丢弃缓冲器。

• cache

public final Flux<T> cache()

将此Flux量转换为热源，并为进一步的用户缓存最后发射的信号。将保留一个无限量的OnNeXT信号。完成和错误也将被重放。

返回一个缓存的 Flux

• cast

public final <E> Flux<E> cast(Class<E> clazz)

将产生的Flux类型转换为目标产生类型。

• collect

public final <E> Mono<E> collect(Supplier<E> containerSupplier,
                                 BiConsumer<E,? super T> collector)

通过应用收集器BiConsumer获取容器和每个元素，将此Flux发出的所有元素收集到用户定义的容器中。收集的结果将在这个序列完成时发出。

• collectList

public final Mono<List<T>> collectList()

将此Flux所发射的所有元素收集到序列完成时由结果Mono发出的列表。

• defaultIfEmpty

public final Flux<T> defaultIfEmpty(T defaultV)

如果没有任何数据完成此序列，则提供默认的唯一值

• distinct

public final Flux<T> distinct()

对于每一个Subscriber，跟踪已经从这个Flux 跟踪元素和过滤出重复。
值本身被记录到一个用于检测的哈希集中。如果希望使用distinct(Object::hashcode)更轻量级的方法，该方法不保留所有对象，但是更容易由于hashcode冲突而错误地认为两个元素是不同的。

• doOnCancel

public final Flux<T> doOnCancel(Runnable onCancel)

当Flux被取消时触发附加行为（side-effect）。

• doOnComplete

public final Flux<T> doOnComplete(Runnable onComplete)

当Flux完成时触发附加行为（side-effect）。

• doOnNext

public final Flux<T> doOnNext(Consumer<? super T> onNext)

当Flux发射一个项目时附加行为（side-effect）。

• filter

public final Flux<T> filter(Predicate<? super T> p)

根据给定谓词评估每个源值。如果谓词测试成功，则发出该值。如果谓词测试失败，则忽略该值，并在上游生成请求1。

• log

public final Flux<T> log()

观察所有 Reactive Streams信号并使用记录器支持跟踪它们。默认值将使用Level.INFO和 java.util.logging日志记录。如果SLF4J是可用的，它将被用来代替。

• map

public final <V> Flux<V> map(Function<? super T,? extends V> mapper)

通过对每个项目应用同步功能来转换由该Flux发出的项目。

• then

public final <V> Mono<V> then(Mono<V> other)

让这个Flux完成，然后播放信号提供的Mono。
换句话说，忽略这个Flux和转换完成信号的发射和提供Mono完成信号。在得到的Mono中重放错误信号。

作者：Mr_1214
链接：https://www.jianshu.com/p/5172c48cb877
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。