Асинхронность в Java

ОБЯЗАТЕЛЬНОДЛЯ НОВИЧКОВ

Разработчику Архитектору

Асинхронность в Java

Асинхронность — "запустил задачу и не жду у стены, пока она закончится". В Java для этого есть несколько уровней — от тяжёлых потоков ОС до лёгких virtual threads (Java 21+).

Как устроены virtual threads на уровне JVM — отдельная глава. Практическая шпаргалка API — справочник.

Эта статья — практический выбор инструмента, не дублирование JVM и потоков (там память, synchronized, GC).

Ваша задача	С чего начать
Параллельно посчитать отчёт	`ExecutorService` с ограниченным пулом
Несколько HTTP/БД вызовов в одном запросе	`CompletableFuture` или virtual threads
Миллионы одновременных блокирующих I/O	Virtual threads (Java 21+)
Весь стек reactive end-to-end	WebFlux — Spring

На Kotlin те же идеи проще выражаются — корутины.

Что такое "конкурентность" и "параллелизм"

Конкурентность — программа умеет вести несколько задач одновременно на уровне управления.
Параллелизм — задачи реально исполняются одновременно на разных ядрах.

Практический вывод: I/O-сервисы чаще упираются в конкурентность и ожидание внешних систем, а вычислительные задачи — в реальный параллелизм CPU.

Thread и Runnable

Поток ОС — тяжёлый ресурс (стек, планировщик). Создание "на каждый запрос" не масштабируется.

Thread t = new Thread(() -> {
    System.out.println(Thread.currentThread().getName());
});
t.start();
t.join();

Разбор:

new Thread(() -> { ... }) создаёт отдельный platform-thread и получает Runnable в виде лямбды.
Thread.currentThread().getName() показывает имя текущего потока, что удобно для первичной диагностики конкурентного кода.
t.start() запускает поток асинхронно; код внутри run выполняется параллельно с вызывающим потоком.
t.join() блокирует текущий поток до завершения t, чтобы корректно дождаться результата/побочного эффекта.
Паттерн полезен как базовый пример, но в production чаще используют пулы (ExecutorService), а не ручное создание потоков.

Интерфейс Runnable (или Callable<V> с результатом) описывает задачу; поток её выполняет.

CPU-bound и I/O-bound нагрузки

Тип задачи	Примеры	Что обычно выбирать
CPU-bound	сжатие, шифрование, расчет отчетов	ограниченный пул platform threads
I/O-bound	HTTP, JDBC, чтение файлов	virtual threads или `CompletableFuture`

Разделение помогает не смешивать тяжелые вычисления и медленное ожидание внешних сервисов в одном пуле.

ExecutorService

Пул переиспользует ограниченное число потоков для множества задач:

ExecutorService pool = Executors.newFixedThreadPool(4);
Future<Integer> future = pool.submit(() -> compute());
Integer result = future.get(5, TimeUnit.SECONDS);
pool.shutdown();

Разбор:

Executors.newFixedThreadPool(4) создаёт пул фиксированного размера: одновременно работают до четырёх задач.
submit(() -> compute()) отправляет Callable в очередь пула и возвращает Future<Integer> для управления выполнением.
future.get(5, TimeUnit.SECONDS) ждёт результат максимум 5 секунд и выбрасывает timeout-исключение при превышении дедлайна.
pool.shutdown() инициирует "мягкое" завершение пула: новые задачи больше не принимаются, текущие завершаются.
Такой шаблон применяют для CPU-задач с контролируемой степенью параллелизма.

Фабрика	Поведение
`newFixedThreadPool(n)`	Ровно n рабочих потоков, очередь задач
`newCachedThreadPool()`	Потоки по требованию, риск раздувания при всплеске
`newSingleThreadExecutor()`	Одна очередь, порядок FIFO

В production предпочтительно явно задавать размер пула и политику очереди (ThreadPoolExecutor), а не безлимитный cached.

ThreadPoolExecutor с явными параметрами

ThreadPoolExecutor pool = new ThreadPoolExecutor(
    8,                       // corePoolSize
    8,                       // maxPoolSize
    0L, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<>(1000),
    new ThreadPoolExecutor.CallerRunsPolicy()
);

Разбор:

corePoolSize и maxPoolSize равны 8, поэтому пул работает как фиксированный и не растёт выше 8 потоков.
ArrayBlockingQueue<>(1000) ограничивает очередь задач, защищая сервис от бесконтрольного накопления backlog.
CallerRunsPolicy() при переполнении очереди выполняет задачу в потоке отправителя, создавая естественный backpressure.
keepAliveTime = 0 при фиксированном пуле практически не влияет, так как число потоков постоянно.
Явная конфигурация пула делает поведение системы предсказуемым под пиковыми нагрузками.

Такой вариант лучше контролирует поведение под пиками нагрузки, чем "магические" дефолты.

Политики отказа в ThreadPoolExecutor

