Асинхронность и сопрограммы

Асинхронная и параллельная обработка

Асинхронная и параллельная обработка — это разные вещи.

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

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

val client = HttpClient()
//Running in the main thread, start a `get` call
client.get<String>("https://example.com/some/rest/call")
//The get call will suspend and let other work happen in the main thread, and resume when the get call completes

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

Диспетчер для изменения потоков

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

suspend fun differentThread() = withContext(Dispatchers.Default){
    println("Different thread")
}

withContext принимает как аргумент диспетчер и блок кода, который будет выполняться в потоке, определённом диспетчером. Подробнее о контексте и диспетчерах сопрограмм.

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

Замороженные захваченные данные

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

fun <R> runOnDifferentThread(functionBlock: () -> R)

Вы вызовете эту функцию следующим образом:

runOnDifferentThread {
    //Code run in another thread
}

Как описано в обзоре асинхронности, состояние, общее для потоков в Kotlin/Native, должно быть заморожено. Функциональный аргумент является состоянием, которое замораживается вместе со всем, что оно захватывает.

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

В следующем примере экземпляр класса данных dc будет захвачен блоком функции и будет заморожен при переходе между потоками. Выражение println выведет true.

val dc = DataClass("Hello")
withContext(Dispatchers.Default) {
    println("${dc.isFrozen}")
}

При выполнении параллельного кода будьте внимательны с захваченным состоянием. Иногда очевидно, когда состояние будет захвачено, но не всегда. Например:

class SomeModel(val id:IdRec){
    suspend fun saveData() = withContext(Dispatchers.Default){
        saveToDb(id)
    }
}

Код внутри saveData выполняется в другом потоке. Это заморозит id, но так как id является свойством родительского класса, то родительский класс также заморозится.

Замороженные возвращаемые данные

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

val dc = withContext(Dispatchers.Default) {
    DataClass("Hello Again")
}

println("${dc.isFrozen}")

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

Дополнительную информацию см. в разделе состояние, изолированное от потоков.

CoroutineWorker

CoroutinesWorker — это библиотека, опубликованная AutoDesk, которая реализует некоторые функции сопрограмм через потоки с использованием однопоточной версии kotlinx.coroutines.

Для простых отложенных функций это довольно хороший вариант, но он не поддерживает Flow и другие структуры.

Reaktive

Reaktive — это библиотека, похожая на Rx, которая реализует реактивные расширения для Kotlin Multiplatform. Она имеет некоторые расширения для сопрограмм, но в основном ориентирована на RX и потоки.

Пользовательский процессор

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

Платформенная конкурентность

В производственной среде вы также можете положиться на платформу для обработки конкурентности. Это может быть полезно, если общий код Kotlin будет использоваться для бизнес-логики или операций с данными, а не для архитектуры.

Чтобы сохранить состояние в iOS между потоками, это состояние должно быть заморожено. Библиотеки конкурентности, упомянутые здесь, автоматически заморозить ваши данные. Вам редко придётся делать это явно, если вообще понадобится.

Если вы возвращаете данные на платформу iOS, которые должны быть доступны в нескольких потоках, убедитесь, что данные заморожены перед выходом за пределы iOS.

В Kotlin есть понятие заморозки только для платформ Kotlin/Native, включая iOS. Чтобы сделать freeze доступным в общем коде, вы можете создать ожидаемые и фактические реализации для freeze, или использовать stately-common, который предоставляет эту функциональность. В Kotlin/Native freeze заморозит ваше состояние, а на JVM ничего не сделает.

Чтобы использовать stately-common, добавьте зависимость для набора источников commonMain в build.gradle(.kts):

val commonMain by getting {
    dependencies {
        implementation ("co.touchlab:stately-common:1.0.x")
    }
}

commonMain {
    dependencies {
        implementation 'co.touchlab:stately-common:1.0.x'
    }
}

Этот материал подготовлен компанией Touchlab для публикации JetBrains.