Виртуальные потоки

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

Первый виртуальный поток

Виртуальный поток создают построителем Thread.ofVirtual(). В остальном всё знакомо: start, join:

void main() throws InterruptedException {
    Thread thread = Thread.ofVirtual().start(() -> {
        IO.println("Виртуальный: " + Thread.currentThread().isVirtual());
    });
 
    thread.join();
    IO.println("Главный поток виртуальный: " + Thread.currentThread().isVirtual());
}

Результат:

Виртуальный: true
Главный поток виртуальный: false

Ещё короче — Thread.startVirtualThread(задача): он создаёт и сразу запускает виртуальный поток.

Как это устроено

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

Поток на каждую задачу

Пул для виртуальных потоков не нужен: они дешёвые, поэтому на каждую задачу создают новый поток. Это делает Executors.newVirtualThreadPerTaskExecutor(). Запустим 10 000 задач, каждая из которых «ждёт ответа» секунду:

void main() {
    Instant start = Instant.now();
 
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int i = 0; i < 10_000; i++) {
            executor.submit(() -> {
                Thread.sleep(Duration.ofSeconds(1));
                return null;
            });
        }
    }
 
    Duration elapsed = Duration.between(start, Instant.now());
    IO.println("Заняло примерно " + elapsed.toSeconds() + " с");
}

Результат (время зависит от компьютера):

Заняло примерно 1 с

Все 10 000 ожиданий шли одновременно, и программа уложилась примерно в секунду. С пулом из 100 платформенных потоков та же работа заняла бы 100 секунд. Лямбда возвращает null, чтобы считаться Callable: тогда внутри можно вызывать Thread.sleep без try-catch.

Результаты задач

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

String fetchPrice(String product) throws InterruptedException {
    Thread.sleep(Duration.ofMillis(300));
    return product + ": $" + product.length() * 3;
}
 
void main() throws Exception {
    List<String> products = List.of("чай", "кофе", "какао");
 
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        List<Future<String>> futures = new ArrayList<>();
        for (String p : products) {
            futures.add(executor.submit(() -> fetchPrice(p)));
        }
        for (Future<String> f : futures) {
            IO.println(f.get());
        }
    }
}

Результат:

чай: $9
кофе: $12
какао: $15

Когда они помогают

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

Частая ошибка

Виртуальные потоки не нужно складывать в пул фиксированного размера, чтобы ограничить нагрузку, — это лишает их смысла. Если нужно, чтобы, например, к базе данных одновременно обращались не больше трёх задач, используют Semaphore — счётчик разрешений:

void main() {
    Semaphore permits = new Semaphore(3);
    AtomicInteger active = new AtomicInteger();
    AtomicInteger maxActive = new AtomicInteger();
 
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int i = 0; i < 20; i++) {
            executor.submit(() -> {
                permits.acquire();
                try {
                    maxActive.accumulateAndGet(active.incrementAndGet(), Math::max);
                    Thread.sleep(50);
                    active.decrementAndGet();
                } finally {
                    permits.release();
                }
                return null;
            });
        }
    }
 
    IO.println("Одновременно работало не больше 3: " + (maxActive.get() <= 3));
}

Результат:

Одновременно работало не больше 3: true

acquire() берёт разрешение или ждёт, пока оно освободится, а release() возвращает его. Задач создано 20, а к «базе» одновременно ходят не больше трёх. AtomicInteger — потокобезопасный счётчик, о нём урок о синхронизации.