如何使用协程延迟迭代大型（大于内存）Kotlin 序列

Question

我有一个延迟生成的大序列，并且太大而无法放入内存中。

我想使用协程来处理这个序列以提高性能，在本例中我使用 10 个并行线程进行处理。

            runBlocking(Dispatchers.IO.limitedParallelism(10)) {
                massiveLazySequenceOfThings.forEach { myThingToProcess ->
                    print("I am processing $myThingToProcess")
                    launch() {
                        process(myThingToProcess)
                    }
                }
            }

这里的问题是，第一个打印语句将对序列中的每个项目执行，因此对于像我这样的非常大的序列，这将导致 OOM。

在这个例子中，有没有办法让我的序列的迭代变得“惰性”，以便在任何时候只处理固定的数字？

我是否被迫在此处使用通道（可能使用缓冲通道？）在序列迭代期间强制执行阻塞调用，直到处理某些项目为止？或者我还缺少其他一些更清洁的解决方案吗？

在我的实际示例中，我还使用supervisorScope来监视每个处理作业，因此如果可能的话我也想保留它。

Answer 1

您可以使用 信号量。

runBlocking(Dispatchers.IO) {
    val semaphore = Semaphore(permits = 10)
    massiveLazySequenceOfThings.forEach { myThingToProcess ->
        semaphore.acquire()
        print("I am processing $myThingToProcess")
        launch {
            try {
                process(myThingToProcess)
            } finally {
                print("I am done processing $myThingToProcess")
                semaphore.release()
            }
        }
    }
}

Answer 2

问题可能是您安排新任务的速度比并行完成任务的速度要快。

您可以通过将数据分块到可以并行处理的大小（在您的示例中为 10）并等待每个块完成然后再开始下一个块来克服此问题：

runBlocking(Dispatchers.IO) {
    massiveLazySequenceOfThings.chunked(10).forEach { chunk ->
        println("I am processing $chunk")
        chunk.map { async { process(it) } }.awaitAll()
    }
}

现在一个块中的 10 个元素中的每一个并行处理，但每个块作为一个整体是按顺序处理的，确保您不会因计划任务太多而耗尽内存。

一个小旁注：您的示例中的输出 "I am Doneprocessing $myThingToProcess" 不正确。该作业可能仍在其自己的线程中运行。你唯一可以肯定的是你已经安排好了。