Основы корутин

Приостанавливаемые функции

Самый простой строительный блок корутин — это приостанавливаемая функция. Она позволяет приостановить выполняющуюся операцию и продолжить её позже, не изменяя структуру кода.

Чтобы объявить приостанавливаемую функцию, используйте ключевое слово suspend:

suspend fun greet() {
    println("Hello world from a suspending function")
}

Вызывать приостанавливаемую функцию можно только из другой приостанавливаемой функции. Чтобы вызывать приостанавливаемые функции в точке входа приложения Kotlin, пометьте функцию main() ключевым словом suspend:

suspend fun main() {
    showUserInfo()
}

suspend fun showUserInfo() {
    println("Loading user...")
    greet()
    println("User: John Smith")
}

suspend fun greet() {
    println("Hello world from a suspending function")
}

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

Хотя ключевое слово suspend входит в ядро языка Kotlin, большинство возможностей корутин доступны через библиотеку kotlinx.coroutines.

Добавление библиотеки kotlinx.coroutines в проект

Чтобы подключить библиотеку kotlinx.coroutines к проекту, добавьте соответствующую конфигурацию зависимости в зависимости от используемой системы сборки:

// build.gradle.kts
repositories {
    mavenCentral()
}

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")
}

// build.gradle
repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0'
}

<!-- pom.xml -->
<project>
    <dependencies>
        <dependency>
            <groupId>org.jetbrains.kotlinx</groupId>
            <artifactId>kotlinx-coroutines-core</artifactId>
            <version>1.11.0</version>
        </dependency>
    </dependencies>
    ...
</project>

Создание первых корутин

Чтобы создать корутину в Kotlin, необходимо следующее:

• Приостанавливаемая функция.

• Область корутины, в которой она может выполняться, например внутри функции withContext().

• Конструктор корутин, например CoroutineScope.launch(), чтобы запустить её.

• Диспетчер, который определяет, какие потоки она использует.

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

1. Импортируйте библиотеку kotlinx.coroutines:

   import kotlinx.coroutines.*
   

2. Пометьте функции, выполнение которых можно приостановить и возобновить, ключевым словом suspend:

   suspend fun greet() {
       println("The greet() on the thread: ${Thread.currentThread().name}")
   }
   
   suspend fun main() {}
   

3. Добавьте функцию delay(), чтобы смоделировать приостанавливаемую задачу, например получение данных или запись в базу данных:

   suspend fun greet() {
       println("The greet() on the thread: ${Thread.currentThread().name}")
       delay(1000L)
   }
   

4. Используйте withContext(Dispatchers.Default), чтобы определить точку входа для многопоточного параллельного кода, выполняющегося в общем пуле потоков:

   suspend fun main() {
       withContext(Dispatchers.Default) {
           // Add the coroutine builders here
       }
   }
   

5. Используйте функцию-конструктор корутин, например CoroutineScope.launch(), чтобы запустить корутину:

   suspend fun main() {
       withContext(Dispatchers.Default) { // this: CoroutineScope
           // Starts a coroutine inside the scope with CoroutineScope.launch()
           this.launch { greet() }
           println("The withContext() on the thread: ${Thread.currentThread().name}")
       }
   }
   

6. Объедините эти части, чтобы одновременно запустить несколько корутин в общем пуле потоков:

   // Imports the coroutines library
   import kotlinx.coroutines.*
   
   // Imports the kotlin.time.Duration to express duration in seconds
   import kotlin.time.Duration.Companion.seconds
   
   // Defines a suspending function
   suspend fun greet() {
       println("The greet() on the thread: ${Thread.currentThread().name}")
       // Suspends for 1 second and releases the thread
       delay(1.seconds) 
       // The delay() function simulates a suspending API call here
       // You can add suspending API calls here like a network request
   }
   
   suspend fun main() {
       // Runs the code inside this block on a shared thread pool
       withContext(Dispatchers.Default) { // this: CoroutineScope
           this.launch() {
               greet()
           }
   
           // Starts another coroutine
           this.launch() {
               println("The CoroutineScope.launch() on the thread: ${Thread.currentThread().name}")
               delay(1.seconds)
               // The delay function simulates a suspending API call here
               // You can add suspending API calls here like a network request
           }
   
           println("The withContext() on the thread: ${Thread.currentThread().name}")
       }
   }
   

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

