Компоновка приостанавливаемых функций

Последовательность по умолчанию

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

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

Что делать, если нам нужно, чтобы они вызывались последовательно — сначала doSomethingUsefulOne и затем doSomethingUsefulTwo, и вычислить сумму их результатов? На практике это делается, если результат первой функции используется для принятия решения о необходимости вызова второй функции или для принятия решения о том, как ее вызвать.

Мы используем обычное последовательное вызывание, потому что код в сопроцессе, как и в обычном коде, по умолчанию последовательный. Следующий пример демонстрирует это, измеряя общее время выполнения обеих приостанавливаемых функций:

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        val one = doSomethingUsefulOne()
        val two = doSomethingUsefulTwo()
        println("The answer is ${one + two}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

Он выдаёт примерно следующее:

The answer is 42
Completed in 2017 ms

Конкурентное использование async

Что если между вызовами doSomethingUsefulOne и doSomethingUsefulTwo нет зависимостей, и мы хотим получить ответ быстрее, выполняя их параллельно? Здесь нам поможет async.

async концептуально похожа на launch. Она запускает отдельную сопрограмму — лёгкий поток, который работает параллельно со всеми другими сопрограммами. Разница в том, что launch возвращает Job и не несёт никакого результата, в то время как async возвращает Deferred — лёгкий неблокируемый фьючерс, который представляет собой обещание предоставить результат позже. Вы можете использовать .await() для получения конечного результата от значения deferred, но Deferred также является Job, поэтому вы можете отменить его при необходимости.

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        val one = async { doSomethingUsefulOne() }
        val two = async { doSomethingUsefulTwo() }
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

Он выдаёт примерно следующее:

The answer is 42
Completed in 1017 ms

Это в два раза быстрее, потому что две сопрограммы выполняются параллельно. Обратите внимание, что параллельность с сопрограммами всегда явная.

Ленивое запуск async

Опционально, async может быть сделана ленивой, установив параметр start в CoroutineStart.LAZY. В этом режиме она запускается только тогда, когда её результат требуется функцией await, или если вызывается функция запуска Job. Запустите следующий пример:

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
        val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
        // some computation
        one.start() // start the first one
        two.start() // start the second one
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

Он выдаёт примерно следующее:

The answer is 42
Completed in 1017 ms

Таким образом, здесь две сопрограммы определены, но не выполнены, как в предыдущем примере, но контроль над временем запуска предоставляется программисту путём вызова start. Сначала запускаем one, затем two, а затем ожидаем завершения каждой сопрограммы.

Обратите внимание, что если мы просто вызовем await в println без предварительного вызова start для каждой сопрограммы, это приведёт к последовательному поведению, так как await запускает выполнение сопрограммы и ожидает её завершения, что не является предполагаемым случаем использования ленивости. Сценарий использования async(start = CoroutineStart.LAZY) — это замена стандартной функции lazy, в случаях, когда вычисление значения включает приостанавливаемые функции.

Функции в стиле async

Мы можем определить функции в стиле async, которые вызывают doSomethingUsefulOne и doSomethingUsefulTwo асинхронно, используя конструктор сопрограммы async с использованием ссылки на GlobalScope для отказа от структурированной параллельности. Такие функции мы называем с суффиксом "...Async", чтобы подчеркнуть тот факт, что они только запускают асинхронные вычисления, и для получения результата необходимо использовать возвращаемое значение deferred.

// The result type of somethingUsefulOneAsync is Deferred<Int>
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

// The result type of somethingUsefulTwoAsync is Deferred<Int>
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

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

Следующий пример показывает их использование вне сопрограммы:

import kotlinx.coroutines.*
import kotlin.system.*

//sampleStart
// note that we don't have `runBlocking` to the right of `main` in this example
fun main() {
    val time = measureTimeMillis {
        // we can initiate async actions outside of a coroutine
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // but waiting for a result must involve either suspending or blocking.
        // here we use `runBlocking { ... }` to block the main thread while waiting for the result
        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }
    println("Completed in $time ms")
}
//sampleEnd

@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

Представьте, что между строкой val one = somethingUsefulOneAsync() и выражением one.await() есть ошибка в коде, и программа выбрасывает исключение, и операция, выполняемая программой, прерывается. Обычно глобальный обработчик ошибок мог бы перехватить это исключение, записать и сообщить об ошибке разработчикам, но программа могла бы продолжить выполнение других операций. Однако здесь somethingUsefulOneAsync всё ещё выполняется в фоновом режиме, даже если операция, которая его инициировала, была прервана. Эта проблема не возникает при структурированной параллельности, как показано в разделе ниже.

Структурная конкурентность с async

Рассмотрим пример Concurrent using async и выделим функцию, которая выполняет doSomethingUsefulOne и doSomethingUsefulTwo одновременно и возвращает сумму их результатов. Поскольку билдер корутины async определён как расширение для CoroutineScope, нам необходимо иметь его в области видимости, для чего используется функция coroutineScope:

suspend fun concurrentSum(): Int = coroutineScope {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    one.await() + two.await()
}

Таким образом, если в коде функции concurrentSum произойдёт ошибка и будет брошено исключение, все запущенные в её области корутины будут отменены.

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        println("The answer is ${concurrentSum()}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun concurrentSum(): Int = coroutineScope {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    one.await() + two.await()
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

При этом одновременное выполнение обеих операций сохраняется, как видно из вывода функции main:

The answer is 42
Completed in 1017 ms

Отмена всегда распространяется по иерархии корутин:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    try {
        failedConcurrentSum()
    } catch(e: ArithmeticException) {
        println("Computation failed with ArithmeticException")
    }
}

suspend fun failedConcurrentSum(): Int = coroutineScope {
    val one = async<Int> { 
        try {
            delay(Long.MAX_VALUE) // Emulates very long computation
            42
        } finally {
            println("First child was cancelled")
        }
    }
    val two = async<Int> { 
        println("Second child throws an exception")
        throw ArithmeticException()
    }
    one.await() + two.await()
}

Обратите внимание, как обе первоначальные async и ожидающая родительская корутина отменяются при ошибке одного из дочерних корутин (именно %%%CODE_BLOCK_42%%):

Second child throws an exception
First child was cancelled
Computation failed with ArithmeticException