1

Задача такая - обработать большой список объектов параллельно, но чтобы одновременно работали не более определенного количества обработчиков.

Например есть такой вариант:

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

Может вообще тут другой подход нужен?

1 ответ 1

0

Сделал так

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
}

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.