Ваш первый сопроцесс с Kotlin

Kotlin 1.1 представил сопроцессы, новый способ написания асинхронного, неблокирующего кода (и многое другое). В этом руководстве мы пройдем некоторые основы использования Kotlin сопроцессов с помощью библиотеки kotlinx.coroutines, которая является набором помощников и обёртки для существующих Java библиотек.

Настройка проекта

Gradle

В IntelliJ IDEA перейдите к Файл -> Новый -> Проект…:

Затем следуйте шагам мастера. У вас будет создан файл build.gradle, с Kotlin, настроенным в соответствии с этим документом. Убедитесь, что он настроен для Kotlin 1.3 или выше.

Так как мы будем использовать kotlinx.coroutines, давайте добавим его последнюю версию в наши зависимости:

dependencies {
    ...
    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.1"
}

Эта библиотека опубликована в репозитории Bintray JCenter, поэтому давайте добавим её:

repositories {
    jcenter()
}

Это всё, мы готовы писать код в src/main/kotlin.

Maven

В IntelliJ IDEA перейдите к Файл -> Новый -> Проект… и отметьте Создать из архетипа:

Затем следуйте шагам мастера. У вас будет создан файл pom.xml, с Kotlin, настроенным в соответствии с этим документом. Убедитесь, что он настроен для Kotlin 1.3 или выше.

<plugin>
    <groupId>org.jetbrains.kotlin</groupId>
    <artifactId>kotlin-maven-plugin</artifactId>
    ...
    <configuration>
        <args>
            <arg>-Xcoroutines=enable</arg>
        </args>
    </configuration>
</plugin>

Так как мы будем использовать kotlinx.coroutines, давайте добавим его последнюю версию в наши зависимости:

<dependencies>
    ...
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-core</artifactId>
        <version>1.4.1</version>
    </dependency>
</dependencies>

Эта библиотека опубликована в репозитории Bintray JCenter, поэтому давайте добавим её:

<repositories>
    ...
    <repository>
        <id>central</id>
        <url>https://jcenter.bintray.com</url>
    </repository>
</repositories>

Это всё, мы готовы писать код в src/main/kotlin.

Мой первый сопроцесс

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

Итак, как мы запускаем сопроцесс? Давайте воспользуемся функцией launch {}:

launch {
    ...
}

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

Давайте посмотрим на полную программу, использующую launch:

import kotlinx.coroutines.*

fun main(args: Array<String>) {
//sampleStart
    println("Start")

    // Start a coroutine
    GlobalScope.launch {
        delay(1000)
        println("Hello")
    }

    Thread.sleep(2000) // wait for 2 seconds
    println("Stop")
//sampleEnd
}

Здесь мы запускаем сопроцесс, который ждёт 1 секунду и печатает Hello.

Мы используем функцию delay(), которая похожа на Thread.sleep(), но лучше: она *не блокирует поток*, а только приостанавливает сам сопроцесс. Поток возвращается в пул, пока сопроцесс ждёт, и когда ожидание завершено, сопроцесс возобновляется на свободном потоке в пуле.

Главный поток (который выполняет функцию main() ) должен ждать, пока наш сопроцесс завершится, иначе программа завершится до того, как Hello будет напечатано.

Упражнение: попробуйте убрать sleep() из программы выше и посмотрите результат.

Если мы попробуем использовать ту же неблокирующую функцию delay() напрямую внутри main(), мы получим ошибку компилятора:

Функции приостановки разрешается вызывать только из сопроцесса или другой функции приостановки

Это связано с тем, что мы не находимся внутри сопроцесса. Мы можем использовать delay, если обернём его в runBlocking {} который запускает сопроцесс и ждёт его завершения:

runBlocking {
    delay(2000)
}

Итак, сначала программа выводит Start, затем она запускает сопроцесс через launch {}, затем запускает другой через runBlocking {} и блокируется, пока он не завершится, а затем выводит Stop. Тем временем первый сопроцесс завершается и выводит Hello. Как и потоки, мы сказали вам об этом :)

Запустим много из них