Когда очередь пула заполнена, срабатывает RejectedExecutionHandler.

Коротко о терминах:

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

Политика	Что происходит
`AbortPolicy`	Бросает `RejectedExecutionException`
`CallerRunsPolicy`	Выполняет задачу в вызывающем потоке (backpressure)
`DiscardPolicy`	Тихо отбрасывает задачу
`DiscardOldestPolicy`	Выбрасывает самую старую из очереди и пытается добавить новую

Для API и интеграций часто начинают с такого подхода:

CallerRunsPolicy для мягкого замедления входящего потока;
AbortPolicy для явного отказа с метрикой, алертом и понятным логом.

CompletableFuture

Композиция асинхронных шагов без ручного wait/notify:

Код ITЗагрузка примера кода…

Разбор:

supplyAsync(..., ioPool) запускает I/O-операции параллельно в выделенном пуле, не блокируя основной поток обработки запроса.
thenCombine(ordersFuture, Profile::new) объединяет результаты двух независимых futures в один объект Profile.
orTimeout(3, TimeUnit.SECONDS) ограничивает максимальное время ожидания и защищает API от вечного зависания на внешнем сервисе.
exceptionally(ex -> Profile.empty()) задаёт fallback, чтобы вернуть деградированный, но валидный ответ.
join() завершает цепочку и получает итоговый результат; исключения здесь приходят как unchecked (CompletionException).

Полезные методы:

thenApply / thenAccept / thenRun — продолжение с результатом / без / void;
thenCompose — когда следующий шаг тоже возвращает CompletableFuture;
allOf / anyOf — ожидание нескольких задач;
completeOnTimeout, orTimeout — дедлайны.

Обработка ошибок: необработанное исключение завершает цепочку; используйте exceptionally, handle или whenComplete.

future.handle((result, ex) -> {
    if (ex != null) {
        log.error("failed", ex);
        return fallback;
    }
    return result;
});

Разбор:

handle((result, ex) -> ...) вызывается всегда: и при успехе, и при ошибке.
Проверка ex != null отделяет аварийную ветку и позволяет централизованно залогировать исходную причину.
Возврат fallback превращает ошибку в управляемый результат и сохраняет работоспособность верхнего уровня.
В ветке успеха возвращается исходный result, то есть pipeline остаётся прозрачным.
Этот метод удобен, когда требуется унифицированно завершить цепочку одним типом результата.

Не вызывайте join()/get() на UI-потоке и не блокируйте event loop без причины.

Практический шаблон для внешних вызовов

Если запрос собирает данные из нескольких сервисов, полезен каркас:

Запускать вызовы параллельно.
Ставить таймаут на каждый источник.
Возвращать частичный результат или fallback, если один источник недоступен.

Этот шаблон повышает устойчивость API во время деградации внешних систем.

Virtual threads (Java 21+)

Виртуальный поток — лёгкий поток JVM, монтируемый на небольшой пул carrier (platform threads). Подходит для массового блокирующего I/O (HTTP-клиент, JDBC, файлы). Схема mount/unmount и слои "программа → JVM → ОС" — в Virtual Threads (Java 21+).

try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    IntStream.range(0, 10_000).forEach(i ->
        executor.submit(() -> handleRequest(i))
    );
}

Разбор:

newVirtualThreadPerTaskExecutor() создаёт executor, где каждая задача получает собственный virtual thread.
IntStream.range(0, 10_000) имитирует массовую нагрузку: 10 тысяч независимых задач.
executor.submit(() -> handleRequest(i)) запускает каждую задачу как блокирующую I/O-операцию в лёгком виртуальном потоке.
try-with-resources автоматически закрывает executor, дожидаясь корректного завершения запущенных задач.
Подход удобен для большого числа одновременно ожидающих операций (сеть, БД, файловая система).

Или напрямую:

Thread.startVirtualThread(() -> process(socket));

Разбор:

Thread.startVirtualThread(...) — краткий API для запуска одной задачи в виртуальном потоке без явного executor.
Лямбда () -> process(socket) обычно содержит блокирующую обработку одного подключения/запроса.
Виртуальный поток снимает часть затрат на управление множеством platform threads.
Такой вызов хорошо подходит для простых "task-per-request" сценариев.

Модель	Когда
Platform threads + фиксированный пул	CPU-bound, предсказуемая нагрузка, legacy
Virtual threads	Много одновременных блокирующих I/O операций
`CompletableFuture` + явный пул	Композиция шагов, интеграция с async API
Reactive (WebFlux)	End-to-end неблокирующий стек, backpressure

Virtual threads не ускоряют CPU-bound задачи: для вычислений по-прежнему нужен ограниченный пул platform threads или ForkJoinPool.

Structured concurrency группировка задач

В Java 21+ удобно объединять связанные подзадачи так, чтобы их жизненный цикл контролировался вместе. Это упрощает отмену "дочерних" задач при ошибке одной из них и делает код обслуживания запросов более предсказуемым.

