Корутины Kotlin и virtual threads

← назад к разделу

Классическая проблема серверного кода: поток, ожидающий ответа от базы или соседнего сервиса, занят. Он ничего не делает, но держит мегабайт стека и место в пуле. Тысячи одновременных запросов, тысячи занятых потоков. Kotlin отвечает корутинами, Java 21 отвечает virtual threads. Идея одна, механика разная, и от механики зависит, что ломается. Разберём обе и сравним честно.

Деталь, которую легко пропустить: корутины не часть языка, а библиотека. Слово suspend знает компилятор, а всё остальное, coroutineScope, launch, async, Dispatchers, runBlocking, Flow, приходит из зависимости org.jetbrains.kotlinx:kotlinx-coroutines-core. Без неё ни один пример ниже не соберётся.

suspend: функция, которая умеет ждать бесплатно

suspend fun loadUser(id: String): User =
    httpClient.get("/users/$id").body()

suspend fun loadProfile(id: String): Profile {
    val user = loadUser(id)
    val orders = loadOrders(user.id)
    return Profile(user, orders)
}

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

Бесплатное ожидание берётся не из магии, а из переписывания. Компилятор превращает каждую suspend-функцию в конечный автомат: тело режется на куски по точкам приостановки, номер текущего куска и все локальные переменные складываются в объект Continuation, и функция получает его скрытым последним параметром. Дойдя до loadUser, loadProfile не ждёт: она сохраняет «я на шаге 1, id такой-то» и возвращает управление потоку. Когда ответ приходит, библиотека вызывает продолжение, автомат прыгает на шаг 2 и достаёт переменные из объекта.

Из той же механики два следствия, которые удивляют на первой неделе. Стектрейс исключения из продолжения начинается с середины: настоящего стека вызовов нет, он лежал в Continuation, и в трейсе видны только кадры библиотеки. И второе: компилятор не помешает позвать из suspend-функции обычный блокирующий вызов, JDBC, старый HTTP-клиент, Thread.sleep. Ошибки не будет, просто вызов займёт поток диспетчера целиком, а потоков в нём немного. Деградация приходит не постепенно: пока свободные потоки есть, всё хорошо, а когда заняты все, встаёт вообще всё.

Позвать suspend-функцию можно только из другой suspend-функции или из запущенной корутины. Точка входа из обычного мира одна, runBlocking: он запускает корутину и блокирует текущий поток, пока она не закончится. Уместен он в трёх местах: main, тест, граница со старым кодом. Внутри корутин runBlocking ошибка: он держит поток диспетчера ровно так, как держал бы JDBC.

Куда уходит ошибка: launch, async и scope

Корутину запускают двумя способами, и различаются они тем, куда уходит исключение. launch запускает задачу, результат которой не нужен; исключение из неё не возвращается никому, а уходит родителю и валит его. Поймать его снаружи можно только CoroutineExceptionHandler в контексте корневой корутины, у вложенных он не срабатывает. async запускает задачу ради результата; исключение сохраняется в Deferred и выбрасывается там, где вызвали await().

suspend fun loadDashboard(id: String): Dashboard = coroutineScope {
    val user = async { loadUser(id) }
    val stats = async { loadStats(id) }
    Dashboard(user.await(), stats.await())
}

Здесь обе загрузки идут параллельно, а coroutineScope не завершится, пока не завершатся обе. Упала loadStats: исключение не ждёт await(), оно тут же отменяет loadUser и поднимается из coroutineScope наверх. Это и есть structured concurrency: у каждой корутины есть родитель, потеряться незамеченной она не может, а отмена родителя, например когда клиент закрыл соединение, гасит всё дерево.

Правило «упала одна, отменяются остальные» принадлежит coroutineScope, а не корутинам вообще. Пять независимых выгрузок, где падение одной не должна рушить соседей, запускают в supervisorScope: там ошибка ребёнка не идёт вверх и не гасит братьев, а всплывает только в await() конкретной задачи или в обработчик. Выбор между ними это ответ на вопрос «эти задачи связаны или независимы», и его стоит задать до написания кода.

Ограничение по времени тоже встроено в модель: withTimeout(5_000) { loadStats(id) } отменяет блок по истечении срока и бросает TimeoutCancellationException. Вариант withTimeoutOrNull вместо исключения возвращает null, что удобнее для необязательных данных.

Отмена по согласию