Область корутины и структурированный параллелизм

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

Согласно этому принципу, корутины образуют древовидную иерархию родительских и дочерних задач со связанными жизненными циклами. Жизненный цикл корутины — это последовательность состояний от её создания до завершения, ошибки или отмены.

Родительская корутина ожидает завершения дочерних, прежде чем завершиться самой. Если родительская корутина завершается с ошибкой или отменяется, все её дочерние корутины также рекурсивно отменяются. Такая связь между корутинами делает отмену и обработку ошибок предсказуемыми и безопасными.

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

Вызов функции-конструктора корутин, например CoroutineScope.launch() для CoroutineScope, запускает дочернюю корутину корутины, связанной с этой областью. Внутри блока конструктора получателем является вложенный CoroutineScope, поэтому все запущенные там корутины становятся его дочерними.

// Imports the kotlin.time.Duration to express duration in seconds
import kotlin.time.Duration.Companion.seconds

import kotlinx.coroutines.*

// If the coroutine context doesn't specify a dispatcher,
// CoroutineScope.launch() uses Dispatchers.Default
//sampleStart
suspend fun main() {
    // Root of the coroutine subtree
    coroutineScope { // this: CoroutineScope
        this.launch {
            this.launch {
                delay(2.seconds)
                println("Child of the enclosing coroutine completed")
            }
            println("Child coroutine 1 completed")
        }
        this.launch {
            delay(1.seconds)
            println("Child coroutine 2 completed")
        }
    }
    // Runs only after all children in the coroutineScope have completed
    println("Coroutine scope completed")
}
//sampleEnd

suspend fun main() {
    coroutineScope { // this: CoroutineScope
        // Calls CoroutineScope.launch() where CoroutineScope is the receiver
        this.launch { println("1") }
        this.launch { println("2") }
    } 
}

import kotlinx.coroutines.*
//sampleStart
suspend fun main() {
    coroutineScope {
        launchAll()
    }
}

fun CoroutineScope.launchAll() { // this: CoroutineScope
    // Calls .launch() on CoroutineScope
    this.launch { println("1") }
    this.launch { println("2") } 
}
//sampleEnd
/* -- Calling launch without declaring CoroutineScope as the receiver results in a compilation error --

fun launchAll() {
    // Compilation error: this is not defined
    this.launch { println("1") }
    this.launch { println("2") }
}
 */

Функции-конструкторы корутин

Функция-конструктор корутин — это функция, принимающая suspend лямбду, которая задаёт корутину для выполнения. Например:

• CoroutineScope.launch()

• CoroutineScope.async()

• runBlocking()

• withContext()

• coroutineScope()

Для работы функциям-конструкторам корутин необходима CoroutineScope. Это может быть существующая область или область, созданная с помощью вспомогательных функций, таких как coroutineScope(), runBlocking() или withContext(). Каждый конструктор определяет, как запускается корутина и как взаимодействовать с её результатом.

// Imports the kotlin.time.Duration to enable expressing duration in milliseconds
import kotlin.time.Duration.Companion.milliseconds

import kotlinx.coroutines.*

suspend fun main() {
    withContext(Dispatchers.Default) {
        performBackgroundWork()
    }
}

//sampleStart
suspend fun performBackgroundWork() = coroutineScope { // this: CoroutineScope
    // Starts a coroutine that runs without blocking the scope
    this.launch {
        // Suspends to simulate background work
        delay(100.milliseconds)
        println("Sending notification in background")
    }

    // Main coroutine continues while a previous one suspends
    println("Scope continues")
}
//sampleEnd

// Imports the kotlin.time.Duration to enable expressing duration in milliseconds
import kotlin.time.Duration.Companion.milliseconds

import kotlinx.coroutines.*

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) { // this: CoroutineScope
    // Starts downloading the first page
    val firstPage = this.async {
        delay(50.milliseconds)
        "First page"
    }

    // Starts downloading the second page in parallel
    val secondPage = this.async {
        delay(100.milliseconds)
        "Second page"
    }

    // Awaits both results and compares them
    val pagesAreEqual = firstPage.await() == secondPage.await()
    println("Pages are equal: $pagesAreEqual")
}
//sampleEnd

