RxJava2：能否将两个存在依赖的Single合并为一条链式调用？

你遇到的场景正是flatMap操作符的核心用途之一——它能让你把第一个Single的发射结果无缝传递给第二个Single，从而将两个嵌套的订阅逻辑合并成一条清晰的链式调用，避免“回调嵌套”的问题。

具体实现步骤 #

首先我们需要调整一下update方法，让它返回Single<Unit>而不是直接订阅（这样才能被flatMap链式调用）：

private fun update(id: Long, url: String, email: String): Single<Unit> {
    return save.execute(FileInfo(id, url, email))
        // 注意：subscribeOn 只需要在流的最上游声明一次即可，下游重复设置不会生效
        // .subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread())
}

然后就可以把两个Single合并成一条链式调用了：

getFile.execute(id)
    .subscribeOn(Schedulers.io()) // 统一指定整个流的订阅线程
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap { fileInfo ->
        // 用第一个Single返回的fileInfo作为参数，启动第二个Single
        update(fileInfo.id, fileInfo.url, fileInfo.email)
    }
    .subscribeBy(
        onSuccess = { view?.onUpdateSuccess() },
        onError = { error ->
            // 这里可以统一处理两个阶段的错误，也可以根据错误类型区分处理
            when (error) {
                is GetFileFailure -> view?.showError()
                is SaveFileFailure -> view?.showUpdateError()
                else -> view?.showError()
            }
        }
    )

关键细节说明 #

1. subscribeOn的作用范围：它只影响流的订阅线程，而且只需要在流的最上游调用一次就够了，下游再调用不会改变线程设置，所以可以把update方法里的subscribeOn(Schedulers.io())去掉。
2. 错误统一处理：合并后的流可以在同一个onError回调里处理两个Single可能抛出的错误，你可以根据错误类型来区分显示不同的错误提示。
3. 如果不想修改update方法：也可以直接把save.execute的逻辑写到flatMap里，这样连方法都不用改：

getFile.execute(id)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap { fileInfo ->
        save.execute(FileInfo(fileInfo.id, fileInfo.url, fileInfo.email))
            .observeOn(AndroidSchedulers.mainThread())
    }
    .subscribeBy(
        onSuccess = { view?.onUpdateSuccess() },
        onError = { error ->
            if (error is GetFileException) {
                view?.showError()
            } else {
                view?.showUpdateError()
            }
        }
    )

这样调整后，代码的可读性和维护性都会提升很多，完全符合RxJava响应式编程的设计理念。

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