Создание функций, приостанавливающих выполнение

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

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

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.

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

Рассмотрим пример 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