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

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

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

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

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

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    launch { // context of the parent, main runBlocking coroutine
        println("main runBlocking      : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Unconfined) { // not confined -- will work with main thread
        println("Unconfined            : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Default) { // will get dispatched to DefaultDispatcher 
        println("Default               : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(newSingleThreadContext("MyOwnThread")) { // will get its own new thread
        println("newSingleThreadContext: I'm working in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

Результат выполнения (порядок может отличаться):

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> {
//sampleStart
    launch(Dispatchers.Unconfined) { // not confined -- will work with main thread
        println("Unconfined      : I'm working in thread ${Thread.currentThread().name}")
        delay(500)
        println("Unconfined      : After delay in thread ${Thread.currentThread().name}")
    }
    launch { // context of the parent, main runBlocking coroutine
        println("main runBlocking: I'm working in thread ${Thread.currentThread().name}")
        delay(1000)
        println("main runBlocking: After delay in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

Результат:

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.

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

С помощью отладчика корутин можно:

• Проверить состояние каждой корутины.

• Просмотреть значения локальных и захваченных переменных как выполняющихся, так и приостановленных корутин.

• Просмотреть полный стек создания корутины, а также стек вызовов внутри корутины. Стек включает все кадры со значениями переменных, даже те, которые были бы потеряны при стандартной отладке.

• Получить полный отчёт, содержащий состояние каждой корутины и её стек. Чтобы создать отчёт, щёлкните правой кнопкой мыши на вкладке Корутины, а затем нажмите Получить дамп корутин.

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

Подробнее об отладке корутин читайте в руководстве.

import kotlinx.coroutines.*

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

fun main() = runBlocking<Unit> {
//sampleStart
    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()}")
//sampleEnd    
}

[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

Переключение между потоками

Запустите следующий код с параметром 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> {
//sampleStart
    println("My job is ${coroutineContext[Job]}")
//sampleEnd    
}

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

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

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

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

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

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

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

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

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

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    // 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
//sampleEnd
}

Результат выполнения кода:

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> {
//sampleStart
    // 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")
//sampleEnd
}

Результат будет следующим:

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")) {
//sampleStart
    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()}")
//sampleEnd    
}

Результат выполнения с параметром 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> {
//sampleStart
    launch(Dispatchers.Default + CoroutineName("test")) {
        println("I'm working in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

Результат выполнения кода с параметром JVM -Dkotlinx.coroutines.debug:

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

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

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

Мы управляем жизненными циклами корутин, создавая экземпляр CoroutineScope, связанный с жизненным циклом Activity. Экземпляр 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

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

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> {
//sampleStart
    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
//sampleEnd    
}

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

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

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

import kotlinx.coroutines.*

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

fun main() = runBlocking<Unit> {
//sampleStart
    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()}'")
//sampleEnd    
}

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'