Теперь давайте убедимся, что сопроцессы действительно дешевле, чем потоки. А как насчёт запуска миллиона из них? Давайте сначала попробуем запустить миллион потоков:

val c = AtomicLong()

for (i in 1..1_000_000L)
    thread(start = true) {
        c.addAndGet(i)
    }

println(c.get())

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

Давайте попробуем то же самое с сопроцессами:

val c = AtomicLong()

for (i in 1..1_000_000L)
    GlobalScope.launch {
        c.addAndGet(i)
    }

println(c.get())

Этот пример завершается менее чем за секунду для меня, но он выводит некоторое произвольное число, потому что некоторые сопроцессы не завершаются, прежде чем main() выводит результат. Давайте исправим это.

Мы могли бы использовать те же средства синхронизации, которые применимы к потокам (блокировка — то, что приходит мне на ум в этом случае), но давайте выберем более безопасный и чистый путь.

Async: возвращение значения из сопроцесса

Другой способ запуска сопроцесса — async {} . Это похоже на launch {}, но возвращает экземпляр Deferred<T>, у которого есть функция await() , которая возвращает результат сопроцесса. Deferred<T> — это очень базовый фучер (полноценные фучеры JDK также поддерживаются, но здесь мы ограничимся Deferred пока).

Давайте создадим миллион сопроцессов снова, сохраняя их объекты Deferred. Теперь нет необходимости в атомарном счётчике, поскольку мы можем просто возвращать числа, подлежащие добавлению, из наших сопроцессов:

val deferred = (1..1_000_000).map { n ->
    GlobalScope.async {
        n
    }
}

Все они уже запущены, всё, что нам нужно, это собрать результаты:

val sum = deferred.sumOf { it.await().toLong() }

Мы просто берём каждый сопроцесс и ждём его результата здесь, а затем все результаты складываются вместе с помощью стандартной функции библиотеки sumOf(). Но компилятор справедливо жалуется:

Функции приостановки разрешается вызывать только из сопроцесса или другой функции приостановки

await() не может быть вызван за пределами сопроцесса, потому что ему нужно приостановиться, пока вычисление не завершится, а только сопроцессы могут приостанавливаться неблокирующим способом. Итак, давайте поместим это внутрь сопроцесса:

runBlocking {
    val sum = deferred.sumOf { it.await().toLong() }
    println("Sum: $sum")
}

Теперь он выводит осмысленное значение: 500000500000, потому что все сопроцессы завершены.

Давайте также убедимся, что наши сопроцессы действительно работают параллельно. Если мы добавим 1-секундную delay() к каждому из async , результирующая программа не будет работать в течение 1 000 000 секунд (более 11,5 дней):

val deferred = (1..1_000_000).map { n ->
    GlobalScope.async {
        delay(1000)
        n
    }
}

На моём компьютере это занимает около 10 секунд, так что да, сопроцессы действительно выполняются параллельно.

Функции приостановки

Теперь давайте скажем, что мы хотим извлечь нашу *нагрузку* (которая «ждёт 1 секунду и возвращает число») в отдельную функцию:

fun workload(n: Int): Int {
    delay(1000)
    return n
}

Появляется знакомая ошибка:

Функции приостановки разрешается вызывать только из сопроцесса или другой функции приостановки

Давайте немного углубимся в то, что это значит. Самое главное преимущество сопроцессов — это возможность *приостановки* без блокирования потока. Компилятор должен сгенерировать специальный код, чтобы это стало возможным, поэтому мы должны явно пометить функции, которые *могут приостанавливаться*, в коде. Мы используем модификатор suspend для этого:

suspend fun workload(n: Int): Int {
    delay(1000)
    return n
}

Теперь, когда мы вызываем workload() из сопроцесса, компилятор знает, что он может приостановиться и подготовится соответствующим образом:

GlobalScope.async {
    workload(n)
}

Наша функция workload() может быть вызвана из сопроцесса (или другой приостанавливающейся функции), но *не может* быть вызвана извне сопроцесса. Естественно, delay() и await() , которые мы использовали выше, сами объявлены как suspend, и именно поэтому нам нужно было поместить их внутрь runBlocking {}, launch {} или async {}.