主页 / 问题 / 730349
Accepted
rjhdby
Asked:2020-10-13 19:40:22 +0000 UTC2020-10-13 19:40:22 +0000 UTC

Kotlin，如何限制并发执行协程的数量

  • 772

任务是并行处理大量对象，但不能超过一定数量的处理程序同时工作。

例如，有这个选项：

fun <A, B>List<A>.pmap(f: suspend (A) -> B): List<B> = runBlocking {
    map { async(CommonPool) { f(it) } }.map { it.await() }
}

但是会同时处理整个列表。

怎么会做出这样的类比delayWhen呢 RxJava？

var i:Int = 0; //atomic
onEach { i++ } }
    .delayWhen{ i > 9 }
    .map{ async(CommonPool){ e -> { doSomeStuff(e) } } }
    .map { it.await() }
    .onEach { i-- }

也许这里需要一种不同的方法？

kotlin

1 个回答

  1. Best Answer

    这样做了

    suspend fun <T, R> List<T>.runParallel(maxThreads:Int, worker: (T) -> R): List<R> {
    val out = ArrayList<R>()
    val processThreads = Math.min(maxThreads, size)
    val processingContext = newFixedThreadPoolContext(processThreads, "process-context")
    val waitingTasksChannel = Channel<T>()
    val finishTasksChannel = Channel<R>()
    val schedulingTask = launch {
        this@runParallel.forEach {
            waitingTasksChannel.send(it)
        }
    }
    val processingTasks = (1..processThreads).map {
        launch(processingContext) {
            for (waitTask in waitingTasksChannel) {
                val processResult = worker(waitTask)
                finishTasksChannel.send(processResult)
            }
        }
    }
    val finishTask = launch {
        for (finishTask in finishTasksChannel) {
            out.add(finishTask)
        }
    }

    schedulingTask.join()
    waitingTasksChannel.close()
    processingTasks.forEach { it.join() }
    finishTasksChannel.close()
    finishTask.join()
    return out
}
    • 0

相关问题