Конкурентность и сопрограммы

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

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

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

Следующий код выполняет сетевой вызов с помощью 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.