Всем привет! Я Дима, Android-разработчик в KTS, и я большой фанат языка Kotlin и библиотеки kotlinx.coroutines.

Здоровый человек может задать справедливый вопрос: зачем нужна очередная статья про корутины, когда есть ИИ? Любая модель моментально расскажет, как написать код с корутинами: где поставить suspend, что передать в launch, какой диспатчер выбрать. Но чтобы понять, почему корутины устроены именно так, нужна выстроенная последовательность из десятка шагов, где каждый следующий опирается на предыдущий. Не осмыслив их устройство, здоровый человек не сможет понять, почему suspend-функцию нельзя вызвать из обычной.

У меня на это ушло много времени. Я писал рабочий код с корутинами, он работал, и всё это время я не до конца понимал, что происходит под капотом. Понимание пришло, когда я наконец сел и разобрал одну-единственную функцию: suspendCoroutine.

Так что я решил написать цикл статей про корутины, чтобы читателям Хабра было проще прийти к этому пониманию, чем мне. Думаю, он будет полезен и Android-разработчикам, и бэкендерам-котлинистам, и iOS-ерам, причём как начинающим, так и матерым сеньорам.

Эта статья будет вводной, и в ней я разберу понятия suspend, Continuation, CPS и стейт-машины.

Оглавление

• Ключевое понятие

• Три слоя, которые часто путают

• Синхронный и асинхронный стиль

  • Задача, с которой всё начинается

  • Как есть и как хочется

• suspendCoroutine: первое знакомство

  • А где здесь корутина?

  • Что такое корутина, версия 1

• suspend и Continuation

  • Что такое Continuation

• Приостановка — это обычный return

  • Что на самом деле делает suspendCoroutine

  • Кто продолжает выполнение

  • Потоки и почему дальше мы про них забудем

• Стейт-машина: почему одна функция вызывается несколько раз

  • Посмотреть своими глазами

  • Что такое корутина, версия 2

• Кто и как этим пользуется

  • Два вызова рядом

  • Всегда используйте Cancellable-версию

• Итоги

Типичное определение корутин звучит так:

«Корутины — это легковесные потоки, которые позволяют писать асинхронный код в синхронном стиле, избавляют от callback hell и не блокируют потоки».

Здесь четыре утверждения. Три из них верны, а одно вводит в заблуждение. К концу статьи вы будете точно знать, какое именно.

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

• «лёгкие потоки»

• «асинхронный код в синхронном стиле»

• «не блокируют потоки»

• «избавляют от callback hell»

Мы пойдём не с launch, а с той единственной функции, из которой выросла вся библиотека: suspendCoroutine.

Слово «корутины» обозначает три разные вещи, и живут они на трёх разных уровнях.

1. Компилятор, то есть сам язык. Ключевое слово suspend и превращение функции в стейт-машину. Это фича языка Kotlin.

2. Стандартная библиотека, пакет kotlin.coroutines. Здесь лежат интерфейс Continuation и функции suspendCoroutine и startCoroutine. Это уже библиотека, пусть и стандартная.

3. kotlinx.coroutines, отдельная библиотека. В ней живут launch, async, Job, CoroutineScope, диспатчеры и Flow. Здесь же появляется класс AbstractCoroutine, и корутина становится самостоятельным объектом.

Введём два термина.

Асинхронность — свойство задачи, результат которой появляется не в момент вызова, а позже. Так устроены сетевой запрос, чтение файла, ожидание геолокации.

Синхронный и асинхронный стиль — это способ записи кода.

// синхронный стиль: результат слева от =
val user = loadUser(1)
println(user.name)

// асинхронный стиль: результат приезжает в колбэк
loadUser(1) { user ->
    println(user.name)
}

Синхронный стиль читается сверху вниз, как обычная программа. Асинхронный читается вложенно, и чем больше шагов, тем глубже вложенность.

Есть сетевой клиент, который умеет загружать профиль пользователя. Результат он отдаёт в колбэк:

data class User(val id: Int, val name: String)
fun loadUser(userId: Int, onResult: (User) -> Unit) {
    thread {
        println("[network] запрос пошёл на потоке: ${Thread.currentThread().name}")
        Thread.sleep(1000L)          // здесь была бы реальная работа с сокетом
        onResult(User(userId, "Дима"))
    }
}

