如何为冷流中的每个元素启动新协程优化处理流程？

你的代码当前是串行处理每个用户ID的网络请求：上游每隔200ms发射一个ID，但map操作会阻塞当前协程，必须等前一个ID的网络请求（2秒）完成后，才会处理下一个ID，总耗时约6.6秒，完全没有利用并行能力。

要实现为每个流元素启动独立协程并行处理，最贴合Flow设计的方案是使用flatMapMerge操作符，它可以为每个上游元素创建独立协程执行异步任务，并并行收集结果。

修改后的代码 #

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds

fun getAllUserIds(): Flow<Int> {
    return flow {
        repeat(3) {
            delay(200.milliseconds)
            println("Emitting ID: $it")
            emit(it)
        }
    }
}

suspend fun getProfileFromNetwork(id: Int): String {
    delay(2.seconds)
    return "Profile[$id]"
}

fun main() = runBlocking {
    getAllUserIds()
        // flatMapMerge会为每个ID启动独立协程，默认最多并行16个任务
        .flatMapMerge { id ->
            flow {
                emit(getProfileFromNetwork(id))
            }
        }
        .collect { profile ->
            println("Got profile: $profile")
        }
}

代码说明 #

1. 并行执行逻辑：flatMapMerge会为上游发射的每个ID，启动一个新协程执行getProfileFromNetwork请求，无需等待前一个请求完成。
2. 耗时优化：总耗时会缩短至约2.2秒（最后一个ID的发射延迟200ms + 网络请求的2秒），对比原串行方案效率提升3倍。
3. 并发控制：如果需要限制最大并发数（避免触发后端限流），可以给flatMapMerge传入参数，比如flatMapMerge(concurrency = 2)，限制同时最多处理2个网络请求。

备选方案（非流式实时处理） #

如果不需要实时收集每个请求的结果，而是等所有请求完成后统一处理，也可以用async+awaitAll的方式：

fun main() = runBlocking {
    val profileDeferreds = getAllUserIds()
        .map { id -> async { getProfileFromNetwork(id) } }
        .toList()
    
    val profiles = profileDeferreds.awaitAll()
    profiles.forEach { println("Got profile: $it") }
}

但这种方式会先收集所有异步任务，再统一等待结果，不如flatMapMerge的流式处理灵活。

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