Rx中如何合并Observable与Single实现原始及处理后数据依次发射

你当前使用flatMapSingle会直接将上游发射的原始数据替换为Single处理后的结果，原始数据不会进入下游，因此无法输出预期内容。只需要在转换阶段先发射原始数据，再拼接Single处理后的结果即可。

核心修改逻辑 #

• 替换flatMapSingle为concatMap（异步场景下可严格保证顺序，同步场景用flatMap也可实现效果）
• 对每个上游发射的列表项，按顺序拼接两个发射逻辑：先发射原始列表，再发射Single转换后的结果，使用concat保证单组内的发射顺序和预期一致

完整可运行代码 #

fun main() {
    val namesObservable = Observable.just(listOf("A", "B", "C"), listOf("D", "E", "F"))
    namesObservable
        .concatMap { names ->
            // 严格按顺序发射：原始数据 -> Single处理后的数据
            Observable.concat(
                Observable.just(names),
                addCharSingle(names).toObservable()
            )
        }
        .subscribe { item ->
            println(item.joinToString())
        }
}

fun addCharSingle(letters: List<String>): Single<List<String>> {
    return Single.fromCallable { letters.map { letter -> "$letter!" } }
}

运行输出 #

A, B, C
A!, B!, C!
D, E, F
D!, E!, F!

说明：

• 选择concatMap而非flatMap，是因为concatMap会严格按照上游发射的顺序处理每一组数据，即使addCharSingle是异步耗时操作，也不会出现第二组数据插到第一组前面的乱序问题，完全匹配要求的输出顺序。
• 不要使用doOnNext打印原始数据的取巧方案，doOnNext是副作用钩子，不会将原始数据作为正式流事件发射到下游，不符合构建完整Rx链式调用的需求。

内容的提问来源于stack exchange，提问作者francisco_ssb