Конкурентность и многопоточность в Java: продвинутые механизмы

Введение: от базовых synchronized к современным инструментам

Java — один из немногих языков, который с рождения имеет встроенную поддержку многопоточности. Однако на начальном уровне разработчики часто ограничиваются ключевым словом synchronized и простыми Thread. В реальных высоконагруженных системах этого недостаточно: блокировки порождают дедлоки, а наивное создание потоков приводит к перегрузке JVM.

В этой статье мы разберём продвинутые механизмы конкурентности в Java, которые позволяют писать эффективный, масштабируемый и безопасный многопоточный код. Материал рассчитан на разработчиков, уже знакомых с основами многопоточности (мониторы, wait/notify, volatile).

Мы рассмотрим четыре ключевые темы: Fork/Join Framework для рекурсивных задач, CompletableFuture для асинхронного программирования, StampedLock как альтернативу ReadWriteLock и грамотное использование ThreadLocal.

1. Fork/Join Framework: рекурсивное разделение задач

Fork/Join Framework появился в Java 7 и предназначен для эффективного использования многоядерных процессоров при решении задач, которые можно разделить на подзадачи (divide and conquer). Основной класс — ForkJoinPool, который использует механизм «воровства задач» (work-stealing).

1.1. Принцип работы

Вместо того чтобы создавать потоки вручную, вы описываете задачу как наследника RecursiveTask (если есть возвращаемое значение) или RecursiveAction (без возврата). Затем вы рекурсивно делите задачу на части, пока они не станут достаточно малыми для прямого выполнения.

1.2. Пример: параллельное суммирование массива

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

public class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;
    private final int[] array;
    private final int start, end;

    public SumTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        }
        int mid = start + (end - start) / 2;
        SumTask left = new SumTask(array, start, mid);
        SumTask right = new SumTask(array, mid, end);
        left.fork();          // асинхронно запускаем левую часть
        long rightResult = right.compute(); // вычисляем правую в текущем потоке
        long leftResult = left.join();      // ждём результат левой
        return leftResult + rightResult;
    }

    public static void main(String[] args) {
        int[] data = new int[1_000_000];
        for (int i = 0; i < data.length; i++) data[i] = i;

        ForkJoinPool pool = ForkJoinPool.commonPool();
        long result = pool.invoke(new SumTask(data, 0, data.length));
        System.out.println("Сумма: " + result);
    }
}

Важно: используйте ForkJoinPool.commonPool() — он уже оптимизирован под количество ядер. Избегайте создания собственных пулов без необходимости.

2. CompletableFuture: асинхронность без callback-ада

До Java 8 асинхронное программирование было возможно через Future, но он был синхронным (блокирующий вызов get()). CompletableFuture решает эту проблему, предоставляя мощный функциональный API для композиции асинхронных операций.

2.1. Базовые сценарии

Вы можете создавать CompletableFuture, запускать задачи асинхронно и объединять их цепочками:

import java.util.concurrent.CompletableFuture;

public class AsyncExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            // Имитация долгой операции
            sleep(1000);
            return "Привет";
        }).thenApply(greeting -> greeting + ", мир!")
          .thenAccept(System.out::println);

        // Ждём завершения (для демонстрации)
        sleep(2000);
    }

    private static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

2.2. Комбинирование нескольких задач

Используйте thenCombine для параллельного выполнения двух независимых задач и объединения их результатов:

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Java");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Concurrency");

String result = future1.thenCombine(future2, (a, b) -> a + " " + b).join();
System.out.println(result); // Java Concurrency

Продвинутый совет: задавайте собственный Executor для пула потоков, чтобы не перегружать ForkJoinPool.commonPool() задачами ввода-вывода:

ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture.supplyAsync(() -> fetchFromDb(), executor);

3. StampedLock: оптимистичные блокировки

Классический ReadWriteLock имеет недостаток: даже при чтении поток блокируется, если другой поток пишет. StampedLock (Java 8) добавляет режим оптимистичного чтения, который не блокирует запись, пока данные действительно не изменились.

3.1. Как это работает

При оптимистичном чтении вы получаете «штамп» (stamp) и проверяете его валидность после чтения. Если данные не изменились — отлично, блокировка не понадобилась. Если изменились — переходите к обычной пессимистичной блокировке.

3.2. Пример реализации счётчика

import java.util.concurrent.locks.StampedLock;

public class StampedCounter {
    private final StampedLock lock = new StampedLock();
    private int count = 0;

    public void increment() {
        long stamp = lock.writeLock();
        try {
            count++;
        } finally {
            lock.unlockWrite(stamp);
        }
    }

    public int getCount() {
        long stamp = lock.tryOptimisticRead();
        int currentCount = count;
        if (!lock.validate(stamp)) { // данные могли измениться
            stamp = lock.readLock();
            try {
                currentCount = count;