Так устроен почти любой асинхронный API. В Android это OkHttp.enqueue() и LocationManager.requestLocationUpdates(), в iOS URLSession.dataTask(with:completionHandler:) и CLLocationManager с делегатом. Сюда же относятся CompletableFuture в Java и промисы в JavaScript. Все они сводятся к одной идее: вот тебе функция, позови её, когда будет результат.

И тут возникает проблема, которая не имеет никакого отношения к потокам:

fun getUser(): User {
    loadUser(1) { user ->
        return user   // так нельзя: мы возвращаемся из лямбды, а не из getUser()
    }
    // сюда мы попадаем МГНОВЕННО, запустив запрос. Результата ещё нет. Что вернуть?
}

Функция getUser() физически не может вернуть результат. Когда она доходит до конца тела, результата ещё не существует. А когда результат приходит в колбэк, функция уже завершилась, её кадр стека уничтожен и локальные переменные потеряны.

Отсюда и проблема. getUser() вернулась, её кадра больше нет, и возвращать результат через секунду просто некуда.

Компилятор Kotlin такое и не пропустит. Нужно либо вернуть настоящий User, которого нет, либо объявить тип возврата Unit.

Это корень всех проблем колбэчных API. Отсюда растут трудности с обработкой ошибок, callback hell и всё остальное.

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

Как есть, асинхронный стиль:

fun showFirstPostTitle(api: ApiService) {
    api.loadUser(1) { user ->
        api.loadUserPosts(user) { posts ->
            api.loadPostDetails(posts.postIds.first()) { post ->
                println(post.title)
            }
        }
    }
}

Уже неприятно, а мы ещё даже не обработали ошибки. Добавим их, и получим то, что называют callback hell:

fun main() {
    doAsyncTask1 { result1 ->
        if (result1.isSuccess) {
            doAsyncTask2(result1.value) { result2 ->
                if (result2.isSuccess) {
                    doAsyncTask3(result2.value) { result3 ->
                        if (result3.isSuccess) {
                            println(result3.value)
                        } else {
                            handleError(result3.error)
                        }
                    }
                } else {
                    handleError(result2.error)
                }
            }
        } else {
            handleError(result1.error)
        }
    }
}

Как хочется, синхронный стиль:

fun showFirstPostTitle(api: ApiService) {
    val user = api.loadUser(1)
    val posts = api.loadUserPosts(user)
    val post = api.loadPostDetails(posts.postIds.first())
    println(post.title)
}

Четыре строки, читаются сверху вниз, ошибки ловятся обычным try/catch. Вот только в таком виде код не будет асинхронным. Чтобы loadUser вернула User обычным return, ей придётся заблокировать поток до ответа сети.

Всё, что делают корутины — позволяют написать второй вариант так, чтобы он работал как первый.

Вот минимальный рабочий пример целиком:

data class User(val id: Int, val name: String)

// Легаси-API: результат отдаёт в колбэк, работу делает на своём потоке.
// Та же реализация, что выше
fun loadUser(userId: Int, onResult: (User) -> Unit) { ... }

// Обёртка: тот же вызов, но результат возвращается «наверх»
suspend fun getUser(userId: Int): User = suspendCoroutine { continuation ->
    loadUser(userId) { user ->
        println("[resume] вызван на потоке: ${Thread.currentThread().name}")
        continuation.resume(user)
    }
}

suspend fun main() {
    println("main на потоке: ${Thread.currentThread().name}")
    println("До приостановки")
    val user: User = getUser(1)
    println("Получили: $user")
    println("После приостановки")
}

В строчке val user: User = getUser(1) слева от = стоит переменная типа User, и она действительно получит объект. Хотя внутри getUser живёт loadUser, которая по-прежнему ничего не возвращает.

При этом getUser() не вернула значение обычным return. Так не бывает, мы только что это разобрали. Функция всё ещё завершается мгновенно, до того как появится результат.

То есть val user: User = getUser(1) читается как «попросили доставить User в переменную, когда будет готово».

