java.util.concurrent и виртуальные потоки: урок по Java

В реальном коде потоки руками не создают и wait/notify не пишут. На собесе ждут, что вы знаете пулы, CompletableFuture, ConcurrentHashMap и в 2026 году виртуальные потоки Java 21.

ExecutorService и пулы

try (ExecutorService pool = Executors.newFixedThreadPool(8)) {   // close() с Java 19
    Future<Integer> f = pool.submit(() -> compute());
    Integer r = f.get(2, TimeUnit.SECONDS);
}

Под капотом ThreadPoolExecutor(core, max, keepAlive, queue, threadFactory, rejectionHandler). Логика при новой задаче: потоков меньше core, создать поток; иначе положить в очередь; очередь полна, создать поток до max; иначе сработает политика отказа (AbortPolicy бросает RejectedExecutionException, CallerRunsPolicy выполняет в вызывающем потоке).

Ловушки фабрик Executors:

newFixedThreadPool использует неограниченную LinkedBlockingQueue: при перегрузке очередь растёт до OOM, а max не работает.
newCachedThreadPool создаёт потоки без ограничения.

В проде пул создают явно, с ограниченной очередью, именами потоков и метриками. Пул обязательно закрывают (shutdown, awaitTermination или try-with-resources).

Исключение в задаче submit не печатается: оно сохраняется в Future и вылетит из get() в виде ExecutionException.

CompletableFuture

CompletableFuture<Order> order = CompletableFuture.supplyAsync(() -> loadOrder(id), ioPool);
CompletableFuture<User> user = CompletableFuture.supplyAsync(() -> loadUser(uid), ioPool);

order.thenCombine(user, OrderView::new)
     .orTimeout(2, TimeUnit.SECONDS)
     .exceptionally(e -> OrderView.empty())
     .thenAccept(this::send);

thenApply преобразует результат, thenCompose подставляет следующий асинхронный шаг (аналог flatMap), thenCombine объединяет два независимых.
allOf ждёт все, anyOf первый.
exceptionally и handle обрабатывают ошибки, orTimeout и completeOnTimeout ограничивают время.
Без явного executor задачи *Async идут в ForkJoinPool.commonPool(). Блокирующий I/O туда класть нельзя.

Конкурентные коллекции

ConcurrentHashMap: чтение без блокировок, запись через CAS в пустую корзину или блокировку первого узла корзины (с Java 8, сегментов больше нет). null запрещён. Атомарные составные операции: putIfAbsent, computeIfAbsent, merge. Итераторы weakly consistent, не бросают CME.
CopyOnWriteArrayList: каждая запись копирует массив. Для редко меняемых списков с частым чтением (слушатели).
BlockingQueue (ArrayBlockingQueue, LinkedBlockingQueue): put ждёт места, take ждёт элемент. Основа producer-consumer.

Ошибка: if (!map.containsKey(k)) map.put(k, v) даже на ConcurrentHashMap это гонка check-then-act. Правильно map.computeIfAbsent(k, key -> v).

Синхронизаторы и блокировки

ReentrantLock: как synchronized, плюс tryLock с таймаутом, прерываемое ожидание, несколько Condition. unlock() всегда в finally.
ReadWriteLock, StampedLock: много читателей или один писатель.
CountDownLatch: дождаться N событий, одноразовый. CyclicBarrier: N потоков ждут друг друга, многоразовый. Semaphore: не больше N одновременных доступов.
Атомики: AtomicInteger, AtomicReference на CAS; LongAdder для счётчиков под высокой конкуренцией.

Виртуальные потоки (Java 21)

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

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
        executor.submit(() -> callHttp());   // поток на задачу
    }
}
Thread.ofVirtual().start(() -> work());

Что важно знать:

Ускоряют пропускную способность при блокирующем I/O, а не вычисления. CPU-задачам они не помогают.
Не пулят: создают новый на каждую задачу. Ограничение параллелизма делают через Semaphore, а не размером пула.
Pinning: в Java 21-23 блокировка внутри synchronized закрепляет виртуальный поток за носителем, и носители кончаются. Совет был заменять synchronized на ReentrantLock. В Java 24 это исправлено (JEP 491), в Java 21 актуально.
ThreadLocal работает, но при миллионе потоков это миллион копий. Для передачи контекста есть ScopedValue (стабилен с Java 25).
В Spring Boot 3.2+ включаются свойством spring.threads.virtual.enabled=true.

Что спросят на собесе

Как работает ThreadPoolExecutor, что будет при переполнении очереди?
Чем опасен Executors.newFixedThreadPool в проде?
Как устроен ConcurrentHashMap в Java 8+? Почему он не принимает null?
thenApply против thenCompose, куда уходят задачи без executor?
Что такое виртуальные потоки, когда они помогают и что такое pinning?

java.util.concurrent и виртуальные потоки