Отмена в корутинах не прерывает выполнение силой. Она взводит флаг, который корутина замечает в ближайшей точке приостановки: любой delay, любой suspend-вызов из библиотеки проверяет флаг и бросает CancellationException. Поэтому плотный вычислительный цикл без единого suspend-вызова не отменится никогда. В таком цикле проверяют isActive сами или зовут ensureActive().

Самая частая ошибка в корутинах растёт отсюда. CancellationException наследует Exception, и такой код её проглатывает:

try {
    loadStats(id)
} catch (e: Exception) {
    log.warn("не удалось загрузить статистику", e)
    Stats.EMPTY
}

Корутину отменили, исключение поймано и залогировано как ошибка загрузки, а корутина живёт дальше, будто ничего не было, и следующий suspend-вызов бросит его снова. Правило одно: CancellationException пробрасывают, всегда. Либо ловят конкретные исключения, либо в общем catch первой строкой пишут if (e is CancellationException) throw e. Очистка после отмены, закрыть файл, снять блокировку, делается в finally, а если ей самой нужен suspend-вызов, его заворачивают в withContext(NonCancellable), иначе он не выполнится: корутина уже отменена.

Диспетчеры: цифры и одна ловушка

Диспетчер задаёт, на каких потоках исполняется корутина, и тут важны конкретные числа. Dispatchers.Default для вычислений: потоков в нём столько, сколько у машины ядер, но не меньше двух. Dispatchers.IO для блокирующего ввода-вывода: 64 потока или число ядер, смотря что больше. Потоки они берут из общего пула, поэтому «перевести на IO» не значит «завести отдельный мир»: забьёте IO блокирующими вызовами, станет тесно и вычислениям.

Своя ограниченная доля задаётся одной строкой, Dispatchers.IO.limitedParallelism(10), когда к капризному соседнему сервису нельзя ходить больше чем в десять потоков. Смена диспетчера тоже одной функцией: withContext(Dispatchers.IO) { jdbcRepository.find(id) }.

Ловушка в том, что от переноса на IO блокирующий вызов блокирующим быть не перестаёт. JDBC держит поток на всё время запроса к базе, и Dispatchers.IO просто выделяет для этого один из 64 потоков. Корутины здесь ничего не экономят: 64 одновременных запроса к базе займут все 64 потока, шестьдесят пятый будет ждать. Выигрыш корутин появляется только там, где ожидание действительно suspend: неблокирующий HTTP-клиент, R2DBC, брокер с асинхронным клиентом. Для сервиса на Spring MVC с JDBC корутины дают удобную модель ошибок и отмены, но не дешёвое ожидание; его на таком стеке дают virtual threads.

Сам диспетчер при этом не отдельная настройка, а один из элементов контекста корутины (CoroutineContext): набора пар «ключ — значение», где кроме диспетчера лежат Job запущенной корутины, её имя для логов (CoroutineName) и обработчик необработанных ошибок.

Контекст наследуется от родителя: launch берёт родительский набор и подменяет в нём только то, что вы передали аргументом, а собственный Job заводит всегда — на нём и держится дерево отмены. withContext(Dispatchers.IO) делает то же самое, но на время блока и без нового уровня в дереве задач. Отсюда практическое следствие: launch(Dispatchers.IO + CoroutineName("import")) — это не магия оператора сложения, а сборка контекста из элементов, и в логе такая корутина будет подписана именем.

GlobalScope в этой картине почти всегда ошибка: корутина без родителя, та самая утечка, против которой придумана structured concurrency. Фоновую задачу, которая должна пережить запрос, запускают в scope с известным временем жизни, например привязанном к приложению.

Flow: поток значений с приостановкой

Flow это последовательность значений, которые приходят со временем, и каждый шаг в ней умеет приостанавливаться. Он холодный: код внутри flow { } не выполняется, пока кто-то не вызвал collect, и выполняется заново для каждого собирающего.

fun orders(userId: String): Flow<Order> = flow {
    var page = 0
    while (true) {
        val batch = api.orders(userId, page++)
        if (batch.isEmpty()) break
        batch.forEach { emit(it) }
    }
}

suspend fun total(userId: String): Money =
    orders(userId).filter { it.paid }.map { it.amount }.fold(Money.ZERO) { a, b -> a + b }

Управление скоростью здесь бесплатное: emit приостанавливается, пока собирающий не обработал предыдущее значение, поэтому быстрый источник не заваливает медленного потребителя буфером. Если буфер нужен намеренно, его объявляют явно, buffer(64), и тогда видно, где он есть.

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

