Котлин сопрограммы занимает больше времени, чем темы

Question

Я новичок в Kotlin и сопрограммах и пытаюсь понять API сопрограмм, поэтому вполне возможно, что я делаю что-то не так. Итак, у меня есть список объектов определенного типа, и я пытаюсь применить длительную обработку к каждому из этих объектов.

val listOfFoos = listOf(Foo(1), ..., Foo(n))
listOfFoos.forEach { longRunningJob(it) }

fun longRunningJob(foo: Foo) {
    runBlocking{
        delay(2000) //hardcoded delay for testing
    }
    //do something else
}

Конечно, это идеальный кандидат для одновременного запуска, поэтому здесь используются старые добрые потоки:

listOfFoos.map { thread(start = true) { longRunningJob(it) } }.forEach { it.join() }

Когда я измеряю время его выполнения с помощью measureTimeMillis, это дает мне около 2 секунд, что кажется совершенно нормальным, поскольку каждый longRunningJob работает параллельно. Но сопрограммы намного лучше, поскольку у них нет таких издержек, как потоки для переключения контекста. Итак, моя реализация с использованием сопрограмм:

val deferredResults =
    listOfFoos.map { GlobalScope.async { longRunningJob(it) } }
runBlocking {
    deferredResults.awaitAll()
}

Но эта реализация завершает выполнение примерно за 4 секунды, а это совсем не то, чего я ожидал, и если я добавлю больше элементов в список, время выполнения также увеличится.

Так что я здесь не так делаю?

Answer 1

Необходимое время для выполнения этого кода зависит от количества потоков, которые используются для вычисления.Ваш пример с потоками не определяет границы и порождает столько потоков, сколько у вас есть заданий.Пример сопрограммы, с другой стороны, отправляет все задачи в GlobalScope, который внутренне использует пул потоков Dispatchers.Default.Этот пул ограничен:

CoroutineDispatcher по умолчанию, который используется всеми стандартными сборщиками, такими как launch, async и т. Д., Если в их контексте не указан ни диспетчер, ни какой-либо другой ContinuationInterceptor.

Он поддерживается общим пулом потоков в JVM.По умолчанию максимальное количество потоков, используемых этим диспетчером, равно количеству ядер ЦП, но не менее двух.

Допустим, у вас 4 ядра.Выполнение кода с 4 заданиями приведет к времени выполнения ~ 2 с, потому что все работает параллельно (обратите внимание на параллелизм <> параллелизм).Но как только у вас будет более 4 задач, нужно будет подождать, пока одна из первых задач не завершится, так как в любой момент одновременно могут выполняться только 4 задачи.

Вы можете изменить пул диспетчера на один сбольше потоков:

GlobalScope.async(Dispatchers.IO)

Обратите внимание, что delay является плохим примером длительной задачи.Он не блокирует поток вызывающих, поскольку это реальная функция приостановки, которая только приостанавливает сопрограмму.Вы можете фактически запустить свой код на main полностью:

runBlocking {
    val deferredResults =
        (0..10).map { async(Dispatchers.IO) { longRunningJob() } }
    deferredResults.awaitAll()
}

Answer 2

runBlocking - это функция сопрограммы. Что он делает, так это «запускает код в основном потоке / вызывающем потоке». Так что это не создает параллельные потоки для запуска вещей.

Для асинхронного запуска вашего кода вы должны использовать функцию запуска вместо runBlocking. Он работает на Dispatchers.Default общий пул потоков.

GlobalScope.launch {
   delay(2000);
}