RxJava中嵌套的Single.FlatMap与Single.zip，是否相同？

Question

val s1 : Single<String> = service1.execute().subscribeOn(io())
val s2 : Single<Int> = service2.execute().subscribeOn(io())
val s3 : Single<Int> = service3.execute().subscribeOn(io())
val s4 : Single<String> = service4.execute().subscribeOn(io())
val ....
val s10 : Single<Int> = service10.execute().subscribeOn(io())

数据类MyObj(Field1:String、Field2:Int、Field3:Int、Field4:String......Field10:Int)

我有一个service10.execute(s1:String s2:Int s3:Int s4:String)

如果我这样做：

s1.flatMap { str -> 
    s2.flatMap { int1 ->
        s3.flatMap { int2 ->
            s4.flatMap { str2 ->
                ...
                s10.flatmap { int10
                  service10.execute(myObj(str, int1, int2, str2..., int10))
                }
            }
        }
    }
}

Single.zip(
            listOf(
                s1,
                s2,
                s3,
                s4
              ...,
              ...,
              s10
            )
        ) { array ->
            val str = array[0] as String
            val int1 = array[1] as Int
            val int2 = array[2] as Int
            val str2 = array[3] as String
            ...
            val str10 = array[9] as Int
        }

1）flatMap在那里是并行执行还是顺序执行？2）如果嵌套的flatMap是顺序的，有没有办法让它们像zip一样并行？

Answer 1

不，嵌套的FlatMap不能使并行运行，下面的测试证明了这一点：

    // so we can be sure service1 and service2 are active
    val bothSubscribed = CountDownLatch(2)
    // so we can simulate a blocking, long running operation on both services
    val subscribeThreadsStillRunning = CountDownLatch(1)

    val service5 = { str: String, str2: String ->
        Observable.just("service5: $str, $str2").singleOrError()
    }

    val scheduler = Schedulers.io()

    val createSingle = { value: String ->
        Observable
            .create<String> { emitter ->
                println("subscribe $value on ${Thread.currentThread().name}")
                bothSubscribed.countDown()
                subscribeThreadsStillRunning.await(10, SECONDS)
                emitter.onNext(value)
            }
            .singleOrError()
            .subscribeOn(scheduler)
    }

    val s1 = createSingle("outer")
    val s4 = createSingle("inner")

    s1.flatMap { outer ->
        s4.flatMap { inner ->
            service5(outer, inner)
        }
    }.subscribe()

    assert(bothSubscribed.await(5, SECONDS))
    subscribeThreadsStillRunning.countDown()

原因可以通过记住lambda中的代码在lambda执行之前是不运行的来理解的（似乎是这样说的，但我花了一点时间才得到它）。S4.FlatMap触发订阅S4，但是在outer可用之前，即在S1已经发出并因此完成之前，此代码不会执行。

Zip似乎是这方面的完美解决方案，我不知道为什么要使用平面映射。我想不出做这件事的办法。它还有一个类型安全的API，因此您不必在示例中使用基于数组的API。

Singles
        .zip(s1, s4) { outer, inner -> service5(outer, inner) }
        .flatMap { it }
        .subscribe()