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

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

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

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 также является 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
}

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

Лениво запущенный async

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

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, и ожидающая родительская корутина отменяются при ошибке одной из дочерних корутин (в частности, two):

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