Какие знаешь инструменты из библиотеки Concurrency в Java?

Инструменты библиотеки java.util.concurrent

Это мощная библиотека Java для работы с многопоточностью (concurrency). Она предоставляет готовые инструменты для безопасной работы с потоками без использования низкоуровневых синхронизационных примитивов.

Основные компоненты

1. Thread Pools (ExecutorService)

Управление потоками через пулы вместо создания новых потоков на каждую задачу.

// Пул с фиксированным количеством потоков
ExecutorService executor = Executors.newFixedThreadPool(5);

// Отправка задачи
executor.submit(() -> {
    System.out.println("Задача выполняется в потоке: " + Thread.currentThread().getName());
});

// Корректное завершение
executor.shutdown();
if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
    executor.shutdownNow();
}

// Другие типы пулов
ExecutorService fixedPool = Executors.newFixedThreadPool(10);
ExecutorService cachedPool = Executors.newCachedThreadPool();
ExecutorService singlePool = Executors.newSingleThreadExecutor();

2. Future и Callable

Для получения результатов асинхронных операций.

ExecutorService executor = Executors.newFixedThreadPool(2);

// Callable возвращает результат, Runnable нет
Callable<String> task = () -> {
    Thread.sleep(2000);
    return "Результат вычисления";
};

Future<String> future = executor.submit(task);

try {
    // Ждём результат с таймаутом
    String result = future.get(3, TimeUnit.SECONDS);
    System.out.println(result);
} catch (TimeoutException e) {
    System.out.println("Задача не завершилась вовремя");
    future.cancel(true);
}

executor.shutdown();

3. CountDownLatch

Синхронизация потоков: один или больше потоков ждут, пока другие потоки завершат работу.

CountDownLatch latch = new CountDownLatch(3);

// Запускаем 3 рабочих потока
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        try {
            System.out.println("Рабочий начал: " + Thread.currentThread().getName());
            Thread.sleep(2000);
            System.out.println("Рабочий завершил");
        } finally {
            latch.countDown();  // Уменьшаем счётчик
        }
    }).start();
}

// Главный поток ждёт
System.out.println("Главный ждёт...");
latch.await();
System.out.println("Все рабочие завершили!");

4. CyclicBarrier

Повторно используемая барьер для синхронизации группы потоков.

CyclicBarrier barrier = new CyclicBarrier(3, () -> {
    System.out.println("Все потоки достигли барьера!");
});

for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        try {
            System.out.println("Поток начал работу");
            Thread.sleep((long)(Math.random() * 5000));
            System.out.println("Поток ждёт барьера");
            barrier.await();  // Ждём, пока все потоки не придут сюда
            System.out.println("Поток продолжил после барьера");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }).start();
}

5. Semaphore

Ограничение доступа к ресурсам (пермиты).

Semaphore semaphore = new Semaphore(2);  // Максимум 2 потока одновременно

for (int i = 0; i < 5; i++) {
    new Thread(() -> {
        try {
            System.out.println("Поток ожидает доступ");
            semaphore.acquire();  // Получить пермит
            
            System.out.println("Поток получил доступ " + Thread.currentThread().getName());
            Thread.sleep(3000);
            
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();  // Освободить пермит
            System.out.println("Поток освободил ресурс");
        }
    }).start();
}

6. ReentrantLock

Больше возможностей, чем synchronized, включая попытку захвата с таймаутом.

ReentrantLock lock = new ReentrantLock();
private int counter = 0;

public void increment() {
    lock.lock();
    try {
        counter++;
    } finally {
        lock.unlock();  // Важно в finally!
    }
}

// С таймаутом
public boolean incrementIfAvailable() {
    try {
        if (lock.tryLock(1, TimeUnit.SECONDS)) {
            try {
                counter++;
                return true;
            } finally {
                lock.unlock();
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return false;
}

7. ReadWriteLock

Дозволяет несколько одновременных читателей, но исключительный доступ для писателя.

ReadWriteLock lock = new ReentrantReadWriteLock();
private Map<String, String> cache = new HashMap<>();

public String get(String key) {
    lock.readLock().lock();
    try {
        return cache.get(key);
    } finally {
        lock.readLock().unlock();
    }
}

public void put(String key, String value) {
    lock.writeLock().lock();
    try {
        cache.put(key, value);
    } finally {
        lock.writeLock().unlock();
    }
}

8. Concurrent Collections

Токобезопасные коллекции без явной синхронизации.

// Вместо Collections.synchronizedMap(new HashMap<>())
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
map.put("key1", 100);
map.putIfAbsent("key2", 200);
map.replace("key1", 100, 150);

// Другие concurrent коллекции
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(10);

9. BlockingQueue

Чередь для взаимодействия потоков с блокировкой при пустоте/переполнении.

BlockingQueue<String> queue = new LinkedBlockingQueue<>(5);

// Producer
ExecutorService executor = Executors.newFixedThreadPool(1);
executor.submit(() -> {
    try {
        queue.put("Сообщение 1");
        queue.put("Сообщение 2");
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

// Consumer
ExecutorService executor2 = Executors.newFixedThreadPool(1);
executor2.submit(() -> {
    try {
        String msg = queue.take();  // Блокирует, если пусто
        System.out.println("Получено: " + msg);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

10. AtomicInteger, AtomicReference

Атомарные переменные для безопасных операций без блокировок.

AtomicInteger counter = new AtomicInteger(0);

// Потокобезопасные операции
counter.incrementAndGet();     // ++counter
counter.decrementAndGet();     // --counter
counter.getAndAdd(5);          // counter += 5
counter.compareAndSet(10, 20); // if (counter == 10) counter = 20

// Для ссылок
AtomicReference<String> ref = new AtomicReference<>("initial");
ref.set("new value");
String value = ref.get();

Таблица сравнения инструментов

Пример: Producer-Consumer паттерн

BlockingQueue<String> queue = new LinkedBlockingQueue<>(10);
ExecutorService executor = Executors.newFixedThreadPool(2);

// Producer
executor.submit(() -> {
    for (int i = 0; i < 5; i++) {
        try {
            queue.put("Item " + i);
            System.out.println("Produced: Item " + i);
            Thread.sleep(500);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
});

// Consumer
executor.submit(() -> {
    try {
        while (true) {
            String item = queue.take();
            System.out.println("Consumed: " + item);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});