Reactor compose vs flatMap

Question

我继续使用 Reactor，现在我看到 compose 运算符的行为与 flatMap 完全一样，我想知道是否有什么我不明白的区别。

    @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

@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 适用于整个流，但flatMap 适用于流中的单个项目

Answer 2

An excellent explanation by Dan Lew:

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

• compose() 是从流中获取原始Observable<T> 的唯一方法。因此，影响整个流的运算符（如subscribeOn() 和observeOn()）需要使用compose()。

  相反，如果您将subscribeOn()/observeOn() 放入flatMap()，它只会影响您在flatMap() 中创建的Observable，而不影响流的其余部分。

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

• flatMap() 的效率必然较低，因为它必须在每次调用 onNext() 时创建一个新的 Observable。 compose() 按原样在流上运行。如果您想用可重用代码替换某些运算符，请使用compose()。 flatMap() 有很多用途，但这不是其中之一。