StructuredTaskScope в Java 21/22 относится к preview API. Используйте его в production, только если в проекте разрешены preview-функции языка.

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    var profile = scope.fork(() -> loadProfile(userId));
    var orders = scope.fork(() -> loadOrders(userId));

    scope.join();           // ждать завершения задач
    scope.throwIfFailed();  // если одна упала, ошибка поднимется здесь

    return new UserView(profile.get(), orders.get());
}

Что важно для новичка:

Если одна подзадача упала, ShutdownOnFailure отменяет остальные.
Дочерние задачи живут внутри одного scope и завершаются вместе.
Код сервиса проще сопровождать, когда жизненный цикл задач собран в одном месте.
Для базового понимания сначала изучите Virtual Threads, затем вернитесь к этому блоку.

synchronized, volatile, Lock

Кратко (подробнее в JVM и потоки; демон-потоки и shutdown hook — там же):

synchronized — взаимное исключение и видимость для блока/метода;
volatile — видимость записи без атомарности составных операций;
ReentrantLock, ReadWriteLock — гибче, с tryLock и таймаутами;
java.util.concurrent.atomic — счётчики и ссылки без блокировок.

Асинхронность не отменяет необходимость синхронизации общего изменяемого состояния.

Отмена задач и graceful shutdown

Практика для серверов и фоновых обработчиков:

хранить ссылки на Future для управляемой отмены;
на завершении приложения вызывать shutdown() и ожидать awaitTermination;
при таймауте переходить к shutdownNow() и корректно логировать незавершенные задачи.

Это важно в контейнерах и при rolling update, когда процессу дают ограниченное время на остановку.

parallelStream и async

parallelStream() делит данные между потоками ForkJoinPool — не путать с асинхронным I/O. См. Stream API.

Практикум — параллелизм в Java

Демо ParallelismDemo проходит по цепочке тем из раздела Практикум: потоки, гонка данных, синхронизация, Runnable, пул.

Код ITЗагрузка примера кода…

Гонка данных без синхронизации

Операция sharedCounter++ — это прочитать → увеличить → записать. Два потока могут прочитать одно значение и оба записать N+1 вместо N+2. Десять потоков по 1000 инкрементов часто дают счётчик меньше 10 000:

sharedCounter = 0;
for (int i = 0; i < 10; i++) {
    new Thread(() -> {
        for (int j = 0; j < 1000; j++) sharedCounter++;
    }).start();
}

Исправление — synchronized, AtomicInteger или потокобезопасные структуры — см. JVM и потоки.

Пул потоков ExecutorService

new Thread на каждую задачу дорог по памяти. ExecutorService держит фиксированное число рабочих потоков:

ExecutorService executor = Executors.newFixedThreadPool(3);
for (int i = 1; i <= 6; i++) {
    final int taskId = i;
    executor.submit(() -> System.out.println("Задача " + taskId));
}
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);

submit — поставить задачу в очередь пула;
shutdown — новые задачи не принимаются;
awaitTermination — дождаться завершения уже запущенных.

Сравнение с Python, C#, Go — практикум раздела "Асинхронность".

Практические правила

Разделяйте пулы для CPU и для I/O.
Всегда задавайте таймауты на внешние вызовы.
Закрывайте ExecutorService (shutdown / shutdownNow) при остановке приложения.
Логируйте необработанные исключения в whenComplete.
Для Spring Boot 3.2+ рассмотрите virtual threads через spring.threads.virtual.enabled=true на совместимых стеках.
Для разных внешних систем делайте отдельные пулы по принципу bulkhead, чтобы деградация одного канала не занимала все рабочие потоки.

Наблюдаемость и диагностика

Асинхронный код проще сопровождать, когда заранее продуманы:

correlation id в логах;
метрики времени внешних вызовов и размера очередей;
отдельные алерты на таймауты и отклоненные задачи.

Связанный раздел: диагностика JVM.

Частые ошибки

Ошибка	Последствие
Новый `Thread` на каждый запрос	Исчерпание памяти и CPU
`get()` без таймаута	Зависший поток навсегда
Virtual threads для CPU-bound	Не даёт выигрыша, нужен пул
Общий `ArrayList` без синхронизации	Гонки и "случайные" баги

Что попробовать

Один сервис с CompletableFuture.orTimeout(2, SECONDS).
Spring Boot 3.2+: spring.threads.virtual.enabled=true и сравнить нагрузку.
Диагностика JVM при "всё тормозит" под нагрузкой.

Дальше

Virtual Threads — устройство · JVM и потоки · Stream API · Spring · Kotlin coroutines

Основа по протоколу

Базовый разбор HTTP и HTTPS находится в отдельной статье — HTTP как основа веб-интеграций.

Практикум — последовательное и параллельное выполнение

Асинхронность в Java