Виртуальные потоки: тысячи задач без страха

Что изменилось в Java 21

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

Виртуальные потоки (стандарт с Java 21) — это потоки, которыми управляет сама JVM. Они дёшевы: их можно создать сотни тысяч. Когда виртуальный поток блокируется на вводе-выводе, JVM снимает его с потока-носителя и ставит туда другой. Код при этом остаётся обычным блокирующим кодом — без коллбэков и реактивных цепочек.

Как их запускать

import java.util.concurrent.Executors;
import java.time.Duration;

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 1; i <= 10_000; i++) {
        int number = i;
        executor.submit(() -> {
            Thread.sleep(Duration.ofSeconds(1));   // имитация ожидания сети
            return number;
        });
    }
}   // close() ждёт завершения всех задач
System.out.println("Готово");

Десять тысяч задач по секунде выполняются примерно за секунду, а не за час: они ждут одновременно. С обычным пулом из 200 потоков это заняло бы около пятидесяти секунд.

Executor реализует AutoCloseable, поэтому try-with-resources сам дожидается завершения задач. Отдельный поток можно запустить и напрямую:

Thread worker = Thread.ofVirtual().name("loader").start(() -> {
    System.out.println("Работаю в " + Thread.currentThread());
});
worker.join();
// VirtualThread[#21,loader]/runnable@ForkJoinPool-1-worker-1

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

С виртуальными потоками пул больше не нужен и вреден. Пул существовал, чтобы переиспользовать дорогой ресурс; виртуальный поток дешёвый, и его создают под каждую задачу. Не пытайтесь ограничивать параллелизм размером пула — для этого есть семафор:

var limit = new java.util.concurrent.Semaphore(20);   // не больше 20 запросов к API одновременно

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (var url : urls) {
        executor.submit(() -> {
            limit.acquire();
            try {
                return fetch(url);
            } finally {
                limit.release();
            }
        });
    }
}

Где выигрыша не будет

• Вычисления. Если задача считает, а не ждёт, узкое место — процессор. Виртуальные потоки не добавят ядер; здесь по-прежнему нужен пул размером с число ядер.
• Блокировки в synchronized. В Java 21 блок synchronized, внутри которого происходит ожидание, «прикалывает» виртуальный поток к носителю, и выигрыш теряется. Замена — ReentrantLock. В более поздних версиях JDK это ограничение снято, но код, который должен работать и на 21-й, лучше писать с явным замком.
• ThreadLocal с тяжёлыми объектами. Раньше их было мало — по числу потоков в пуле. Теперь потоков сотни тысяч, и такой кеш съест память.

Структурированная параллельность

Типичная задача: запросить две системы и дождаться обеих, а при ошибке одной — отменить вторую. Раньше это писали руками через Future и try/finally. В Java 21 появился предварительный API StructuredTaskScope, который связывает задачи в область видимости:

import java.util.concurrent.StructuredTaskScope;   // в Java 21 — именно этот пакет

// Предварительная возможность: требует --enable-preview при компиляции и запуске
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    var user = scope.fork(() -> loadUser(id));
    var orders = scope.fork(() -> loadOrders(id));

    scope.join();
    scope.throwIfFailed();

    return new Dashboard(user.get(), orders.get());
}

Если любая подзадача упала, остальные отменяются, а исключение приходит вызывающему коду. Обратите внимание, насколько быстро менялся этот предварительный API: в JDK 19 и 20 класс лежал в модуле jdk.incubator.concurrent, а fork возвращал Future с методом resultNow(); в JDK 21 пакет стал java.util.concurrent, а fork возвращает Subtask с методом get(). Пример, скопированный из статьи про JDK 20, на 21-й просто не соберётся — поэтому в производственном коде эту возможность пока лучше не использовать. Обычные виртуальные потоки — окончательный стандарт и готовы к работе.

Пример: параллельная загрузка страниц

import java.net.URI;
import java.net.http.*;
import java.util.concurrent.Executors;

record Result(String url, int status, int length) {}

var client = HttpClient.newHttpClient();
var urls = java.util.List.of(
        "https://school-fortuna.ru/",
        "https://school-fortuna.ru/uroki/",
        "https://school-fortuna.ru/robots.txt");

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    var tasks = urls.stream()
            .map(url -> executor.submit(() -> {
                var request = HttpRequest.newBuilder(URI.create(url)).GET().build();
                var response = client.send(request, HttpResponse.BodyHandlers.ofString());
                return new Result(url, response.statusCode(), response.body().length());
            }))
            .toList();

    for (var task : tasks) {
        var r = task.get();
        System.out.println(r.status() + "  " + r.length() + "\t" + r.url());
    }
}

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

Что дальше

Вы прошли одиннадцать уроков: от первой команды в JShell до параллельной загрузки данных. Дальше стоит взять небольшой собственный проект — консольную утилиту, разбор выгрузки, клиент к открытому API — и применить всё вместе: записи для данных, sealed-интерфейсы для исходов, конвейеры для обработки, java.time для дат и виртуальные потоки для ожидания. Материалы уроков останутся открытыми и будут дополняться.