Корутины и Spring

В WebFlux suspend-функции в контроллерах поддерживаются давно и работают естественно: обработчик приостанавливается на неблокирующем вызове, поток цикла свободен. В Spring MVC suspend-контроллеры поддерживаются с Spring Framework 6.1: запрос обрабатывается через асинхронную обработку Servlet, и тут стоит помнить ловушку из прошлого раздела: JDBC внутри остаётся блокирующим.

Главная ловушка встречи со Spring это @Transactional. Транзакция в Spring привязана к потоку: соединение лежит в ThreadLocal, и commit вызывается в том же потоке, где был begin. Корутина же между двумя suspend-вызовами может переехать на другой поток, и вторая половина метода окажется вне транзакции, а первая половина зафиксируется отдельно.

С блокирующим PlatformTransactionManager аннотация на suspend-функции не работает так, как ожидают; честно поддержаны только реактивные менеджеры транзакций поверх R2DBC. На стеке с JDBC транзакционный код оставляют в обычной, не suspend, функции сервиса и зовут её из корутины через withContext(Dispatchers.IO): внутри одного вызова поток не меняется, и транзакция цела.

В тестах вход в корутины делают через runTest из kotlinx-coroutines-test, а не через runBlocking: он пропускает delay без ожидания и ловит незавершённые корутины.

Глубже: virtual threads: тот же выигрыш без нового синтаксисарасширенное

Java 21 решает ту же задачу с другой стороны. Virtual threads это потоки, которыми управляет JVM, а не операционная система: блокирующий вызов на виртуальном потоке не держит системный поток, JVM отстёгивает виртуальный от несущего сама. Код остаётся обычным блокирующим, JDBC остаётся JDBC, и именно поэтому на Spring MVC выигрыш достаётся почти бесплатно. Одна оговорка про Java 21–23: блокировка внутри synchronized носитель не отпускала, и старые драйверы с synchronized вокруг сетевого вызова съедали выигрыш; починено в Java 24 (JEP 491).

Честный вывод: если цель только «не держать потоки на ожидании» в Spring-приложении с JDBC, virtual threads дают это без изменения кода, а корутины не дают вовсе. Корутины выигрывают там, где нужна модель целиком: structured concurrency, отмена, Flow, Android, неблокирующий стек. В Kotlin-проекте на Spring Boot оба пути рабочие, и они не исключают друг друга: виртуальные потоки под Tomcat и корутины там, где нужны Flow и отмена.

Глубже: диагностика: почему поток занят и где корутинарасширенное

Обычный thread dump корутин не видит: в нём только потоки диспетчера с кадрами библиотеки, а какая корутина на каком шаге, не написано. Первый инструмент это флаг -Dkotlinx.coroutines.debug: имя корутины дописывается к имени потока в логах и стектрейсах, а трейс из продолжения библиотека достраивает кадрами точек приостановки. Второй инструмент, зависимость kotlinx-coroutines-debug и DebugProbes.dumpCoroutines(): дамп всех живых корутин с состоянием и местом приостановки, аналог дампа потоков для этого мира. Именно им находят корутину, которая заняла поток Dispatchers.Default блокирующим вызовом.

Коротко

• suspend это конечный автомат: состояние в Continuation, поток свободен на ожидании, стектрейс из продолжения начинается с середины.
• launch отдаёт ошибку родителю, async хранит её до await(); в coroutineScope падение одной задачи отменяет остальные, в supervisorScope нет.
• Отмена это флаг в точках приостановки; CancellationException пробрасывают всегда, очистку с suspend делают в withContext(NonCancellable).
• Default = число ядер, IO = 64 или число ядер; блокирующий JDBC на IO остаётся блокирующим, дешёвого ожидания корутины на нём не дают.
• Flow холодный и приостанавливается на emit; Sequence блокирует поток.
• @Transactional привязан к потоку: с JDBC транзакцию держат в не-suspend функции внутри одного withContext.
• Virtual threads дают дешёвое ожидание блокирующему коду без правок; корутины берут ради модели: scope, отмена, Flow.

Что почитать дальше

• Виртуальные потоки — носитель, стек в куче и что изменилось в Java 24.
• Spring MVC, WebFlux и виртуальные потоки — та же развилка со стороны веб-стека.
• Структурная конкурентность в Java — StructuredTaskScope и его нынешний статус.
• Kotlin и Spring Boot — последний шаг маршрута.