Kotlin中大型列表的最优过滤方案及复杂场景优化探讨

先说说你现有代码的问题 #

你给500万条数据每条都创建一个async协程，这会导致协程数量爆炸——虽然协程轻量，但500万个协程的调度开销会完全抵消并行带来的收益，甚至比同步处理更慢。而且这种写法并没有真正高效利用CPU，反而会让调度器陷入频繁切换的压力中。

普通场景（轻量过滤逻辑）：直接同步过滤最优 #

如果你的过滤逻辑像示例里的number % 2 == 0这么简单，同步过滤就是最快的方案，因为协程调度的开销远大于简单判断的成本。代码直接简化成：

val filteredList = hugeCollection.filter { it % 2 == 0 }

没有多余的协程创建、await等待，直接遍历过滤，性能拉满。

计算密集型过滤逻辑：控制并行度是关键 #

如果过滤逻辑是CPU密集型（比如复杂计算、加密、大量数据处理），需要并行处理，但必须限制协程数量，避免过度调度。以下是两种最优方案：

方案1：分块并行处理 #

把大列表拆分成若干块，每块用一个协程处理，最后合并结果。这样协程数量可控，能充分利用CPU核心：

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.coroutineScope

suspend fun main() = coroutineScope {
    val hugeCollection = buildList { repeat(5_000_000) { add(it) } }
    println("Size of hugeCollection: ${hugeCollection.size}")

    // 根据CPU核心数调整块大小，比如核心数*1000，这里示例用10万
    val chunkSize = 100_000
    val filteredList = hugeCollection
        .chunked(chunkSize)
        .map { chunk ->
            // 用Dispatchers.Default处理CPU密集型任务
            async(Dispatchers.Default) {
                chunk.filter { isHeavyFilter(it) }
            }
        }
        .flatMap { it.await() } // 合并所有块的结果

    println("Size of filteredList: ${filteredList.size}")
}

// 模拟计算密集型过滤逻辑
fun isHeavyFilter(number: Int): Boolean {
    // 比如耗时的计算
    repeat(1000) { number * it }
    return number % 2 == 0
}

方案2：用Flow优雅实现并行过滤 #

Kotlin Flow的flatMapMerge可以控制并行度，更简洁地实现并行过滤：

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.flatMapMerge
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.toList

suspend fun main() = coroutineScope {
    val hugeCollection = buildList { repeat(5_000_000) { add(it) } }
    println("Size of hugeCollection: ${hugeCollection.size}")

    // concurrency设置为CPU核心数左右（比如8）
    val filteredList = hugeCollection.asFlow()
        .flatMapMerge(concurrency = 8) { number ->
            flow {
                if (isHeavyFilter(number)) {
                    emit(number)
                }
            }.flowOn(Dispatchers.Default)
        }
        .toList()

    println("Size of filteredList: ${filteredList.size}")
}

fun isHeavyFilter(number: Int): Boolean {
    repeat(1000) { number * it }
    return number % 2 == 0
}

总结 #

• 轻量过滤：直接同步filter，没有比这更快的方式。
• 计算密集型过滤：控制并行协程数量（匹配CPU核心数），用Dispatchers.Default调度CPU密集任务，分块或Flow都是可靠的方案。

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