Композиция приостанавливающих функций

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

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

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
}

Результат будет примерно таким:

The answer is 42
Completed in 1017 ms

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

Ленивый запуск 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()
}

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

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