Reactor组成与平面图

Question

我继续玩Retor，现在我看到compose运算符的行为完全类似于平面图，我想知道是否有任何我不明白的区别。

    @Test
public void compose() throws InterruptedException {
    Scheduler mainThread = Schedulers.single();
    Flux.just(("old element"))
            .compose(element ->
                    Flux.just("new element in new thread")
                            .subscribeOn(mainThread)
                            .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName())))
            .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName()))
            .subscribe(System.out::println);
    Thread.sleep(1000);
}

@Test
public void flatMapVsCompose() throws InterruptedException {
    Scheduler mainThread = Schedulers.single();
    Flux.just(("old element"))
            .flatMap(element ->
                    Flux.just("new element in new thread")
                            .subscribeOn(mainThread)
                            .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName())))
            .doOnNext(value -> System.out.println("Thread:" + Thread.currentThread().getName()))
            .subscribe(System.out::println);
    Thread.sleep(1000);
}

这两个示例的行为和返回的结果相同。

问候。

Answer 1

Dan Lew的精彩解释：

不同之处在于，compose（）是一个更高级别的抽象：它对整个流进行操作，而不是单独发出的项。更具体地说：

>

相反，如果在flatMap（）中放入subscribeOn（）/observeOn（），它只会影响在flatMap（）中创建的可观察的，而不会影响流的其余部分。

compose（）在创建Observable流时立即执行，就好像您已经内联编写了运算符一样flatMap（）在每次调用其onNext（）时执行。换句话说，flatMap（）转换每个项目，而compose（）转换整个流。

flatMap（）的效率必然较低，因为每次调用onNext（）时，它都必须创建一个新的Observablecompose（）按原样对流进行操作。如果要用可重用代码替换某些运算符，请使用compose（）flatMap（）有很多用途，但这不是其中之一。

Answer 2

来自@Andrew的解释很好。只是想增加一个更好理解的例子。

Flux.just("1", "2")
        .compose( stringFlux -> {
            System.out.println("In compose"); // It takes whe whole Flux as input
           return stringFlux.collectList();
        }).subscribe(System.out::println);


Flux.just("1", "2").flatMap(s -> { //Input to the anonymous function is individual items in stream
            System.out.println("In flatMap");
            return Flux.just(Integer.parseInt(s));
        }).subscribe(System.out::println);

这就产生了输出

In compose
[1, 2]
In flatMap
1
In flatMap
2

这表明compose适用于整个流，但平面图适用于流中的单个项目