Оригинал: https://kotlinlang.org/docs/coroutine-context-and-dispatchers.html

Контекст корутины и диспетчеры

Корутины всегда выполняются в некотором контексте, представленном значением типа CoroutineContext, определенным в стандартной библиотеке Kotlin.

Контекст корутины представляет собой набор различных элементов. Основными элементами являются Job корутины, который мы видели ранее, и ее диспетчер, рассматриваемый в этом разделе.

Диспетчеры и потоки

Контекст корутины включает диспетчер корутины (см. CoroutineDispatcher), который определяет, какой поток или потоки использует соответствующая корутина для своего выполнения. Диспетчер корутины может ограничить выполнение корутины определенным потоком, отправить ее в пул потоков или позволить ей выполняться без ограничений.

Все билдеры корутин, такие как launch и async, принимают необязательный параметр CoroutineContext, который можно использовать для явного указания диспетчера для новой корутины и других элементов контекста.

Попробуйте следующий пример:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    launch { // контекст родителя, основная корутина runBlocking
        println("main runBlocking      : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Unconfined) { // не ограничено -- будет работать с основным потоком
        println("Unconfined            : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Default) { // будет отправлено в DefaultDispatcher 
        println("Default               : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(newSingleThreadContext("MyOwnThread")) { // получит свой собственный новый поток
        println("newSingleThreadContext: I'm working in thread ${Thread.currentThread().name}")
    }
}

Он выдает следующий вывод (возможно, в другом порядке):

Unconfined            : I'm working in thread main
Default               : I'm working in thread DefaultDispatcher-worker-1
newSingleThreadContext: I'm working in thread MyOwnThread
main runBlocking      : I'm working in thread main

Когда launch { ... } используется без параметров, он наследует контекст (и, следовательно, диспетчер) от CoroutineScope, из которого он запускается. В этом случае он наследует контекст основной корутины runBlocking, которая выполняется в main потоке.

Dispatchers.Unconfined — это специальный диспетчер, который также работает в main потоке, но на самом деле это другой механизм, который будет объяснен позже.

Диспетчер по умолчанию используется, когда в области действия явно не указан другой диспетчер. Он представлен Dispatchers.Default и использует общий фоновый пул потоков.

[newSingleThreadContext] создает поток для запуска корутины. Выделенный поток — очень дорогой ресурс. В реальном приложении он должен быть либо освобожден, когда он больше не нужен, с помощью функции close, либо сохранен в переменной верхнего уровня и повторно использован во всем приложении.

Неограниченный и ограниченный диспетчер

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

С другой стороны, диспетчер по умолчанию наследуется от внешнего CoroutineScope. Диспетчер по умолчанию для корутины runBlocking, в частности, ограничен вызывающим потоком, поэтому его наследование приводит к ограничению выполнения этим потоком с предсказуемым планированием FIFO.

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    launch(Dispatchers.Unconfined) { // не ограничено -- будет работать с основным потоком
        println("Unconfined      : I'm working in thread ${Thread.currentThread().name}")
        delay(500)
        println("Unconfined      : After delay in thread ${Thread.currentThread().name}")
    }
    launch { // контекст родителя, основная корутина runBlocking
        println("main runBlocking: I'm working in thread ${Thread.currentThread().name}")
        delay(1000)
        println("main runBlocking: After delay in thread ${Thread.currentThread().name}")
    }
}

Он выдает следующий вывод:

Unconfined      : I'm working in thread main
main runBlocking: I'm working in thread main
Unconfined      : After delay in thread kotlinx.coroutines.DefaultExecutor
main runBlocking: After delay in thread main

Таким образом, корутина с контекстом, унаследованным от runBlocking {...}, продолжает выполняться в main потоке, а неограниченная возобновляется в потоке исполнителя по умолчанию, который использует функция delay.

Отладка корутин и потоков

Корутины могут приостанавливаться в одном потоке и возобновляться в другом потоке. Даже с однопоточным диспетчером может быть сложно понять, что делала корутина, где и когда, если у вас нет специального инструментария.

Отладка с помощью IDEA

Плагин отладки корутин для Kotlin упрощает отладку корутин в IntelliJ IDEA.

Окно инструмента Debug содержит вкладку Coroutines. На этой вкладке вы можете найти информацию как о запущенных, так и о приостановленных корутинах. Корутины группируются по диспетчеру, на котором они выполняются.

Debugging coroutines

С отладчиком корутин вы можете: * Проверить состояние каждой корутины. * Смотреть значения локальных и захваченных переменных как для запущенных, так и для приостановленных корутин. * Смотреть полный стек создания корутин, а также стек вызовов внутри корутин. В стек входят все кадры с переменными значениями, даже те, которые были бы потеряны при стандартной отладке. * Получить полный отчет, содержащий состояние каждой корутины и ее стека. Чтобы получить его, щелкните правой кнопкой мыши по вкладке Coroutines и выберите Get Coroutines Dump.

Чтобы начать отладку корутины, вам просто нужно установить точки останова и запустить приложение в режиме отладки.

Узнайте больше об отладке корутин в руководстве.

Отладка с использованием логирования

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

Запустите следующий код с параметром JVM -Dkotlinx.coroutines.debug:

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() = runBlocking<Unit> {

    val a = async {
        log("I'm computing a piece of the answer")
        6
    }
    val b = async {
        log("I'm computing another piece of the answer")
        7
    }
    log("The answer is ${a.await() * b.await()}")
 
}

Есть три корутины. Основная корутины (№1) внутри runBlocking и две корутины, вычисляющие отложенные значения a (№2) и b (№3). Все они выполняются в контексте runBlocking и ограничены основным потоком. Вывод этого кода:

[main @coroutine#2] I'm computing a piece of the answer
[main @coroutine#3] I'm computing another piece of the answer
[main @coroutine#1] The answer is 42

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

Перескакивание между потоками

Запустите следующий код с параметром JVM -Dkotlinx.coroutines.debug (см. отладку):

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() {
    newSingleThreadContext("Ctx1").use { ctx1 ->
        newSingleThreadContext("Ctx2").use { ctx2 ->
            runBlocking(ctx1) {
                log("Started in ctx1")
                withContext(ctx2) {
                    log("Working in ctx2")
                }
                log("Back to ctx1")
            }
        }
    }
}

Пример выше демонстрирует новые приемы использования корутин.

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

В результате вывод кода выше будет таким:

[Ctx1 @coroutine#1] Started in ctx1
[Ctx2 @coroutine#1] Working in ctx2
[Ctx1 @coroutine#1] Back to ctx1

В приведенном выше примере используется функция use из стандартной библиотеки Kotlin, чтобы корректно освобождать ресурсы потоков, созданные с помощью newSingleThreadContext, когда они больше не нужны.

Job в контексте

Job корутины является частью ее контекста и может быть извлечен из него с помощью выражения coroutineContext[Job]:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    println("My job is ${coroutineContext[Job]}")
}

В режиме отладки выдает что-то вроде этого:

My job is "coroutine#1":BlockingCoroutine{Active}@6d311334

Обратите внимание, что isActive в CoroutineScope — это просто удобный ярлык для coroutineContext[Job]?.isActive == true.

Дети корутины

Когда корутина запускается в CoroutineScope другой корутины, она наследует свой контекст через CoroutineScope.coroutineContext, и Job новой корутины становится дочерним по отношению к Job родительской корутины. Когда родительская корутина отменяется, все ее потомки также рекурсивно отменяются.

Однако это отношение родитель-потомок может быть явно переопределено одним из двух способов:

Если при запуске корутины явно указана другая область видимости (например, GlobalScope.launch), она не наследует Job от родительской области видимости.
Когда в качестве контекста для новой корутины передается другой объект Job (как показано в примере ниже), он переопределяет Job родительской области видимости.

В обоих случаях запущенная корутина не привязана к области видимости, из которой она была запущена, и работает независимо.

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        // it spawns two other jobs
        launch(Job()) { 
            println("job1: I run in my own Job and execute independently!")
            delay(1000)
            println("job1: I am not affected by cancellation of the request")
        }
        // and the other inherits the parent context
        launch {
            delay(100)
            println("job2: I am a child of the request coroutine")
            delay(1000)
            println("job2: I will not execute this line if my parent request is cancelled")
        }
    }
    delay(500)
    request.cancel() // cancel processing of the request
    println("main: Who has survived request cancellation?")
    delay(1000) // delay the main thread for a second to see what happens
}

Вывод этого кода:

job1: I run in my own Job and execute independently!
job2: I am a child of the request coroutine
main: Who has survived request cancellation?
job1: I am not affected by cancellation of the request

Родительские обязанности

Родительская корутина всегда ожидает завершения всех своих потомков. Родителю не нужно явно отслеживать всех дочерних элементов, которые он запускает, и ему не нужно использовать Job.join, чтобы дождаться их в конце:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        repeat(3) { i -> // launch a few children jobs
            launch  {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, 600ms
                println("Coroutine $i is done")
            }
        }
        println("request: I'm done and I don't explicitly join my children that are still active")
    }
    request.join() // wait for completion of the request, including all its children
    println("Now processing of the request is complete")
}

Результат будет:

request: I'm done and I don't explicitly join my children that are still active
Coroutine 0 is done
Coroutine 1 is done
Coroutine 2 is done
Now processing of the request is complete

Именование корутин для отладки

Автоматически назначаемые идентификаторы хороши, когда корутины часто логируются, и вам просто нужно сопоставить записи логов, поступающие от одной и той же корутины. Однако, когда корутина связана с обработкой определенного запроса или выполнением какой-либо конкретной фоновой задачи, лучше назвать ее явно для отладки. Элемент контекста CoroutineName служит той же цели, что и имя потока. Он включается в имя потока, выполняющего эту корутину, когда включен режим отладки.

Следующий пример демонстрирует эту концепцию:

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() = runBlocking(CoroutineName("main")) {
    log("Started main coroutine")
    // run two background value computations
    val v1 = async(CoroutineName("v1coroutine")) {
        delay(500)
        log("Computing v1")
        6
    }
    val v2 = async(CoroutineName("v2coroutine")) {
        delay(1000)
        log("Computing v2")
        7
    }
    log("The answer for v1 * v2 = ${v1.await() * v2.await()}")
}

Результат, с опцией JVM -Dkotlinx.coroutines.debug, будет:

[main @main#1] Started main coroutine
[main @v1coroutine#2] Computing v1
[main @v2coroutine#3] Computing v2
[main @main#1] The answer for v1 * v2 = 42

Объединение элементов контекста

Иногда нам нужно определить несколько элементов для контекста корутины. Для этого мы можем использовать оператор +. Например, мы можем запустить корутину с явно указанным диспетчером и с явно указанным именем одновременно:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    launch(Dispatchers.Default + CoroutineName("test")) {
        println("I'm working in thread ${Thread.currentThread().name}")
    }
}

Результат, с опцией JVM -Dkotlinx.coroutines.debug, будет:

I'm working in thread DefaultDispatcher-worker-1 @test#2

Область видимости корутины

Давайте объединим наши знания о контекстах, дочерних корутинах и заданиях. Предположим, что в нашем приложении есть объект с жизненным циклом, но этот объект не является корутиной. Например, мы пишем приложение для Android и запускаем различные корутины в контексте Android-активити, чтобы выполнять асинхронные операции по получению и обновлению данных, запускать анимации и т. д. Эти корутины должны отменяться при уничтожении активити, чтобы избежать утечек памяти. Мы, конечно, можем вручную управлять контекстами и заданиями, чтобы связать жизненные циклы активити и ее корутин, но kotlinx.coroutines предоставляет инкапсулирующую это абстракцию: CoroutineScope. Вы уже должны быть знакомы с областью видимости корутины, так как все билдеры корутин объявлены как расширения для нее.

Мы управляем жизненными циклами наших корутин, создавая экземпляр CoroutineScope, привязанный к жизненному циклу нашей активити. Экземпляр CoroutineScope может быть создан фабричными функциями CoroutineScope() или MainScope(). Первая создает область общего назначения, а вторая создает область для приложений пользовательского интерфейса и использует Dispatchers.Main в качестве диспетчера по умолчанию:

class Activity {
    private val mainScope = MainScope()
    
    fun destroy() {
        mainScope.cancel()
    }
    // to be continued ...

Теперь мы можем запускать корутины в области видимости этого Activity, используя заданный mainScope. Для демонстрации мы запускаем десять корутин с разной задержкой:

    // class Activity continues
    fun doSomething() {
        // launch ten coroutines for a demo, each working for a different time
        repeat(10) { i ->
            mainScope.launch {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
                println("Coroutine $i is done")
            }
        }
    }
} // class Activity ends

В функции main мы создаем активити, вызываем тестовую функцию doSomething и уничтожаем активити через 500 мс. Это отменяет все корутины, запущенные из doSomething. Мы видим это по тому, что после уничтожения активити сообщения больше не печатаются, даже если подождать еще немного.

import kotlinx.coroutines.*

class Activity {
    private val mainScope = CoroutineScope(Dispatchers.Default) // use Default for test purposes
    
    fun destroy() {
        mainScope.cancel()
    }

    fun doSomething() {
        // launch ten coroutines for a demo, each working for a different time
        repeat(10) { i ->
            mainScope.launch {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
                println("Coroutine $i is done")
            }
        }
    }
} // class Activity ends

fun main() = runBlocking<Unit> {
    val activity = Activity()
    activity.doSomething() // run test function
    println("Launched coroutines")
    delay(500L) // delay for half a second
    println("Destroying activity!")
    activity.destroy() // cancels all coroutines
    delay(1000) // visually confirm that they don't work
}

Результат этого примера:

Launched coroutines
Coroutine 0 is done
Coroutine 1 is done
Destroying activity!

Как видите, только первые две корутины печатают сообщение, а остальные отменяются одним вызовом mainScope.cancel() в Activity.destroy().

Локальные данные потока

Иногда удобно иметь возможность передавать локальные для потока данные в корутины или между ними. Однако, поскольку корутины не привязаны к какому-либо конкретному потоку, ручная реализация, скорее всего, приведет к шаблонному коду.

Для ThreadLocal на помощь приходит функция расширения asContextElement. Она создает дополнительный элемент контекста, который сохраняет значение данного ThreadLocal и восстанавливает его каждый раз, когда корутина переключает свой контекст.

Это легко продемонстрировать в действии:

import kotlinx.coroutines.*

val threadLocal = ThreadLocal<String?>() // declare thread-local variable

fun main() = runBlocking<Unit> {
    threadLocal.set("main")
    println("Pre-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    val job = launch(Dispatchers.Default + threadLocal.asContextElement(value = "launch")) {
        println("Launch start, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
        yield()
        println("After yield, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    }
    job.join()
    println("Post-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
}

В этом примере мы запускаем новую корутину в пуле фоновых потоков, используя Dispatchers.Default, поэтому она работает в разных потоках из пула, но при этом сохраняет значение локальной переменной потока, которое мы указали с помощью threadLocal.asContextElement(value = "launch"), независимо от того, в каком потоке выполняется корутина. Таким образом, вывод (с отладкой) будет таким:

Pre-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'
Launch start, current thread: Thread[DefaultDispatcher-worker-1 @coroutine#2,5,main], thread local value: 'launch'
After yield, current thread: Thread[DefaultDispatcher-worker-2 @coroutine#2,5,main], thread local value: 'launch'
Post-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'

Легко забыть установить соответствующий элемент контекста. Локальная переменная потока, к которой обращается корутина, может иметь неожиданное значение, если поток, выполняющий корутину, отличается. Чтобы избежать таких ситуаций, рекомендуется использовать метод ensurePresent и быстро завершать выполнение при неправильном использовании.

ThreadLocal имеет первоклассную поддержку и может использоваться с любыми примитивами, предоставляемыми kotlinx.coroutines. Однако у него есть одно ключевое ограничение: когда локальная переменная потока изменяется, новое значение не передается вызывающей корутине (поскольку элемент контекста не может отслеживать все обращения к объекту ThreadLocal), а обновленное значение теряется при следующей приостановке. Используйте withContext для обновления локальной переменной потока в корутине, дополнительные сведения смотрите в asContextElement.

В качестве альтернативы значение может храниться в классе-обертке, таком как class Counter(var i: Int), который, в свою очередь, хранится в локальной переменной потока. Однако в этом случае вы несете полную ответственность за синхронизацию потенциально одновременных изменений переменной в этом классе-обертке.

Для расширенного использования, например для интеграции с логированием MDC, контекстами транзакций или любыми другими библиотеками, которые внутри используют локальные переменные потока для передачи данных, смотрите документацию по интерфейсу ThreadContextElement, который необходимо реализовать.