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

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

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

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, или если вызывается функция start его 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", чтобы подчеркнуть тот факт, что они только запускают асинхронное вычисление, и для получения результата необходимо использовать полученное отложенное значение.

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

Рассмотрим пример Конкурентное использование 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