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

2023-10-31 10:23:10

数据合并函数

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

concat

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

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

concatWithValues

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

将值连接到Flux的末尾。

concatDelayError

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

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

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

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

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

concat是合并的flux，按照顺序分别运行，flux1运行完成以后再运行flux2
merge是同时运行，根据时间先后运行

下面对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用法

跟concat有些相似，得到的结果类似
跟concat不同在于，订阅的源是hot型，接收数据后根据订阅顺序重新排序

@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用法

跟concat和merge不同该方法是将多个源的最后得到元素通过函数进行融合的到新的值

代码示例

  @Test
  public void combineLatestTest() throws InterruptedException {

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

运行结果

结果都是flux1和flux2中元素进行融合之后的元素

转换操作函数

first

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

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

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

switchOnNext

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

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

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

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

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

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

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

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。

public final Flux<T> log()

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

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中重放错误信号。

码农公寓