import kotlin.time.Duration.Companion.milliseconds
import kotlinx.coroutines.*

// A third-party interface you can't change
interface Repository {
    fun readItem(): Int
}

object MyRepository : Repository {
    override fun readItem(): Int {
        // Bridges to a suspending function
        return runBlocking {
            myReadItem()
        }
    }
}

suspend fun myReadItem(): Int {
    delay(100.milliseconds)
    return 4
}

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

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

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

Библиотека kotlinx.coroutines содержит разные диспетчеры для различных случаев использования. Например, Dispatchers.Default выполняет корутины в общем пуле потоков, обрабатывая задачи в фоновом режиме отдельно от главного потока. Поэтому он хорошо подходит для операций с интенсивной нагрузкой на ЦП, таких как обработка данных.

Чтобы задать диспетчер для конструктора корутин, например CoroutineScope.launch(), передайте его в качестве аргумента:

suspend fun runWithDispatcher() = coroutineScope { // this: CoroutineScope
    this.launch(Dispatchers.Default) {
        println("Running on ${Thread.currentThread().name}")
    }
}

Вместо этого можно использовать блок withContext(), чтобы выполнять весь код внутри него в указанном диспетчере:

// Imports the kotlin.time.Duration to enable expressing duration in milliseconds
import kotlin.time.Duration.Companion.milliseconds

import kotlinx.coroutines.*

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) { // this: CoroutineScope
    println("Running withContext block on ${Thread.currentThread().name}")

    val one = this.async {
        println("First calculation starting on ${Thread.currentThread().name}")
        val sum = (1L..500_000L).sum()
        delay(200L)
        println("First calculation done on ${Thread.currentThread().name}")
        sum
    }

    val two = this.async {
        println("Second calculation starting on ${Thread.currentThread().name}")
        val sum = (500_001L..1_000_000L).sum()
        println("Second calculation done on ${Thread.currentThread().name}")
        sum
    }

    // Waits for both calculations and prints the result
    println("Combined total: ${one.await() + two.await()}")
}
//sampleEnd

Подробнее о диспетчерах корутин и их использовании, в том числе о других диспетчерах, таких как Dispatchers.IO и Dispatchers.Main, см. в разделе Контекст корутин и диспетчеры.

Сравнение корутин и потоков JVM

Хотя корутины — это приостанавливаемые вычисления, которые, как и потоки JVM, выполняют код параллельно, внутри они работают иначе.

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

Корутина, напротив, не привязана к определённому потоку. Она может приостановиться в одном потоке и возобновиться в другом, поэтому множество корутин могут совместно использовать один пул потоков. При приостановке корутины поток не блокируется и остаётся свободным для выполнения других задач. Благодаря этому корутины намного легче потоков и позволяют запускать миллионы корутин в одном процессе, не исчерпывая системные ресурсы.

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

import kotlin.time.Duration.Companion.seconds
import kotlinx.coroutines.*

suspend fun main() {
    withContext(Dispatchers.Default) {
        // Launches 50,000 coroutines that each wait five seconds, then print a period
        printPeriods()
    }
}

//sampleStart
suspend fun printPeriods() = coroutineScope { // this: CoroutineScope
    // Launches 50,000 coroutines that each wait five seconds, then print a period
    repeat(50_000) {
        this.launch {
            delay(5.seconds)
            print(".")
        }
    }
}
//sampleEnd

Теперь рассмотрим тот же пример с использованием потоков JVM:

import kotlin.concurrent.thread

fun main() {
    repeat(50_000) {
        thread {
            Thread.sleep(5000L)
            print(".")
        }
    }
}

Эта версия использует гораздо больше памяти, поскольку каждому потоку нужен собственный стек. Для 50 000 потоков это может составить до 100 ГБ, тогда как такое же количество корутин потребует примерно 500 МБ.

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

Что дальше

• Узнайте больше о комбинировании приостанавливаемых функций в разделе Композиция приостанавливаемых функций.

• Узнайте, как отменять корутины и обрабатывать тайм-ауты, в разделе Отмена и тайм-ауты.

• Подробнее о выполнении корутин и управлении потоками читайте в разделе Контекст корутин и диспетчеры.

• Узнайте, как возвращать несколько значений, вычисляемых асинхронно, в разделе Потоки.