Контекст сопрограммы и диспетчеры

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

Контекст сопрограммы включает диспетчер сопрограммы (см. 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() {
//sampleStart
    newSingleThreadContext("Ctx1").use { ctx1 ->
        newSingleThreadContext("Ctx2").use { ctx2 ->
            runBlocking(ctx1) {
                log("Started in ctx1")
                withContext(ctx2) {
                    log("Working in ctx2")
                }
                log("Back to ctx1")
            }
        }
    }
//sampleEnd    
}

Он демонстрирует несколько новых техник. Одна из них — использование runBlocking со явно заданным контекстом, а другая — использование функции withContext для изменения контекста корутины, оставаясь при этом в той же корутине, как показано в выводе ниже:

[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 новой корутины становится дочерним для задания родительской корутины. Если родительская корутина отменяется, все ее дочерние корутины также отменяются рекурсивно.

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

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

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

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

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")
        252
    }
    val v2 = async(CoroutineName("v2coroutine")) {
        delay(1000)
        log("Computing v2")
        6
    }
    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 и запускаем различные сопрограммы в контексте активности Android для выполнения асинхронных операций по извлечению и обновлению данных, выполнению анимаций и т. д. Все эти сопрограммы должны быть отменены при уничтожении активности, чтобы избежать утечек памяти. Конечно, мы можем вручную управлять контекстами и задачами, чтобы связать жизненные циклы активности и её сопрограмм, но kotlinx.coroutines предоставляет абстракцию, инкапсулирующую это: CoroutineScope. Вы уже знакомы с областью действия сопрограммы, так как все билдеры сопрограмм объявлены как расширения для неё.

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

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

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

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

В нашей главной функции мы создаём активность, вызываем нашу тестовую 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> {
//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!

Как видите, только первые две сопрограммы выведут сообщение, а остальные будут отменены одним вызовом job.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'