Так можно писать, потому что приёмник результата написал за вас компилятор. Раз он его написал, он знает его тип и может показать этот тип в позиции возврата. Что это за приёмник, разберём через минуту. Это и есть Continuation.

suspendCoroutine объявлена как inline, а её лямбда как crossinline. Практический вывод один: из этой лямбды нельзя выйти через return. Подробности в документации по non-local return.

В коде выше нет ни launch, ни CoroutineScope, ни других корутин-билдеров. Слово «coroutine» встречается только в имени функции.

Корутина здесь всё же есть, и её создал компилятор из suspend fun main(). Когда вы помечаете main словом suspend, компилятор генерирует настоящую точку входа, которая запускает ваш main как корутину и ждёт её завершения.

Дадим первое, грубое определение. Уточним его в разделе 5.

Корутина — это ваш код и объект, который помнит, где этот код остановился. То есть блок кода, который может остановиться и продолжиться когда-то позже, возможно, на другом потоке.

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

ВАЖНО

Слово suspend относится к корутине. Сама функция ничего не приостанавливает. На паузу встаёт корутина, в пределах которой эта функция выполняется.

Посмотрите, что компилятор делает с сигнатурой:

// то, что пишете вы
suspend fun getUser(id: Int): User
suspend fun saveUser(user: User)
suspend fun checkAvailability(flight: Flight): Boolean

// то, что видит JVM (упрощённо)
fun getUser(id: Int, continuation: Continuation<User>): Any?
fun saveUser(user: User, continuation: Continuation<Unit>): Any?
fun checkAvailability(flight: Flight, continuation: Continuation<Boolean>): Any?

suspend просто дописывает в конец списка параметров ещё один, типа Continuation. Компилятор передаёт его из функции в функцию автоматически.

Отсюда сразу следует правило: suspend-функцию можно вызвать только из другой suspend-функции или из корутины. Причина чисто механическая. Обычной функции неоткуда взять Continuation, чтобы его передать.

Так же устроен компилятор-плагин Jetpack Compose. В @Composable-функции компилятор подставляет скрытый параметр Composer, и по той же причине @Composable нельзя вызвать из обычной функции.

Тип результата переехал из одной позиции в другую:

suspend fun getUser(id: Int): User
// User в позиции РЕЗУЛЬТАТА

fun getUser(id: Int, c: Continuation<User>): Any?
// User в позиции ПАРАМЕТРА

Этот приём называется CPS, Continuation Passing Style. Идея та же, что у колбэков. Вместо того чтобы возвращать результат наверх, функция получает аргументом то, что нужно сделать дальше.

Continuation<T> — интерфейс из stdlib, и он простой:

public interface Continuation<in T> {
    public val context: CoroutineContext
    public fun resumeWith(result: Result<T>)
}

Одно свойство и один метод, больше ничего.

Дословный перевод — «продолжение». Это объект, внутри которого лежит вся оставшаяся часть вашей функции и функций, которые её вызвали.

Проще всего понять его через сравнение с колбэком, потому что Continuation — это и есть колбэк, только написанный компилятором:

В сигнатуре стоит Result<T>, то есть успех и ошибка едут по одному и тому же каналу. continuation.resume(value) и continuation.resumeWithException(e) — это два удобных расширения над одним resumeWith. Поэтому исключение, брошенное через resumeWithException, вылетит в вызывающем коде как обычное исключение, и его можно поймать обычным try/catch.

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

Теперь можно закрыть первый из четырёх тезисов.

Закрыли: «асинхронный код в синхронном стиле».

val user: User = getUser(1) работает потому, что компилятор сам написал колбэк-приёмник, знает его тип и показывает этот тип слева от =. Асинхронность осталась там же, где была, изменился только синтаксис.

Осталась одна деталь сигнатуры. Почему тип возврата Any?, а не User?

У suspend-функции два разных исхода, и одним типом их не выразить.

1. Результат готов прямо сейчас. Функция отработала синхронно и вернула User. Так бывает, когда значение лежит в кэше, сработал ранний return или ветка if прошла без единой приостановки.

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

Для второго случая в stdlib есть специальный объект-маркер:

