其他分享
首页 > 其他分享> > Reactor之发射器（Flux、Mono）转换操作函数

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

作者：互联网

数据合并函数

由于业务需求有的时候需要将多个数据源进行合并，Reactor提供了concat方法和merge方法：

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

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

img

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

将值连接到Flux的末尾。

img

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

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

img

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

在这里插入图片描述

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

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

img

返回一个合并的Flux

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

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

img

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

从图中可以很清楚的看出这两种合并方法的不同：

下面对concat和merge相关的方法进行测试，先准备测试数据：

private Flux<Integer> flux1() {
    return Flux.range(1,4);
}

private Flux<Integer> flux2() {
    return Flux.range(5,8);
}


private Flux<String> hotFlux1() {
    return flux1().map(i-> "[1]"+i).delayElements(Duration.ofMillis(10));
}

private Flux<String> hotFlux2() {
    return flux2().map(i-> "[2]"+i).delayElements(Duration.ofMillis(4));
}

concat相关方法

concat代码演示

    @Test
    public void concatTest() throws InterruptedException {

        Flux.concat(hotFlux1(), hotFlux2())
                .subscribe(i -> System.out.print("->"+i));

        Thread.sleep(200);
    }

运行结果
从结果可以看出先运行完flux1之后再运行flux2

在这里插入图片描述

concatWith方法

用法和concat基本相同，写法略有不同：

@Test
public void concatWithTest () {
    flux1().concatWith(flux2())
            .log()
            .subscribe();
}

merge相关方法

merge代码演示

@Test
public void mergeTest() throws InterruptedException {

    Flux.merge(hotFlux1(), hotFlux2())
            .subscribe(i -> System.out.print("->"+i));
    
    Thread.sleep(200);
}

运行结果
很明显顺序和concat的区别，是按照时间先后执行

在这里插入图片描述

mergeWith用法

用法和merge相同，写法不同而已

@Test
public void mergeWithTest() throws InterruptedException {

    hotFlux1().mergeWith(hotFlux2())
            .subscribe(i -> System.out.print("->"+i));
    
    Thread.sleep(200);
}

mergeSequential用法

在这里插入图片描述

@Test
public void mergeSequentialTest() throws InterruptedException {
    Flux.mergeSequential(hotFlux1(), hotFlux2())
            .subscribe(i -> System.out.print("->"+i));

    Thread.sleep(200);
}

结果和concat的一样都是

在这里插入图片描述

mergeOrdered用法

合并接收之后再排序

       @Test
    public void mergeOrderedTest() throws InterruptedException {

        Flux.mergeOrdered(Comparator.reverseOrder(), hotFlux1(), hotFlux2())
                .subscribe(i -> System.out.print("->"+i));

        Thread.sleep(200);
    }

在这里插入图片描述

combineLatest用法

在这里插入图片描述

代码示例

  @Test
  public void combineLatestTest() throws InterruptedException {

    Flux.combineLatest(hotFlux1(), hotFlux2(), (v1, v2) -> v1 + ":" + v2)
            .subscribe(i -> System.out.print("->"+i));
    
    Thread.sleep(200);
}

运行结果

在这里插入图片描述

转换操作函数

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

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

img

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

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

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

img

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”处理在分散聚集场景中特别有用。

img

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

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

img

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

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

img

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

将此Flux转换为目标类型。

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

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

img

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

public final Flux<T> cache()

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

img

返回一个缓存的 Flux

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

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

img

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

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

img

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

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

img

public final Flux<T> defaultIfEmpty(T defaultV)

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

img

public final Flux<T> distinct()

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

img

public final Flux<T> doOnCancel(Runnable onCancel)

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

img

public final Flux<T> doOnComplete(Runnable onComplete)

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

img

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

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

img

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

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

img

public final Flux<T> log()

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

img

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

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

img

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

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

img

标签：Publisher,Reactor,Mono,public,Flux,final,concat
来源： https://www.cnblogs.com/satire/p/14971179.html