Kotlin中如何发起多个Rx异步调用并等待全部完成

• 跨线程读写全局布尔标记没有同步保障，存在内存可见性问题，极端情况下所有标记被置true的状态可能被轮询流读取不到
• 持续运行的轮询Flowable如果没有及时解绑，会长期占用资源，造成不必要的性能消耗
• 缺失错误处理链路：任意一个异步任务执行失败、抛异常时，对应的布尔标记永远不会被置为true，后续业务逻辑永远不会触发，也没有失败回调通道
• 全局可变的布尔标记容易因为意外的状态重置造成逻辑异常，代码耦合度高，可维护性差

核心思路：把每一个异步任务封装成独立的Rx可订阅源，完全不需要手动维护状态标记，直接用Rx内置的组合操作符等待所有流的完成信号，自动触发后续逻辑

第一步：将原生异步函数改造为Rx可订阅源 #

不要用裸回调+全局布尔标记的写法，优先把每个异步任务封装为Completable——这种源最适配“只关心任务执行成功/失败，不需要取返回值”的场景，如果需要拿任务执行结果就换成Single类型。改造示例：

// 改造后的async1，不再依赖全局布尔标记
fun async1(): Completable {
    return Completable.fromAction {
        // 放入原async1内的耗时异步逻辑
        // 逻辑执行完成后会自动向下游传递完成信号，不需要手动设置标记
    }.subscribeOn(Schedulers.io()) // 指定任务在IO线程执行
}

async2、async3按照相同逻辑改造即可。

兼容历史代码的桥接方案：如果部分老异步接口只有原始回调、无法直接改成Rx源，可以通过Completable.create()把原有回调的完成/失败信号桥接到Rx流，不需要修改原有异步任务的内部实现：

fun async1Legacy(): Completable {
    return Completable.create { emitter ->
        // 调用原有老异步接口
        oldAsync1(
            onComplete = { emitter.onComplete() },
            onError = { err -> emitter.onError(err) }
        )
    }
}

第二步：通过内置组合操作符等待所有任务完成 #

完全不需要写持续轮询的Flowable，直接用Completable.mergeArray()把三个任务的完成信号合并，等所有上游任务都正常发射完成信号时，自动触发后续业务逻辑：

fun useResultOfAsyncFuns() {
    Completable.mergeArray(
        async1(),
        async2(),
        async3()
    )
    .observeOn(AndroidSchedulers.mainThread()) // 移动端场景可切回主线程执行业务，非移动端可按需指定线程
    .subscribe(
        {
            // 三个任务全部执行完成后，此处逻辑会自动触发
            // 直接写入目标业务代码即可
        },
        { error ->
            // 任意一个任务执行失败时，此处会接收到异常，不会出现永久卡等的情况
            error.printStackTrace()
        }
    )
}

场景适配说明：如果你的异步任务需要向上游返回执行结果，将所有Completable替换为对应类型的Single，使用Single.zip()操作符即可实现完全相同的等待逻辑。

该方案的优势 #

• 完全移除了全局可变的布尔标记，不存在跨线程状态不同步的问题
• 流的生命周期自动收敛：三个任务全部完成/任意一个失败后，整个合并流就会自动结束，不需要手动处理轮询流的解绑，不会出现资源泄漏
• 内置完整的错误传导机制：任意异步任务的执行异常都会直接传递到下游的错误处理回调，不会出现逻辑永久卡死的情况
• 线程调度灵活可控：可以灵活指定每个异步任务的执行线程、后续业务逻辑的执行线程，不需要手动处理线程切换的样板代码

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