val COROUTINE_SUSPENDED: Any   // синглтон-«флажок», больше в нём ничего нет

Итого тип возврата читается как «либо T, либо маркер COROUTINE_SUSPENDED». Общего надтипа получше, чем Any?, для этих двух вариантов не существует.

Вызывающий код после каждого suspend-вызова делает одну и ту же проверку:

val tmp: Any? = getUser(1, cont)          // Any? появился
if (tmp === COROUTINE_SUSPENDED)          // нужно ли приостанавливаться?
    return COROUTINE_SUSPENDED            // да → выходим и передаём сигнал выше
val user: User = tmp as User              // Any? исчез — каст к нужному типу

Приостановка — это обычный return из обычной функции. Поток сворачивает кадры стека и уходит заниматься другими делами. Никто ничего не замораживает, никакой поток не висит в ожидании. Просто функция вышла раньше времени, а знание о том, где она остановилась, осталось лежать объектом Continuation в куче.

return COROUTINE_SUSPENDED идёт вверх по цепочке вызовов, пока не дойдёт до корневой корутины.

Закрыли: «не блокируют потоки».

Блокировать просто нечего. К моменту приостановки кадры стека уже уничтожены, поток свободен и ушёл работать над другими задачами. Невозможность блокировки напрямую следует из того, что приостановка реализована через return.

Теперь можно точно разобрать функцию, с которой мы начали. Она берёт текущую корутину, упаковывает всё её состояние в объект Continuation и отдаёт этот объект вам в лямбду. Когда лямбда отработает, корутина приостановится. Дальше дело за вами. Кто-то должен позже вызвать .resume() и разбудить корутину.

suspend fun getUser(userId: Int): User = suspendCoroutine { continuation ->
    // continuation — «остаток main()», упакованный в объект
    loadUser(userId) { user ->                    // запускаем сетевой запрос (он уйдёт
                                                  // на свой поток и вернётся не скоро)
        continuation.resume(user)                 // а вот отсюда main() продолжится
    }
}

Лямбда выполняется синхронно. Сначала её тело отрабатывает целиком, и только после выхода из неё сгенерированный код возвращает COROUTINE_SUSPENDED. Пока вы внутри лямбды, корутина ещё не приостановлена.

Лямбда выполняется в том же потоке. Никакой смены потока не происходит, потому что suspendCoroutine ничего для этого не делает. Это обычный inline-вызов в текущем кадре стека. Потоки в нашем примере появляются только потому, что loadUser внутри себя вызывает thread { }.

Значит, тяжёлую работу сюда класть нельзя. Задача лямбды в том, чтобы пристроить continuation тому, кто позовёт его позже, и выйти как можно быстрее. Если положить сюда долгую работу без другого потока, мы честно заблокируем текущий поток. А если после этого ещё и вызвать resume прямо внутри лямбды, корутина даже не приостановится, потому что suspendCoroutine просто вернёт результат сразу.

Это последний момент, когда вы можете что-то сделать с continuation до приостановки.

Запустим наш пример и посмотрим вывод:

main на потоке: main
До приостановки
[network] запрос пошёл на потоке: Thread-0
[resume] вызван на потоке: Thread-0
Получили: User(id=1, name=Дима)       <- тело main(), но уже на Thread-0
После приостановки                    <- и это тоже

println("Получили: ...") и println("После приостановки") написаны в теле main(), но выполнились на Thread-0.

Корутина не возвращается на тот поток, с которого ушла. resumeWith — обычный вызов обычного метода, и кто его вызвал, тот и выполняет весь дальнейший код корутины, прямо в своём стеке.

ВАЖНО

Корутина продолжается на том потоке, который позвал resume.

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

Разберём работу программы по картинке. Потоков здесь всего два: main и Thread-0.

1. main вызывает getUser(), а та через suspendCoroutine вызывает loadUser(). loadUser() создаёт новый поток, и вся сетевая работа происходит в нём.

2. getUser() возвращает COROUTINE_SUSPENDED. Для главного потока работы по нашей корутине больше нет.

3. В теле loadUser() на Thread-0 мы ждём User.

4. Запрос доработал до конца и вызвал continuation.resume(user). Это заново вызывает функцию main.

