如何在Kotlin中合并并并行化两个异步调用？

这是个很典型的异步数据流合并问题，Kotlin内置的协程（Coroutines）和Flow API正好能帮你优雅解决，不用自己写一堆繁琐的状态管理代码，完美解决你提到的两个痛点：并行获取数据、避免重复订阅。

我分两种常见场景给你讲解：

场景1：一次性获取用户信息（单次回调） #

如果你的subscribeToUserPhone和subscribeToUserStatus是单次获取数据后就结束的异步方法，我们可以先把回调式的API转化为协程的挂起函数，再用async/await实现并行执行：

步骤1：将回调方法转为挂起函数 #

用suspendCoroutine把回调逻辑包装成挂起函数，让它能在协程中使用：

suspend fun getUserPhone(userId: Int): String = suspendCoroutine { continuation ->
    userManager.subscribeToUserPhone(userId) { phone ->
        continuation.resume(phone)
    }
}

suspend fun getUserStatus(userId: Int): Int = suspendCoroutine { continuation ->
    userManager.subscribeToUserStatus(userId) { status ->
        continuation.resume(status)
    }
}

步骤2：并行执行并合并结果 #

用async同时启动两个异步任务，等待两者完成后合并成UserInfo：

fun getUserInfo(userId: Int, callback: (UserInfo) -> Unit) {
    // 注意：这里可以根据实际需求选择CoroutineScope，比如用ViewModelScope（Android）或者自定义Scope
    CoroutineScope(Dispatchers.IO).launch {
        // 并行启动两个异步任务
        val phoneDeferred = async { getUserPhone(userId) }
        val statusDeferred = async { getUserStatus(userId) }
        
        // 等待两个任务都完成
        val phone = phoneDeferred.await()
        val status = statusDeferred.await()
        
        // 回调合并后的结果
        callback(UserInfo(phone, status))
    }
}

这样两个请求是并行发起的，不会串行等待，大大减少了总耗时。

场景2：持续监听用户信息更新（多次回调） #

如果你的subscribeToUserPhone和subscribeToUserStatus是持续推送更新的（比如用户手机号或状态变化时会多次回调），那原代码的问题会更明显：每次手机号更新都重新订阅状态，造成重复订阅和资源浪费。这时候用Kotlin Flow的combine操作符是最佳方案：

步骤1：将回调方法转为Flow #

用callbackFlow把持续回调的API转化为Flow流：

import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.trySend

fun userPhoneFlow(userId: Int): Flow<String> = callbackFlow {
    // 发起订阅
    userManager.subscribeToUserPhone(userId) { phone ->
        // 把新的手机号发送到Flow中
        trySend(phone)
    }
    
    // 当Flow被取消时，执行取消订阅的逻辑（如果有对应的取消方法）
    awaitClose {
        // userManager.unsubscribeFromUserPhone(userId)
    }
}

fun userStatusFlow(userId: Int): Flow<Int> = callbackFlow {
    userManager.subscribeToUserStatus(userId) { status ->
        trySend(status)
    }
    
    awaitClose {
        // userManager.unsubscribeFromUserStatus(userId)
    }
}

步骤2：合并两个Flow流 #

用combine操作符合并两个流，每当其中一个流有新数据时，就用两者的最新值合并成UserInfo：

import kotlinx.coroutines.flow.combine

fun subscribe(userId: Int, callback: (UserInfo) -> Unit) {
    CoroutineScope(Dispatchers.Main).launch {
        combine(
            userPhoneFlow(userId),
            userStatusFlow(userId)
        ) { phone, status ->
            // 用两个流的最新数据合并成UserInfo
            UserInfo(phone, status)
        }.collect { userInfo ->
            // 回调合并后的结果
            callback(userInfo)
        }
    }
}

这个方案的优势：

• 两个订阅是并行发起的，同时监听更新，不会串行等待
• 只会订阅一次手机号和一次状态，不会因为手机号更新而重复订阅状态
• 每当任一数据源更新时，都会自动用最新的手机号和最新的状态合并成UserInfo，保证数据的一致性

总结一下：Kotlin的协程和Flow是官方提供的异步处理方案，完全能覆盖你的需求，不用自己手动管理并行和订阅状态，代码简洁又可靠。

备注：内容来源于stack exchange，提问作者Addev