Виртуальные потоки (Project Loom)
Глубокое изучение виртуальных потоков в Java 21+: архитектура, создание, структурированное параллельное выполнение, области видимости значений и практические паттерны применения
Проблема, которую решают виртуальные потоки
С момента появления Java в 1995 году модель многозадачности основывалась на так называемых потоках платформы (platform threads). Каждый поток платформы напрямую соответствовал операционной системе — это означало, что для каждого Java-потока ОС создавала отдельный нативный поток с собственным стеком, обычно размером от 512 КБ до 1 МБ. При масштабировании до тысяч одновременных соединений этот подход приводил к катастрофическим последствиям.
Представьте типичное веб-приложение: оно обрабатывает HTTP-запросы, каждому из которых выделяется отдельный поток платформы. Если приложение обслуживает 10 000 одновременных пользователей, ему потребуется 10 000 нативных потоков. Каждый такой поток занимает минимум 512 КБ стека, что даёт уже 5 ГБ оперативной памяти только на стеки, не считая прочих структур данных потока. Но главная проблема даже не в памяти — ОС не способна эффективно переключать десятки тысяч потоков. Контекстное переключение между потоками требует времени, а планировщик ОС начинает тратить всё больше ресурсов на саму координацию, а не на полезную работу.
Эта проблема получила название «контрольная точка масштабируемости» (scalability wall). Разработчики вынуждены были использовать асинхронные фреймворки (Reactor, RxJava, Vert.x) с их сложной моделью обратных вызовов (callbacks), чтобы избежать блокировки потоков. Но асинхронный код значительно сложнее для чтения, отладки и сопровождения. Стектрейсы становились бесполезными, так как цепочка вызовов обрывалась на каждом асинхронном переходе. Отладка ночных сбоев превращалась в детективную историю.
Именно эту проблему решил Project Loom — масштабный проект OpenJDK, начатый в 2017 году Роном Пресслером (Ron Pressler) из Oracle Labs. Результатом стал механизм виртуальных потоков, который вошёл в JDK 21 как финальная, стабильная функциональность. Виртуальные потоки решают дилемму между простотой блокирующего кода и масштабируемостью асинхронного подхода.
Архитектура: платформенные vs виртуальные потоки
Для понимания виртуальных потоков необходимо разобраться в том, как устроена двухуровневая модель планирования, которую использует JVM. На нижнем уровне работает планировщик ОС, который управляет нативными потоками — теми, которые действительно выполняются на процессорных ядрах. На верхнем уровне работает планировщик виртуальных потоков в самой JVM, который мультиплексирует тысячи или даже миллионы виртуальных потоков поверх относительно небольшого пула нативных потоков платформы.
Платформенный поток (platform thread) — это привычный вам Thread из предыдущих версий Java. Он создаёт нативный поток ОС, выделяет стек значительного размера и является ресурсоёмким. Каждый такой поток имеет собственный кадр стека (stack frame), в котором хранятся локальные переменные, параметры методов и адреса возврата. Когда поток блокируется (например, при чтении из сети или ожидании ответа от базы данных), нативный поток ОС простаивает, занимая оперативную память и место в планировщике ОС, хотя ничего полезного не делает.
Виртуальный поток (virtual thread) — это совершенно иной механизм. Он не имеет привязки к нативному потоку ОС. Вместо этого виртуальный поток живёт в куче (heap) Java и управляется планировщиком виртуальных потоков, встроенным в JVM. Стек виртуального потока — это не нативный стек, а структура данных в куче, которая может быть динамически уменьшена (continuation) при блокировке и восстановлена при готовности к выполнению. Благодаря этому создание и переключение виртуальных потоков обходится в сотни раз дешевле, чем для платформенных потоков.
Ключевое отличие в том, как这两种 потоки ведут себя при блокировке. Когда платформенный поток блокируется, он занимает нативный поток ОС, и ОС вынуждена ждать, пока блокировка не снимется. Когда блокируется виртуальный поток, его контекст выполнения (continuation) снимается с нативного потока и сохраняется в куче, а нативный поток освобождается для выполнения другого виртуального потока. Это похоже на то, как coroutine в Go (goroutine) или green threads в早期 Erlang — контекст переключается на уровне приложения, а не операционной системы.
Создание виртуальных потоков
Java 21 предоставляет несколько способов создания виртуальных потоков. Рассмотрим каждый из них подробно, разбираясь в нюансах и контексте применения.
Способ 1: Thread.ofVirtual() — это самый гибкий способ создания виртуальных потоков. Фабричный метод Thread.ofVirtual() возвращает объект Thread.Builder, через который можно настроить имя потока, сделать его демон-потоком и установить обработчик необработанных исключений. Этот способ даёт максимальный контроль над создаваемым потоком.
Thread virtualThread = Thread.ofVirtual()
.name("worker-vt-", 0)
.daemon(true)
.start(() -> {
System.out.println("Выполняюсь в виртуальном потоке: "
+ Thread.currentThread());
System.out.println("Это виртуальный поток? "
+ Thread.currentThread().isVirtual());
});
virtualThread.join();
Thread namedThread = Thread.ofVirtual()
.name("my-custom-virtual-thread")
.start(() -> {
String threadName = Thread.currentThread().getName();
System.out.println("Имя виртуального потока: " + threadName);
});
namedThread.join();Обратите внимание на метод name("worker-vt-", 0). Второй аргумент задаёт начальное значение счётчика имён. Если вызвать Thread.ofVirtual().name("worker-vt-", 0) несколько раз и создать потоки, они получат имена worker-vt-0, worker-vt-1, worker-vt-2 и так далее. Это очень удобно для группировки связанных потоков и упрощает диагностику в логах и при отладке.
Способ 2: Thread.startVirtualThread() — упрощённый статический метод для случаев, когда не нужна настройка builder'ом. Этот метод создаёт и немедленно запускает виртуальный поток, работая аналогично new Thread(runnable).start(), но для виртуального потока.
Thread.startVirtualThread(() -> {
System.out.println("Быстрый запуск виртуального потока");
System.out.println("Поток: " + Thread.currentThread());
});Способ 3: try-with-resources с Thread.Builder — этот подход позволяет автоматически присоединить (join) виртуальный поток при завершении блока try. Это удобно, когда нужно гарантировать завершение потока перед продолжением выполнения.
try (var executor = Thread.ofVirtual().name("scoped-vt-", 0).factory().newThread(() -> {
System.out.println("Работаю в виртуальном потоке");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Завершаю работу");
})) {
executor.start();
executor.join();
}
System.out.println("Поток гарантированно завершён");ExecutorService с виртуальными потоками
Один из наиболее мощных паттернов использования виртуальных потоков — замена традиционных пулов потоков (FixedThreadPool, CachedThreadPool) на ExecutorService, работающий с виртуальными потоками. В отличие от traditional пулов, где размер ограничивается количеством нативных потоков (обычно числом ядер процессора + 1), пул виртуальных потоков может создавать сколько угодно потоков без ущерба для производительности.
Метод Executors.newVirtualThreadPerTaskExecutor() создаёт исполнителя, который для каждой поступающей задачи создаёт отдельный виртуальный поток. Это означает, что вам больше не нужно выбирать размер пула, беспокоиться о блокировке потоков или деградации производительности при большом количестве задач. Каждая задача получает свой поток, а JVM берёт на себя управление мультиплексированием.
import java.util.concurrent.*;
public class VirtualThreadExecutorDemo {
public static void main(String[] args) throws Exception {
long start = System.currentTimeMillis();
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
IntStream.range(0, 100_000).forEach(i -> {
executor.submit(() -> {
try {
Thread.sleep(Duration.ofSeconds(1));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return i;
});
});
}
long elapsed = System.currentTimeMillis() - start;
System.out.println("100 000 задач выполнены за " + elapsed + " мс");
start = System.currentTimeMillis();
try (var executor = Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors())) {
IntStream.range(0, 100_000).forEach(i -> {
executor.submit(() -> {
try {
Thread.sleep(Duration.ofSeconds(1));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return i;
});
});
}
elapsed = System.currentTimeMillis() - start;
System.out.println("100 000 задач (fixed pool) за " + elapsed + " мс");
}
}В приведённом примере виртуальные потоки завершают 100 000 задач за ~1-2 секунды, потому что все задачи спят одновременно. FixedThreadPool с 8 потоками выполняет те же задачи за ~12 500 мс, потому что одновременно обрабатываются только 8 задач. Разница в масштабе — это и есть то, ради чего были созданы виртуальные потоки.
Важно понимать, что newVirtualThreadPerTaskExecutor() создаёт неограниченное количество виртуальных потоков. В отличие от traditional пулов, здесь нет верхнего предела — если вы отправите миллион задач, будет создано миллион виртуальных потоков. Это безопасно с точки зрения памяти и производительности, но вы должны убедиться, что ваш код не создаёт неограниченного потребления в других ресурсах (открытые файловые дескрипторы, соединения с базой данных и т.д.).
Если вам нужен ExecutorService с виртуальными потоками, но с ограничением параллелизма, вы можете создать его через Thread.ofVirtual().factory() в сочетании с собственным механизмом семафора, либо использовать Semaphore для ограничения количества одновременно выполняемых задач.
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class BoundedVirtualExecutor {
public static void main(String[] args) throws Exception {
int maxConcurrent = 50;
Semaphore semaphore = new Semaphore(maxConcurrent);
AtomicInteger activeTasks = new AtomicInteger(0);
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < 1000; i++) {
int taskId = i;
executor.submit(() -> {
semaphore.acquire();
try {
int active = activeTasks.incrementAndGet();
System.out.println("Задача " + taskId
+ " (активных: " + active + ")");
Thread.sleep(500);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
activeTasks.decrementAndGet();
semaphore.release();
}
});
}
}
System.out.println("Все задачи завершены");
}
}Сравнение: платформенные потоки vs виртуальные потоки
Чтобы наглядно увидеть разницу между двумя подходами, давайте рассмотрим практический пример. Создадим программу, которая имитирует обработку HTTP-запросов — каждый запрос требует обращения к внешнему сервису, которое занимает около 100 мс. Мы сравним традиционный подход с фиксированным пулом потоков и подход с виртуальными потоками.
import java.util.concurrent.*;
import java.util.stream.*;
public class PlatformVsVirtual {
static String simulateHttpRequest(int id) throws Exception {
Thread.sleep(100);
return "Response-" + id;
}
public static void main(String[] args) throws Exception {
int totalRequests = 10_000;
System.out.println("=== Платформенные потоки (FixedPool) ===");
long start1 = System.nanoTime();
try (var pool = Executors.newFixedThreadPool(200)) {
var futures = IntStream.range(0, totalRequests)
.mapToObj(i -> pool.submit(() -> simulateHttpRequest(i)))
.toList();
for (var f : futures) {
f.get();
}
}
long time1 = (System.nanoTime() - start1) / 1_000_000;
System.out.println("Время: " + time1 + " мс");
System.out.println();
System.out.println("=== Виртуальные потоки ===");
long start2 = System.nanoTime();
try (var pool = Executors.newVirtualThreadPerTaskExecutor()) {
var futures = IntStream.range(0, totalRequests)
.mapToObj(i -> pool.submit(() -> simulateHttpRequest(i)))
.toList();
for (var f : futures) {
f.get();
}
}
long time2 = (System.nanoTime() - start2) / 1_000_000;
System.out.println("Время: " + time2 + " мс");
System.out.println();
System.out.println("Ускорение: " + String.format("%.1f",
(double) time1 / time2) + "x");
}
}Пул из 200 платформенных потоков обрабатывает 10 000 запросов, обрабатывая по 200 одновременно — общее время около 5 секунд. Виртуальные потоки обрабатывают те же 10 000 запросов за близкое к 100 мс (все запросы выполняются одновременно, потому что каждый виртуальный поток при блокировке освобождает нативный поток). Разница в десятки раз — и это при том, что память, потребляемая виртуальными потоками, на порядки меньше.
Память и ресурсы
Давайте измерим потребление памяти при создании большого количества потоков обоих типов. Это поможет понять, почему виртуальные потоки позволяют масштабироваться до миллионов одновременных задач.
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class MemoryComparison {
static final AtomicLong counter = new AtomicLong(0);
public static void main(String[] args) throws Exception {
int count = 100_000;
CountDownLatch latch = new CountDownLatch(count);
System.out.println("Создание " + count + " виртуальных потоков...");
Runtime runtime = Runtime.getRuntime();
long beforeUsed = runtime.totalMemory() - runtime.freeMemory();
for (int i = 0; i < count; i++) {
Thread.startVirtualThread(() -> {
counter.incrementAndGet();
latch.countDown();
});
}
latch.await();
long afterUsed = runtime.totalMemory() - runtime.freeMemory();
long virtualMem = afterUsed - beforeUsed;
System.out.println("Виртуальные потоки — память: ~"
+ (virtualMem / 1024 / 1024) + " МБ");
System.out.println("Создано потоков: " + counter.get());
System.out.println("Память на поток: ~"
+ (virtualMem / count) + " байт");
System.out.println();
System.out.println("Для сравнения:");
System.out.println("Платформенный поток: ~1 МБ стека + ~2 КБ объект");
System.out.println("100 000 платформенных потоков потребовали бы ~100 ГБ RAM");
System.out.println("Виртуальный поток: ~数百 байт в куче (continuation frame)");
}
}Результаты этого теста демонстрируют колоссальную разницу. 100 000 виртуальных потоков потребляют лишь несколько мегабайт дополнительной памяти, тогда как 100 000 платформенных потоков потребовали бы порядка 100 ГБ — больше, чем есть в большинстве серверов. Каждый виртуальный поток при создании выделяет лишь объект-обёртку в куче размером около 200-800 байт. Стек виртуального потока выделяется лениво и только когда это необходимо, а при блокировке он сжимается (yield) и может быть полностью выгружен из памяти до момента возобновления.
Pinning: главная ловушка виртуальных потоков
Одно из наиболее важных понятий, связанных с виртуальными потоками — pinning (закрепление). Pinning происходит, когда виртуальный поток оказывается «закреплён» за нативным потоком ОС и не может быть отсоединён от него. В这样的 случае виртуальный поток ведёт себя точно так же, как платформенный — блокирует нативный поток, и тот не может быть использован другим виртуальным потоком.
Pinning возникает в двух случаях: при входе в синхронизированный блок (synchronized) и при вызове нативных методов (JNI). Это не баг, а ограничение текущей реализации. JVM не может безопасно сохранить и восстановить состояние нативного метода, поэтому вынуждена удерживать нативный поток.
Чтобы обнаружить pinning, JVM предоставляет специальную системную характеристику: -Djdk.tracePinnedThreads=short (или full для подробного вывода). При запуске с этим параметром JVM будет выводить в stderr информацию о каждом случае pinning, включая стектрейс. Это незаменимый инструмент при поиске узких мест в коде, работающем с виртуальными потоками.
import java.util.concurrent.locks.ReentrantLock;
public class PinningDemo {
static final Object lock = new Object();
static final ReentrantLock reentrantLock = new ReentrantLock();
static void blockingTask(String name) {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
static void synchronizedBlock() {
synchronized (lock) {
blockingTask("synchronized");
}
}
static void reentrantLockBlock() {
reentrantLock.lock();
try {
blockingTask("ReentrantLock");
} finally {
reentrantLock.unlock();
}
}
public static void main(String[] args) throws Exception {
System.out.println("Тест с synchronized — вызывает pinning:");
long start1 = System.nanoTime();
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < 10; i++) {
executor.submit(PinningDemo::synchronizedBlock);
}
}
long time1 = (System.nanoTime() - start1) / 1_000_000;
System.out.println("Время: " + time1 + " мс");
System.out.println();
System.out.println("Тест с ReentrantLock — НЕ вызывает pinning:");
long start2 = System.nanoTime();
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < 10; i++) {
executor.submit(PinningDemo::reentrantLockBlock);
}
}
long time2 = (System.nanoTime() - start2) / 1_000_000;
System.out.println("Время: " + time2 + " мс");
}
}Запустите программу с параметром -Djdk.tracePinnedThreads=short, чтобы увидеть pinning. При использовании synchronized 10 виртуальных потоков блокируют друг друга по одному нативному потоку, и общее время близко к 2 секундам (10 × 200 мс). При использовании ReentrantLock все 10 виртуальных потоков выполняются параллельно на разных carrier threads, и время близко к 200 мс.
Как избежать pinning? Замените synchronized на ReentrantLock (или ReentrantReadWriteLock, Semaphore, и другие lock-механизмы из пакета java.util.concurrent.locks). ReentrantLock не является нативным блокирующим примитивом — он реализован на уровне Java и поддерживает сохранение и восстановление состояния continuation виртуального потока. Если вы используете библиотеки, которые внутренне используют synchronized, вам нужно либо обновить их до версий, поддерживающих виртуальные потоки, либо обернуть вызовы в отдельные потоки платформы.
Стоит отметить, что в JDK 24 планируется расширение поддержки synchronized для виртуальных потоков — в некоторых случаях pinning будет предотвращён автоматически. Но до этого момента рекомендуется активно использовать ReentrantLock в коде, предназначенном для работы с виртуальными потоками.
Обработка ошибок в виртуальных потоках
Обработка ошибок в виртуальных потоках имеет особенности, о которых необходимо знать. Поскольку виртуальные потоки управляются JVM, а не ОС, механизм uncaught exception handlers работает иначе, чем для платформенных потоков. Когда в виртуальном потоке возникает неперехваченное исключение, оно не «всплывает» на уровень ExecutorService, а вместо этого вызывает навешенный на поток обработчик (Thread.UncaughtExceptionHandler).
Это означает, что если вы используете Executors.newVirtualThreadPerTaskExecutor() и задача выбрасывает исключение, которое не перехвачено внутри задачи, исключение будет потеряно. ExecutorService не прокинет его наружу при вызове close(). Чтобы не терять ошибки, всегда оборачивайте тело задачи в try-catch.
import java.util.concurrent.*;
import java.util.logging.Logger;
public class VirtualThreadErrorHandling {
static final Logger logger = Logger.getLogger(
VirtualThreadErrorHandling.class.getName());
public static void main(String[] args) throws Exception {
System.out.println("=== Без обработки ошибок (ошибка теряется) ===");
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
executor.submit(() -> {
throw new RuntimeException("Ошибка в виртуальном потоке!");
});
}
System.out.println("Executor завершён — ошибка потеряна!");
System.out.println();
System.out.println("=== С обработкой ошибок через try-catch ===");
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
executor.submit(() -> {
try {
throw new RuntimeException("Важная ошибка!");
} catch (Exception e) {
logger.severe("Поймана ошибка: " + e.getMessage());
}
});
}
System.out.println("Executor завершён — ошибка обработана");
System.out.println();
System.out.println("=== Future для получения результата ошибки ===");
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
Future<String> future = executor.submit(() -> {
Thread.sleep(100);
if (true) {
throw new IllegalStateException("Условие не выполнено");
}
return "Успех";
});
try {
String result = future.get();
} catch (ExecutionException e) {
System.out.println("Ошибка из Future: "
+ e.getCause().getMessage());
}
}
System.out.println();
System.out.println("=== Настраиваемый обработчик необработанных исключений ===");
ThreadFactory factory = Thread.ofVirtual()
.name("safe-vt-", 0)
.uncaughtExceptionHandler((thread, throwable) -> {
logger.severe("Необработанное исключение в потоке "
+ thread.getName() + ": " + throwable.getMessage());
})
.factory();
try (var executor = Executors.newThreadPerTaskExecutor(factory)) {
executor.submit(() -> {
throw new NullPointerException("Тест NPE");
});
}
System.out.println("Готово с кастомным обработчиком");
}
}Structured Concurrency (StructuredTaskScope)
Структурированная конкурентность — это экспериментальная функциональность (в JDK 21 — preview), которая изменяет парадигму управления параллельными задачами. Ключевая идея: время жизни параллельных задач должно быть привязано к блоку кода (scope). Когда scope завершается, все задачи внутри него гарантированно завершаются — либо успешно, либо с ошибкой. Это предотвращает утечку задач и упрощает обработку ошибок.
В традиционном подходе с ExecutorService задачи могут «выжить» после завершения метода, в котором они были запущены. Если вы забыли вызвать shutdown() или задача зависла, она продолжает работать в фоне. Это источник утечек ресурсов и трудноуловимых багов. StructuredTaskScope решает эту проблему: когда вы выходите из блока try, все дочерние задачи гарантированно завершаются. JVM не даёт вам «забыть» о задачах.
Паттерн StructuredTaskScope особенно полезен при обработке запросов, где нужно выполнить несколько параллельных операций и объединить их результаты. Например, при обработке HTTP-запроса вам может потребоваться одновременно обратиться к базе данных, микросервису и кэшу, а затем собрать все три результата. Если любая из операций завершится ошибкой, вы можете отменить остальные и быстро вернуть ошибку клиенту.
import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.*;
public class StructuredConcurrencyDemo {
record User(String name, String email) {}
record Order(String orderId, double amount) {}
record UserProfile(User user, Order lastOrder) {}
static User fetchUserFromDB(int userId) throws Exception {
Thread.sleep(200);
return new User("User-" + userId, "user" + userId + "@mail.com");
}
static Order fetchLastOrder(int userId) throws Exception {
Thread.sleep(300);
return new Order("ORD-" + (userId * 100), 99.99);
}
static UserProfile loadUserProfile(int userId) throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Subtask<User> userTask = scope.fork(() -> fetchUserFromDB(userId));
Subtask<Order> orderTask = scope.fork(() -> fetchLastOrder(userId));
scope.join();
scope.throwIfFailed();
return new UserProfile(userTask.get(), orderTask.get());
}
}
public static void main(String[] args) throws Exception {
long start = System.nanoTime();
UserProfile profile = loadUserProfile(42);
long elapsed = (System.nanoTime() - start) / 1_000_000;
System.out.println("Профиль: " + profile);
System.out.println("Загружено за " + elapsed + " мс");
System.out.println("(параллельно, общее время = максимум из двух запросов)");
}
}В этом примере StructuredTaskScope.ShutdownOnFailure создаёт scope, который отменяет все оставшиеся задачи при возникновении ошибки в любой из них. Это поведение по умолчанию — самое безопасное. Если одна из операций завершится ошибкой, вторая будет отменена, и ошибка будет проброшена через throwIfFailed().
Существует также StructuredTaskScope.ShutdownOnSuccess, который, наоборот, отменяет все остальные задачи, как только одна из них успешно завершается. Это полезно, когда вы обращаетесь к нескольким репликам одного сервиса и хотите получить ответ от самого быстрого, отменив остальные запросы.
Обратите внимание на структуру кода: задачи создаются через scope.fork(), scope блокирующий join() ждёт завершения всех задач, а затем результаты извлекаются через Subtask.get(). Если какая-либо задача завершилась ошибкой, throwIfFailed() пробросит исключение. Весь код сбалансирован — для каждого fork есть соответствующий get, а блок try-with-resources гарантирует завершение scope.
Scoped Values: замена ThreadLocal
ThreadLocal — один из самых проблематичных механизмов в Java-многозадачности. Он привязывает данные к потоку, но не привязывает поток к задаче. Это значит, что данные из ThreadLocal могут «утечь» из одного запроса в другой, если пул потоков переиспользует потоки. Кроме того, ThreadLocal не работает с виртуальными потоками так же, как с платформенными — хотя технически поддерживается, семантика утекающих данных делает его небезопасным.
Scoped Values (экспериментальная функция, preview в JDK 21) решают эту проблему, привязывая значения к конкретному scope выполнения, а не к потоку. Scoped Value — это immutable, поле типа ScopedValue<T>, которое доступно только внутри блока ScopedValue.runWhere() или ScopedValue.where(). Когда scope завершается, значение автоматически удаляется, и попытка доступа к нему за пределами scope приводит к ошибке.
Scoped Values обладают свойством « inheriting» — они автоматически наследуются всеми виртуальными потоками, созданными внутри scope. Это идеально подходит для передачи контекста запроса (authentication info, request ID, tracing context) по всей иерархии вызовов без явной передачи параметров. Если виртуальный поток создаёт дочерние потоки через Thread.ofVirtual().start(), они увидят те же Scoped Values, что и родительский поток.
import jdk.incubator.concurrent.ScopedValue;
import java.util.concurrent.*;
public class ScopedValueDemo {
static final ScopedValue<String> CURRENT_USER =
ScopedValue.newInstance();
static final ScopedValue<String> REQUEST_ID =
ScopedValue.newInstance();
static void processRequest(String user, String requestId) {
ScopedValue.runWhere(CURRENT_USER, user,
() -> ScopedValue.where(REQUEST_ID, requestId,
() -> {
handleRequest();
}));
}
static void handleRequest() {
String user = CURRENT_USER.get();
String reqId = REQUEST_ID.get();
System.out.println("Обработка запроса " + reqId
+ " от пользователя " + user);
Thread.startVirtualThread(() -> {
String innerUser = CURRENT_USER.get();
String innerReqId = REQUEST_ID.get();
System.out.println(" Дочерний поток видит: user="
+ innerUser + ", reqId=" + innerReqId);
});
}
public static void main(String[] args) throws Exception {
System.out.println("=== Scoped Values с виртуальными потоками ===");
System.out.println();
processRequest("alice", "req-001");
processRequest("bob", "req-002");
System.out.println();
System.out.println("=== ThreadLocal (проблема утечки) ===");
ThreadLocal<String> threadLocalUser = new ThreadLocal<>();
try (var executor = Executors.newFixedThreadPool(1)) {
executor.submit(() -> {
threadLocalUser.set("alice");
System.out.println("Поток 1: " + threadLocalUser.get());
Thread.sleep(100);
}).get();
executor.submit(() -> {
System.out.println("Поток 1 (переиспользован): "
+ threadLocalUser.get());
threadLocalUser.remove();
}).get();
}
System.out.println();
System.out.println("Scoped Values гарантируют отсутствие утечек.");
}
}Scoped Values значительно превосходят ThreadLocal по безопасности. ThreadLocal хранит данные в самом потоке и не удаляет их при завершении задачи. Если пул потоков переиспользует поток, данные от предыдущей задачи остаются доступны. ScopedValue привязана к scope, а не к потоку — при выходе из scope значение уничтожается, и никакая утечка невозможна.
Важно отметить, что Scoped Values оптимизированы для виртуальных потоков. Они хранятся не в каждом потоке отдельно, а в специальной структуре данных, которая копируется при fork виртуального потока. Это делает их значительно быстрее, чем ThreadLocal, для сценариев с короткоживущими задачами. JVM может оптимизировать доступ к Scoped Value через JIT-компиляцию, поскольку значение гарантированно неизменяемо в рамках scope.
Практический пример: веб-скрапер на виртуальных потоках
Рассмотрим реальную задачу: написание веб-скрапера, который загружает множество страниц одновременно. Традиционный подход с фиксированным пулом потоков ограничен размером пула и требует аккуратного управления ресурсами. Подход с виртуальными потоками значительно проще и масштабируемее.
В приведённом ниже примере мы создаём скрапер, который загружает страницы с нескольких URL одновременно, извлекает заголовки и подсчитывает статистику. Мы используем HttpClient, который поддерживает виртуальные потоки начиная с Java 21, и структурированную конкурентность для управления жизненным циклом задач.
import java.net.URI;
import java.net.http.*;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class VirtualThreadScraper {
static final HttpClient client = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(5))
.build();
record PageResult(String url, int statusCode,
String title, int contentLength) {}
static PageResult fetchPage(String url) throws Exception {
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(url))
.timeout(Duration.ofSeconds(10))
.GET()
.build();
HttpResponse<String> response = client.send(request,
HttpResponse.BodyHandlers.ofString());
String title = extractTitle(response.body());
return new PageResult(url, response.statusCode(),
title, response.body().length());
}
static String extractTitle(String html) {
int start = html.indexOf("<title>");
int end = html.indexOf("</title>");
if (start != -1 && end != -1) {
return html.substring(start + 7, end).trim();
}
return "(без заголовка)";
}
public static void main(String[] args) throws Exception {
List<String> urls = List.of(
"https://example.com",
"https://httpbin.org/html",
"https://httpbin.org/status/200",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/2",
"https://www.google.com",
"https://github.com",
"https://stackoverflow.com",
"https://news.ycombinator.com",
"https://reddit.com"
);
System.out.println("Загрузка " + urls.size()
+ " страниц виртуальными потоками...");
long start = System.nanoTime();
AtomicInteger successCount = new AtomicInteger(0);
AtomicInteger failCount = new AtomicInteger(0);
try (var executor =
Executors.newVirtualThreadPerTaskExecutor()) {
List<Future<PageResult>> futures = urls.stream()
.map(url -> executor.submit(() -> {
try {
PageResult result = fetchPage(url);
successCount.incrementAndGet();
return result;
} catch (Exception e) {
failCount.incrementAndGet();
return new PageResult(url, -1,
"ОШИБКА: " + e.getMessage(), 0);
}
}))
.toList();
for (Future<PageResult> future : futures) {
PageResult result = future.get();
String status = result.statusCode() == 200
? "OK" : "ERR " + result.statusCode();
System.out.printf("[%s] %s — %s (%d bytes)%n",
status, result.url(),
result.title(), result.contentLength());
}
}
long elapsed = (System.nanoTime() - start) / 1_000_000;
System.out.println();
System.out.println("Успешно: " + successCount.get());
System.out.println("Ошибок: " + failCount.get());
System.out.println("Общее время: " + elapsed + " мс");
System.out.println("Среднее время на страницу: "
+ (elapsed / urls.size()) + " мс");
}
}Этот скрапер использует только 10 виртуальных потоков для одновременной загрузки 10 страниц. Все HTTP-запросы выполняются параллельно — общее время определяется самым медленным запросом, а не суммой всех запросов. При использовании traditional пула с, например, 5 потоками, загрузка заняла бы значительно больше времени.
Давайте расширим этот пример и добавим поддержку обработки тысяч URL, обработку ошибок с retry, логирование и сбор метрик — всё это на виртуальных потоках.
import java.net.URI;
import java.net.http.*;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.stream.*;
public class AdvancedScraper {
static final HttpClient client = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(5))
.followRedirects(HttpClient.Redirect.NORMAL)
.build();
static final AtomicInteger totalFetched = new AtomicInteger(0);
static final AtomicInteger totalErrors = new AtomicInteger(0);
static final AtomicLong totalBytes = new AtomicLong(0);
static String fetchWithRetry(String url, int maxRetries)
throws Exception {
Exception lastException = null;
for (int attempt = 1; attempt <= maxRetries; attempt++) {
try {
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(url))
.timeout(Duration.ofSeconds(10))
.GET()
.build();
HttpResponse<String> response = client.send(request,
HttpResponse.BodyHandlers.ofString());
totalFetched.incrementAndGet();
totalBytes.addAndGet(response.body().length());
return response.body();
} catch (Exception e) {
lastException = e;
totalErrors.incrementAndGet();
if (attempt < maxRetries) {
long delay = (long) Math.pow(2, attempt) * 100;
Thread.sleep(delay);
}
}
}
throw new Exception("Все попытки исчерпаны для " + url,
lastException);
}
public static void main(String[] args) throws Exception {
List<String> urls = IntStream.rangeClosed(1, 100)
.mapToObj(i -> "https://httpbin.org/delay/"
+ ThreadLocalRandom.current().nextInt(1, 4))
.toList();
System.out.println("Скрапинг " + urls.size()
+ " URL с retry...");
long start = System.nanoTime();
try (var executor =
Executors.newVirtualThreadPerTaskExecutor()) {
List<Future<Boolean>> futures = urls.stream()
.map(url -> executor.submit(() -> {
try {
String content = fetchWithRetry(url, 3);
return true;
} catch (Exception e) {
return false;
}
}))
.toList();
for (Future<Boolean> f : futures) {
f.get();
}
}
long elapsed = (System.nanoTime() - start) / 1_000_000;
System.out.println("Загружено: " + totalFetched.get() + " URL");
System.out.println("Ошибок: " + totalErrors.get());
System.out.println("Объём данных: "
+ (totalBytes.get() / 1024) + " КБ");
System.out.println("Общее время: " + elapsed + " мс");
System.out.println("Скорость: "
+ (totalFetched.get() * 1000L / elapsed) + " URL/сек");
}
}Отладка и мониторинг виртуальных потоков
Отладка виртуальных потоков требует некоторых знаний о том, как JVM представляет их в различных инструментах. Виртуальные потоки отображаются в Java Flight Recorder, JFR, и в jcmd. Поскольку виртуальные потоки управляются JVM, а не ОС, traditional инструменты уровня ОС (top, htop) не увидят их как отдельные потоки.
Для отладки в IntelliJ IDEA убедитесь, что вы используете JDK 21+ и включён experimental support для virtual threads в настройках (Settings → Build → Compiler → Java Compiler). В отладчике виртуальные потоки отображаются вместе с платформенными, и вы можете ставить breakpoints, inspect переменные и step through код точно так же, как для обычных потоков.
Для мониторинга в продакшене рекомендуется использовать JFR (Java Flight Recorder). JFR записывает события, связанные с виртуальными потоками: pinning events, carrier thread usage, continuation events. Это позволяет анализировать производительность и обнаруживать проблемы, такие как чрезмерный pinning.
import java.lang.management.*;
import java.util.concurrent.*;
public class VirtualThreadMonitoring {
public static void main(String[] args) throws Exception {
ThreadMXBean threadBean =
ManagementFactory.getThreadMXBean();
System.out.println("=== До создания виртуальных потоков ===");
System.out.println("Платформенных потоков: "
+ threadBean.getThreadCount());
int count = 10_000;
CountDownLatch latch = new CountDownLatch(count);
try (var executor =
Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < count; i++) {
executor.submit(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
latch.countDown();
});
}
Thread.sleep(500);
System.out.println();
System.out.println("=== С виртуальными потоками ===");
System.out.println("Платформенных потоков (в JVM): "
+ threadBean.getThreadCount());
System.out.println("Пик потоков: "
+ threadBean.getPeakThreadCount());
System.out.println("Всего создано: "
+ threadBean.getTotalStartedThreadCount());
System.out.println();
System.out.println("Обратите внимание: JVM видит только");
System.out.println("carrier threads (платформенные),");
System.out.println("но не 10 000 виртуальных потоков.");
System.out.println("Виртуальные потоки — это объекты в куче!");
System.out.println();
System.out.println("Некоторые виртуальные потоки:");
Thread.getAllStackTraces().keySet().stream()
.filter(Thread::isVirtual)
.limit(5)
.forEach(t -> System.out.println(" " + t));
}
}
}Советы по работе с виртуальными потоками
При работе с виртуальными потоками есть несколько важных рекомендаций, которые помогут избежать типичных ошибок и получить максимальную выгоду от этой технологии.
Совет 1: Используйте виртуальные потоки для I/O- bound задач. Виртуальные потоки идеально подходят для задач, которые большую часть времени проводят в ожидании — HTTP-запросы, чтение файлов, работа с базой данных, отправка сообщений в очередь. В таких сценариях виртуальные потоки позволяют обрабатывать огромное количество одновременных операций без пропорционального увеличения потребления ресурсов. Для CPU-bound задач (вычисления, шифрование, компиляция) виртуальные потоки не дают преимущества, потому что нативный поток не освобождается при CPU-вычислениях.
Совет 2: Избегайте пулов потоков. С виртуальными потоками традиционные пулы потоков (FixedThreadPool, CachedThreadPool) теряют смысл. Вместо Executors.newFixedThreadPool(10) используйте Executors.newVirtualThreadPerTaskExecutor(). Единственная причина оставлять traditional пул — если ваша библиотека или framework не поддерживает виртуальные потоки и требует привязки к specific thread pool (например, некоторые older версии Spring).
Совет 3: Замените synchronized на ReentrantLock. Как мы видели в разделе о pinning, synchronized блокирует carrier thread. В коде, который будет работать с виртуальными потоками, замените все synchronized блоки на ReentrantLock. Это безопасно и не меняет семантику для traditional потоков. Начните с поиска по кодовой базе: grep -rn "synchronized" src/ и оцените каждый случай.
Совет 4: Never pool виртуальные потоки. В отличие от traditional потоков, виртуальные потоки настолько дешевы в создании, что их пуллирование бессмысленно. Создавайте новый виртуальный поток для каждой задачи — это и быстрее, и проще в коде. Thread.ofVirtual().factory().newThread(...) создаёт поток за ~100 наносекунд, а его запуск обходится в ~200 наносекунд.
Совет 5: Используйте ScopedValue вместо ThreadLocal. Если вы передаёте контекст через ThreadLocal, замените на ScopedValue. Это безопаснее, быстрее и идеально интегрируется с виртуальными потоками. ThreadLocal всё ещё работает, но его семантика утечки данных делает его ненадёжным при масштабировании.
Распространённые ошибки и ловушки
При переходе на виртуальные потоки разработчики часто сталкиваются с рядом проблем. Рассмотрим наиболее распространённые из них и способы их решения.
Ошибка 1: Использование ThreadLocal в библиотеках. Многие библиотеки (Spring, Hibernate, Log4j) используют ThreadLocal для хранения контекста. При работе с виртуальными потоками ThreadLocal может работать некорректно, потому что виртуальные потоки могут быть созданы и уничтожены без привязки к одному нативному потоку. Решение: обновите библиотеки до версий, поддерживающих виртуальные потоки (Spring 6.1+, Hibernate 6.4+, Log4j 3.0+).
Ошибка 2: Ограниченные ресурсы. Создание миллионов виртуальных потоков безопасно с точки зрения памяти потоков, но может исчерпать другие ресурсы: файловые дескрипторы, соединения с базой данных, HTTP-соединения. Используйте семафоры или bounded executors для контроля количества одновременных операций с внешними ресурсами.
Ошибка 3: Неправильное использование Future.get(). Вызов Future.get() блокирует текущий виртуальный поток. Это безопасно — виртуальный поток освобождает carrier thread — но если вы ждёте результат нескольких Future последовательно, вы теряете параллелизм. Используйте CompletableFuture.allOf() или StructuredTaskScope для параллельного ожидания.
Ошибка 4: Использование Thread.interrupt() неправильно. Прерывание виртуальных потоков работает иначе, чем прерывание traditional потоков. Когда вы вызываете Thread.interrupt() на виртуальном потоке, он устанавливает флаг прерывания и, если виртуальный поток заблокирован, возобновляет его и выбрасывает InterruptedException. Но если виртуальный поток выполняет CPU-bound работу, флаг просто устанавливается, и код должен проверять его самостоятельно.
Ошибка 5: Миксрование платформенных и виртуальных потоков. Если вы создаёте ExecutorService с traditional pool внутри виртуального потока, вы создаёте «бутылочное горлышко». Задачи в virtual thread блокируются при ожидании traditional pool. Вместо этого используйте виртуальные потоки на всех уровнях иерархии вызовов.
Сравнение с другими языками
Виртуальные потоки Java не являются революционной идеей — аналогичные механизмы существуют в других языках программирования уже много лет. Go имеет goroutine с 2009 года, Kotlin поддерживает coroutine с 2018 года, Erlang имеет lightweight processes с 1986 года. Даже Python получил asyncio ещё в 2014 году.
Отличие Java в том, что виртуальные потоки полностью совместимы с существующим блокирующим кодом. В отличие от Go, где goroutine требуют использования каналов (channels) для коммуникации, виртуальные потоки Java работают с привычными примитивами синхронизации: synchronized, wait(), notify(), ReentrantLock, Semaphore. Это означает, что существующий код можно мигрировать на виртуальные потоки, просто заменив способ создания потоков.
Kotlin coroutine, в свою очередь, требуют использования suspend функций и suspending scope, что вносит изменения в сигнатуры методов. В Java virtual threads работают без каких-либо изменений в API — любой Runnable или Callable можно запустить в виртуальном потоке.
Миграция существующего кода на виртуальные потоки
Если у вас есть существующее приложение на Java 17 или ранних версиях, миграция на виртуальные потоки может быть surprisingly простой. Вот пошаговый план.
Шаг 1: Обновите JDK до 21+. Это обязательное требование. Виртуальные потоки стабильны начиная с JDK 21. Убедитесь, что все ваши зависимости совместимы с JDK 21. Большинство популярных библиотек (Spring Boot 3.x, Hibernate 6.x, Jackson 2.15+) уже поддерживают JDK 21.
Шаг 2: Замените пулы потоков. Найдите все места, где создаются Executors.newFixedThreadPool(), Executors.newCachedThreadPool(), ThreadPoolExecutor. Замените на Executors.newVirtualThreadPerTaskExecutor(). Это безопасная замена, которая не меняет семантику большинства приложений.
Шаг 3: Замените synchronized. Найдите все synchronized блоки в коде, который будет работать в виртуальных потоках. Замените на ReentrantLock. В большинстве случаев это простая замена: объявите private final ReentrantLock lock = new ReentrantLock(), замените synchronized (lock) { ... } на lock.lock(); try { ... } finally { lock.unlock(); }.
Шаг 4: Проверьте ThreadLocal. Найдите все использования ThreadLocal. Если данные хранятся на время запроса и очищаются через remove(), это безопасно. Если ThreadLocal используется для кэширования долгоживущих данных, рассмотрите замену на ScopedValue.
Шаг 5: Тестируйте с -Djdk.tracePinnedThreads=short. Запустите ваше приложение с этой JVM-опцией и проверьте, нет ли pinning. Если pinning обнаружен, проанализируйте стектрейс и устраните источник.
Полный пример: сервер обработки запросов
Давайте создадим полноценный пример сервера обработки запросов на виртуальных потоках. Этот пример объединяет все рассмотренные концепции: создание виртуальных потоков через ExecutorService, структурированную конкурентность для параллельных запросов к БД и микросервису, ScopedValue для передачи контекста запроса и обработку ошибок.
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.logging.Logger;
public class VirtualThreadServer {
static final Logger logger =
Logger.getLogger(VirtualThreadServer.class.getName());
static final ScopedValue<String> REQUEST_ID =
ScopedValue.newInstance();
static final ScopedValue<String> USER_TOKEN =
ScopedValue.newInstance();
static final AtomicInteger activeRequests = new AtomicInteger(0);
static final AtomicInteger completedRequests = new AtomicInteger(0);
record Request(String requestId, String userToken,
String path) {}
record Response(int statusCode, String body) {}
static Response handleRequest(String requestId, String token,
String path) {
return ScopedValue.runWhere(REQUEST_ID, requestId,
() -> ScopedValue.where(USER_TOKEN, token,
() -> processRequest(path)));
}
static Response processRequest(String path) {
try {
String reqId = REQUEST_ID.get();
String token = USER_TOKEN.get();
logger.info("Обработка " + reqId
+ " path=" + path + " token=" + token);
String dbResult = queryDatabase(path);
String apiResult = callExternalApi(path);
return new Response(200,
"DB: " + dbResult + ", API: " + apiResult);
} catch (Exception e) {
logger.severe("Ошибка: " + e.getMessage());
return new Response(500, "Internal Error: "
+ e.getMessage());
}
}
static String queryDatabase(String path) throws Exception {
Thread.sleep(50);
return "data-from-db(" + path + ")";
}
static String callExternalApi(String path) throws Exception {
Thread.sleep(100);
return "api-response(" + path + ")";
}
public static void main(String[] args) throws Exception {
var requests = java.util.List.of(
new Request("req-001", "token-alice", "/users/1"),
new Request("req-002", "token-bob", "/orders/42"),
new Request("req-003", "token-carol", "/products/7"),
new Request("req-004", "token-dave", "/users/2"),
new Request("req-005", "token-eve", "/orders/100")
);
System.out.println("Запуск сервера на виртуальных потоках...");
long start = System.nanoTime();
try (var executor =
Executors.newVirtualThreadPerTaskExecutor()) {
var futures = requests.stream()
.map(req -> executor.submit(() -> {
activeRequests.incrementAndGet();
try {
return handleRequest(req.requestId(),
req.userToken(), req.path());
} finally {
activeRequests.decrementAndGet();
completedRequests.incrementAndGet();
}
}))
.toList();
for (var future : futures) {
Response response = future.get();
System.out.printf("[%d] %s%n",
response.statusCode(), response.body());
}
}
long elapsed = (System.nanoTime() - start) / 1_000_000;
System.out.println();
System.out.println("Обработано: "
+ completedRequests.get() + " запросов");
System.out.println("Общее время: " + elapsed + " мс");
System.out.println("Среднее: "
+ (elapsed / requests.size()) + " мс/запрос");
System.out.println("Параллельно: максимум "
+ requests.size() + " одновременно");
}
}Виртуальные потоки и Spring Framework
Spring Framework 6.1+ и Spring Boot 3.2+ полностью поддерживают виртуальные потоки. Для включения достаточно добавить одну строку в конфигурацию приложения. Spring автоматически заменит traditional DispatcherServlet thread pool на виртуальные потоки, что даёт значительное повышение производительности без каких-либо изменений в бизнес-логике.
Важно понимать, что Spring использует synchronized в некоторых внутренних компонентах, что может вызвать pinning. Рекомендуется тестировать приложение с -Djdk.tracePinnedThreads=short и сообщать о pinning в issue tracker Spring Framework. В целом, поддержка виртуальных потоков в Spring находится на высоком уровне и продолжает улучшаться.
Будущее: что дальше для виртуальных потоков
Виртуальные потоки — это не终点, а начало новой эры в Java-многозадачности. OpenJDK команда продолжает работу над улучшениями. В JDK 24 планируется устранение pinning для многих случаев использования synchronized, что значительно уменьшит количество кода, требующего модификации. Также продолжается работа над Virtual Threads v2 с улучшенной поддержкой continuation и более эффективным использованием памяти.
Долгосрочная стратегия Java в области并发 — это сделать блокирующий код столь же масштабируемым, как и async/await в других языках, без изменения семантики языка. Виртуальные потоки — первый и самый важный шаг на этом пути. В сочетании со структурированной конкурентностью и ScopedValue, они формируют новую модель concurrent programming в Java.
Итоги урока
- Виртуальные потоки (Project Loom) — это lightweight потоки, управляемые JVM, которые мультиплексируются поверх небольшого пула нативных потоков (carrier threads)
- Основное преимущество: блокирующий код становится масштабируемым — можно создавать миллионы виртуальных потоков без ущерба для производительности
- Создание:
Thread.ofVirtual(),Thread.startVirtualThread(),Executors.newVirtualThreadPerTaskExecutor() - Pinning — главная ловушка:
synchronizedблокирует carrier thread; заменяйте наReentrantLock - StructuredTaskScope обеспечивает гарантию завершения всех задач при выходе из scope
- ScopedValue — безопасная замена ThreadLocal для передачи контекста по иерархии вызовов
- Обработка ошибок: всегда оборачивайте тело задачи в try-catch, используйте Future для получения ошибок
- Виртуальные потоки идеальны для I/O-bound задач, но не дают преимущества для CPU-bound
- Миграция: обновите JDK до 21+, замените пулы потоков, замените synchronized, проверьте ThreadLocal
- Spring Boot 3.2+ поддерживает виртуальные потоки из коробки
Тест: Виртуальные потоки (Project Loom)
6 вопросов