5. main() доделывает всё, что идёт после вызова getUser(), но уже на Thread-0.

Разведём понятия потока и корутины.

Поток — ресурс операционной системы. Под его стек выделяется порядка мегабайта памяти, а переключение между потоками требует участия ОС.

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

Между ними нет жёсткой связи. Одна корутина за свою жизнь может поработать на трёх разных потоках, а тысяча корутин может жить на одном.

suspend не создаёт потоков и не переключает их. Он умеет поставить корутину на паузу и снять с паузы. Многопоточность в нашем примере обеспечивает thread { } внутри loadUser, то есть само асинхронное API. Корутины прячут этот поток за нормальным возвращаемым значением.

ВАЖНО

Корутины — надстройка над асинхронным API. Многопоточность здесь будет просто частным случаем. Асинхронность прекрасно существует и без второго потока, весь JavaScript живёт именно так.

В этой части мы говорим о потоках только потому, что разбираем обёртку над потоковым API. Начиная со второй части потоки почти уйдут из кадра. Останутся корутины и диспатчеры, а вопрос «на каком я сейчас потоке» перестанет вас волновать.

Мы разобрали одну точку приостановки. Теперь возьмём две:

suspend fun loadAll() {
    val user = fetchUser()          // точка приостановки №1
    println(user)
    val posts = fetchPosts(user)    // точка приостановки №2
    println(posts)
}

Компилятор режет эту функцию на куски по точкам приостановки и превращает в конечный автомат.

Стейт-машина (конечный автомат) — модель поведения, при которой система находится в одном конкретном состоянии из набора и переходит между ними по чётким правилам. В каждый момент времени активно ровно одно состояние.

Вот как выглядят состояния нашей loadAll():

А вот сильно упрощённый, но идейно точный код, в который всё это превращается:

// Объект состояния. Он же — Continuation этой функции.
class LoadAllContinuation(
    val completion: Continuation<Unit>
) : Continuation<Any?> {
    override val context get() = completion.context
    var label = 0                  // «на каком куске мы остановились»
    var lastResult: Any? = null
    var user: User? = null         // локальная переменная переехала в ПОЛЕ объекта
    override fun resumeWith(result: Result<Any?>) {
        lastResult = result.getOrThrow()
        loadAll(this)              // ВЫЗЫВАЕМ ТУ ЖЕ САМУЮ ФУНКЦИЮ ЕЩЁ РАЗ
    }
}

