multithreading - 如何使用协程懒惰地迭代一个大(大于内存)的 kotlin 序列

Question

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

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

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

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

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

我是否被迫在这里使用 channel (可能使用缓冲 channel ？)在序列迭代期间强制阻塞调用，直到处理某些项目？还是我缺少其他一些更清洁的解决方案。

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

Answer 1

您可以使用 Semaphore 限制并行操作的数量.

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()
            }
        }
    }
}