我有一個延遲生成的大序列，并且太大而無法放入記憶體。
我想使用協程來處理這個序列以提高性能，在這個例子中我使用 10 個并行執行緒進行處理。
runBlocking(Dispatchers.IO.limitedParallelism(10)) {
massiveLazySequenceOfThings.forEach { myThingToProcess ->
print("I am processing $myThingToProcess")
launch() {
process(myThingToProcess)
}
}
}
這里的問題是第一個列印陳述句將針對序列中的每個專案執行，因此對于像我這樣的非常大的序列，這將導致 OOM。
在這個例子中，有沒有辦法讓我的序列的迭代“懶惰”，以便在任何時候只處理一個固定的數字？
我是否被迫在這里使用通道（可能使用緩沖通道？）在序列迭代期間強制阻塞呼叫，直到處理某些專案？還是我缺少其他一些更清潔的解決方案。
在我的實際示例中，我還使用 supervisorScope 來監視每個處理作業，所以如果可能的話，我也想保留它。
uj5u.com熱心網友回復：
您可以使用 限制并行操作的數量Semaphore。
runBlocking(Dispatchers.IO) {
val semaphore = Semaphore(permits = 10)
massiveLazySequenceOfThings.forEach { myThingToProcess ->
semaphore.acquire()
print("I am processing $myThingToProcess")
launch {
try {
process(myThingToProcess)
} finally {
semaphore.release()
}
}
print("I am done processing $myThingToProcess")
}
}
uj5u.com熱心網友回復：
問題可能是您安排新任務的速度比并行完成任務的速度要快。
您可以通過將資料分塊為可以并行處理的大小（在您的示例中為 10）來解決此問題，并等待每個塊完成，然后再開始下一個塊：
runBlocking(Dispatchers.IO) {
massiveLazySequenceOfThings.chunked(10).forEach { chunk ->
println("I am processing $chunk")
chunk.map { async { process(it) } }.awaitAll()
}
}
現在一個塊中的 10 個元素中的每一個都是并行處理的，但是每個塊作為一個整體是順序處理的，確保不會因為計劃任務過多而耗盡記憶體。
一點旁注："I am done processing $myThingToProcess"您示例中的輸出不正確。該作業可能仍在其自己的執行緒中運行。您唯一可以肯定地說的是您安排了它。
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/450257.html