fun loadAll(c: Continuation<Any?>): Any? {
    // первый заход — создаём состояние; повторный — переиспользуем
    val cont = if (c is LoadAllContinuation) c
               else LoadAllContinuation(c as Continuation<Unit>)
    if (cont.label == 0) {                                   // ─── КУСОК 0 ───
        cont.label = 1                                       // закладку ставим ДО вызова
        val r = fetchUser(cont)                              // передаём СЕБЯ как continuation
        if (r === COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED
        cont.lastResult = r                                  // fast path: результат был сразу
    }
    if (cont.label == 1) {                                   // ─── КУСОК 1 ───
        cont.user = cont.lastResult as User
        println(cont.user)
        cont.label = 2
        val r = fetchPosts(cont.user!!, cont)
        if (r === COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED
        cont.lastResult = r
    }
    if (cont.label == 2) {                                   // ─── КУСОК 2 ───
        println(cont.lastResult as Posts)
        cont.completion.resumeWith(Result.success(Unit))     // сообщаем вызывающему
        return Unit
    }
    error("unreachable")
}

Вся работа по пробуждению корутины происходит в resumeWith. Он вызывает ту же самую функцию ещё раз, но с уже заполненным состоянием, и switch по label отправляет выполнение сразу в нужный кусок.

Локальные переменные стали полями объекта. user больше не живёт в стеке, она лежит в поле LoadAllContinuation, в куче. Поэтому она и переживает приостановку, хотя кадр стека давно уничтожен.

lastResult имеет тип Any?. Это сделано намеренно, чтобы в него можно было положить что угодно. Сначала туда кладётся User, потом Posts, а компилятор сам расставляет корректные касты.

Верить мне на слово не обязательно. Всё это можно увидеть в реальном коде.

В IntelliJ IDEA или Android Studio откройте Tools → Kotlin → Show Kotlin Bytecode и нажмите Decompile в открывшейся панели. Вы получите Java-код, в который превратилась ваша функция.

Возьмите функцию с двумя suspend-вызовами и посмотрите на неё. Там будет класс-наследник ContinuationImpl с полями label и result, в начале метода switch по label, а ваши локальные переменные окажутся полями этого объекта. Сравнения с IntrinsicsKt.getCOROUTINE_SUSPENDED() тоже на месте.

Читается тяжело, но один раз это стоит сделать.

Теперь можно дать точное определение вместо грубого.

Корутина — это объект, конкретный экземпляр класса в куче, и этот объект реализует Continuation.

Осталось самое проблемное обещание из тезиса.

ВАЖНО: «лёгкие потоки» — неверно.

Корутина — объект в куче плюс стейт-машина. Слово «лёгкие» справедливо: один объект против примерно мегабайта стека на поток. А вот слово «потоки» вредное. Из-за него кажется, что у корутины есть свой стек, который где-то ждёт, и что Thread.sleep внутри корутины — это нормально. Ни то, ни другое не верно.

Continuation — низкоуровневый API, и обращаться с ним надо аккуратно. Мы уже знаем, что внутри него лежит стейт-машина с сохранёнными локальными переменными и ссылками на другие continuation'ы. Сломать это легко.

Объект continuation должен уйти ровно в одно место, к тому, кто позовёт resume один раз. Складывать его в поля, кэшировать и передавать дальше может только компилятор.

Вот код, который на первый взгляд выглядит нормально, но при этом плох сразу тремя способами:

// ТАК ДЕЛАТЬ НЕЛЬЗЯ

object LocationHolder {
    var continuation: Continuation<Location>? = null
}

suspend fun awaitLocation(): Location = suspendCoroutine { cont ->
    LocationHolder.continuation = cont       // «сохраню, позову позже»
    locationManager.startUpdates()
}

// где-то в другом месте
fun onLocationChanged(loc: Location) {
    LocationHolder.continuation?.resume(loc)
}

Что здесь сломается.

1. Двойной resume. onLocationChanged — это подписка на обновления, и он вызовется много раз. Второй вызов resume на том же continuation даст IllegalStateException: Already resumed.

2. Утечка памяти. Пока LocationHolder держит ссылку на continuation, живо всё, что тот захватил, а захватил он остаток вашей функции вместе с её локальными переменными. Если функция жила внутри Activity или ViewController, экран не получится уничтожить даже после того, как пользователь с него ушёл.

3. resume после отмены. Корутину отменили, экран закрыт, а колбэк от locationManager приходит через две секунды и дёргает сохранённый continuation. В лучшем случае ничего не произойдёт, в худшем вы получите поведение, которое невозможно воспроизвести.

ВАЖНО

continuation живёт от входа в лямбду suspendCoroutine до момента, когда вы отдали его тому, кто позовёт resume один раз.

В Swift хранить CheckedContinuation в свойстве класса (например, для API на делегатах), наоборот, распространённый паттерн. Оговорки те же. После resume его нужно обнулять, а время жизни привязывать к владеющему объекту. Подробнее: How to store continuations to be resumed later.

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

interface Callback<T> {
    fun onResponse(call: Call<T>, response: T)
    fun onFailure(call: Call<T>, t: Throwable)
}

interface Call<T> {
    fun enqueue(callback: Callback<T>)
    fun cancel()
}

data class User(val id: Int, val name: String)
data class UserPosts(val userId: Int, val postIds: List<String>)
data class PostDetails(val id: String, val title: String, val body: String)

interface ApiService {
    // колбэчный вариант
    fun getUser(userId: Int): Call<User>
    fun getUserPosts(user: User): Call<UserPosts>
    fun getPostDetails(postId: String): Call<PostDetails>

    // корутинный вариант
    suspend fun getUserSuspend(userId: Int): User
    suspend fun getUserPostsSuspend(user: User): UserPosts
    suspend fun getPostDetailsSuspend(postId: String): PostDetails
}

Колбэчные методы возвращают объект вызова Call<T>, чтобы запрос можно было где-то сохранить и при необходимости отменить. Точно так же устроены Retrofit и OkHttp.

Реализация Call — всё то же асинхронное и многопоточное API:

class FakeCall<T>(private val generator: () -> T) : Call<T> {
    @Volatile private var isCancelled: Boolean = false
    private var executionThread: Thread? = null
    override fun enqueue(callback: Callback<T>) {
        thread {
            executionThread = Thread.currentThread()
            try {
                Thread.sleep(1000L)
                if (isCancelled) return@thread
                if (Random.nextInt(100) < 80) {
                    callback.onResponse(this, generator())
                } else {
                    callback.onFailure(this, RuntimeException("Network error"))
                }
            } catch (e: InterruptedException) {
                // поток прервали через cancel() — просто выходим
            }
        }
    }
    override fun cancel() {
        isCancelled = true
        executionThread?.interrupt()
    }
}

suspend-варианты держатся на одной extension-функции:

suspend fun <T> Call<T>.await(): T = suspendCancellableCoroutine { continuation ->
    enqueue(object : Callback<T> {
        override fun onResponse(call: Call<T>, response: T) {
            continuation.resume(response)
        }
        override fun onFailure(call: Call<T>, t: Throwable) {
            continuation.resumeWithException(t)
        }
    })
    continuation.invokeOnCancellation {
        cancel()
    }
}

Вот и вся обёртка. В enqueue мы отдаём колбэк, который дёргает методы Continuation. Успех уходит через resume, ошибка через resumeWithException, и, как мы помним, в вызывающем коде она вылетит обычным исключением.

Тяжёлой работы здесь нет. Лямбда назначает колбэки, вешает обработчик отмены и выходит.

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

Extension здесь не случаен. Так обёртку можно прикрутить к чужому интерфейсу, исходники которого вам недоступны. Дальше suspend-методы пишутся в одну строку:

class FakeApiService : ApiService {
    override fun getUser(userId: Int): Call<User> = FakeCall {
        User(userId, "John Doe")
    }
    override fun getUserPosts(user: User): Call<UserPosts> = FakeCall {
        UserPosts(user.id, listOf("post_1", "post_2", "post_3"))
    }
    override fun getPostDetails(postId: String): Call<PostDetails> = FakeCall {
        PostDetails(postId, "Title of $postId", "Body of $postId")
    }
    override suspend fun getUserSuspend(userId: Int): User =
        getUser(userId).await()
    override suspend fun getUserPostsSuspend(user: User): UserPosts =
        getUserPosts(user).await()
    override suspend fun getPostDetailsSuspend(postId: String): PostDetails =
        getPostDetails(postId).await()
}

Так suspendCoroutine и использовали с самого начала, как мост совместимости. Когда корутины появились, весь существующий мир был колбэчным, и нужен был способ подключить его к новому механизму без переписывания. Разработчики писали такие .await(), и библиотека становилась coroutine-ready. Сегодня большинство библиотек даёт suspend-API из коробки, но suspendCancellableCoroutine никуда не делась. Она нужна всякий раз, когда вы имеете дело с колбэчным API.

Вернёмся к задаче из первого раздела. Колбэчный вариант со всеми ветками ошибок:

fun loadFirstPostDetails(api: ApiService) {
    api.getUser(1).enqueue(object : Callback<User> {
        override fun onResponse(c: Call<User>, user: User) {
            api.getUserPosts(user).enqueue(object : Callback<UserPosts> {
                override fun onResponse(c: Call<UserPosts>, posts: UserPosts) {
                    api.getPostDetails(posts.postIds.first()).enqueue(object : Callback<PostDetails> {
                        override fun onResponse(c: Call<PostDetails>, post: PostDetails) {
                            println("Post: ${post.title}")
                        }
                        override fun onFailure(c: Call<PostDetails>, t: Throwable) {
                            println("Post error")
                        }
                    })
                }
                override fun onFailure(c: Call<UserPosts>, t: Throwable) {
                    println("Posts error")
                }
            })
        }
        override fun onFailure(c: Call<User>, t: Throwable) {
            println("User error")
        }
    })
}

Три вызова, и вот он, callback hell. Ошибки обрабатываются в трёх разных местах, объединить их нельзя, а порядок чтения кода не имеет ничего общего с порядком выполнения.

С корутинами:

Читается сверху вниз. Ошибки ловятся одним try/catch вокруг всего блока. В IDE каждый suspend-вызов помечен стрелочкой на полях, по ней сразу видно точки приостановки.

Закрыли: «избавляют от callback hell».

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

Маленькое упражнение. Сколько точек приостановки в loadFirstPostDetailsClean? А сколько значений label будет у сгенерированной стейт-машины?

Точек suspend-а - три, label - четыре: 0, 1, 2 и 3. Последнее нужно, чтобы третьему resume было куда приземлиться, иначе финальный println некому будет выполнить.

В прошлом разделе вместо знакомой suspendCoroutine появилась suspendCancellableCoroutine.

suspendCoroutine ничего не знает об отмене. Если корутину отменят, пока запрос выполняется, она об этом не узнает и останется приостановленной, а сетевой запрос как ни в чём не бывало продолжит работать. Соединение занято, поток занят, батарея тратится, результат никому не нужен. И так с каждым отменённым запросом.

suspendCancellableCoroutine отдаёт вам CancellableContinuation. Это тот же Continuation, но с дополнительными методами, и главный из них такой:

continuation.invokeOnCancellation {
    cancel()          // а вот здесь мы гасим сам легаси-вызов
}

Когда корутину отменяют, этот обработчик даёт вам последний шанс прибраться. Поэтому в интерфейсе Call и есть метод cancel(). Без него отменять было бы нечего.

ВАЖНО

В реальном коде оборачивайте легаси-API через suspendCancellableCoroutine. Обычная suspendCoroutine годится для учебных примеров и для операций, которые отменить невозможно. Во всех остальных случаях она молча ломает отмену.

В статье мы разобрали suspendCoroutine, потому что она проще и лучше показывает механику. Но в продакшене нужно писать cancellable-версию.

Как отмена устроена изнутри, почему она распространяется по дереву корутин и почему CancellationException обрабатывается особым образом, разберём в третьей части.

Вернёмся к тезису, с которого начали. Теперь его можно прочитать целиком и с полным пониманием:

«Корутины — это легковесные потоки объекты, которые позволяют писать асинхронный код в синхронном стиле, избавляют от callback hell и не блокируют потоки».

Девять тезисов, которые стоит зафиксировать.

1. Корутина — это объект. Наследник BaseContinuationImpl на уровне stdlib, AbstractCoroutine в kotlinx.coroutines.

2. suspend-функция — обычная функция, в которую компилятор дописал один параметр типа Continuation. Поэтому suspend можно звать только из suspend или из корутины, потому что обычной функции неоткуда взять Continuation.

3. Continuation — это колбэк, написанный компилятором. Внутри лежит весь остаток вашей функции. Возобновить его можно один раз.

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

5. Функция превращается в стейт-машину. Локальные переменные переезжают из стека в поля объекта, поэтому и переживают приостановку.

6. Приостановка — это return COROUTINE_SUSPENDED. Он идёт вверх по цепочке до корневой корутины. Когда говорят, что корутина заснула, это значит, что все её функции сделали return и исполнять нечего.

7. Пробуждение — это повторный вызов той же функции. resumeWith зовёт её заново, а switch по label отправляет в нужный кусок. Стек не восстанавливается, вместо него из объекта в куче восстанавливается состояние.

8. Корутина продолжается на том потоке, который позвал resume. За возврат на нужный поток отвечают диспатчеры.

9. Для обёртки легаси-API всегда берётся suspendCancellableCoroutine. Обычная версия молча ломает отмену.

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

• Как новый компилятор K2 ускоряет компиляцию Kotlin на 94%

• За 3 дня запустить Android приложение на iOS: опыт адаптации приложения под CMP

• KMP глазами iOS-разработчика

• Compose Multiplatform 1.8.0: поддержка iOS переходит в stable

Проголосовали 4 пользователя. Воздержавшихся нет.