在RxJava中concatMap和flatMap有什么区别

Question

似乎这两个功能非常相似.它们具有相同的签名(接受rx.functions.Func1<? super T, ? extends Observable<? extends R>> func),并且它们的大理石图看起来完全相同.不能在这里粘贴图片,但这里是concatMap的一个,这里是flatMap的一个.在结果的描述中似乎存在一些细微的差别Observable,其中一个由concatMap包含由结合的Observable产生的flatMap项产生,而由包含由首先合并结果的Observable 产生的项产生,并且发出该合并的结果.

然而,这种微妙之处对我来说完全不清楚.任何人都可以更好地解释这种差异,并且理想地给出一些说明这种差异的例子.

Answer 1

正如您所写,这两个函数非常相似,细微差别在于如何创建输出(在应用映射函数之后).

平面地图使用合并运算符,而concatMap使用concat运算符.

如您所见,concatMap输出序列是有序的 - 第一个Observable发出的所有项在第二个Observable发出的任何项之前发出,
而flatMap输出序列合并 - 合并的Observable发出的项可能出现在任何订单,无论他们来自哪个来源Observable.

Answer 2

即使这里的答案很好，但没有示例也不容易发现差异。因此，我为此创建了一个简单的示例：

@Test
public void flatMapVsConcatMap() throws Exception {
    System.out.println("******** Using flatMap() *********");
    Observable.range(1, 15)
            .flatMap(item -> Observable.just(item).delay(1, TimeUnit.MILLISECONDS))
            .subscribe(x -> System.out.print(x + " "));

    Thread.sleep(100);

    System.out.println("\n******** Using concatMap() *********");
    Observable.range(1, 15)
            .concatMap(item -> Observable.just(item).delay(1, TimeUnit.MILLISECONDS))
            .subscribe(x -> System.out.print(x + " "));

    Thread.sleep(100);
}

********使用flatMap（）*********

1 2 3 4 5 6 7 9 8 11 13 15 10 12 14

********使用concatMap（）*********

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

从输出中可以看出，的结果flatMap是无序的concatMap。

Answer 3

一个非常重要的区别:concatMap等待当前发出的observable完成而flatMap不是.flatMap试图尽可能多地开始.简单地说 - 你无法连接无限的东西.只要确保你发出的可观察量concatMap可以完成,否则整个流程将被卡住等待当前的observable完成以连接下一个.

Answer 4

我发现大多数赞成的答案中的例子不太清楚，所以我发布了一个帮助我理解 flatMap 和 concatMap 之间区别的例子。

FlatMap需要排放源可观察到的，然后创建新观察到的，并合并其原来的链条，而concatMap Concat的它原来的链条。

主要区别在于 concatMap() 将按顺序合并每个映射的 Observable 并一次触发一个。只有当当前 Observable 调用 onComplete() 时，它才会移动到下一个 Observable。

这是flatMap示例：

private void flatMapVsConcatMap() throws InterruptedException {
    Observable.just(5, 2, 4, 1)
            .flatMap(
                    second ->
                            Observable.just("Emit delayed with " + second + " second")
                                    .delay(second, TimeUnit.SECONDS)
            )
            .subscribe(
                    System.out::println,
                    Throwable::printStackTrace
            );

    Thread.sleep(15_000);
}

将导致：

Emit 延迟 1 秒
Emit 延迟 2 秒
Emit 延迟 4 秒
Emit 延迟 5 秒

这是concatMap示例：

private void flatMapVsConcatMap() throws InterruptedException {
    Observable.just(5, 2, 4, 1)
            .concatMap(
                    second ->
                            Observable.just("Emit delayed with " + second + " second")
                                    .delay(second, TimeUnit.SECONDS)
            )
            .subscribe(
                    System.out::println,
                    Throwable::printStackTrace
            );

    Thread.sleep(15_000);
}

将导致：

发射延迟 5 秒
发射延迟 2 秒
发射延迟 4 秒
发射延迟 1 秒

注意使用 Thread.sleep()因为delay()默认在计算调度器上运行