Что такое CountDownLatch?

CountDownLatch в Java

CountDownLatch — это синхронизирующий механизм из пакета java.util.concurrent, который позволяет одному или нескольким потокам дождаться, пока другие потоки завершат свои операции. Это очень полезно для координации между потоками и является одним из самых часто используемых инструментов многопоточности.

Основная концепция

CountDownLatch работает как счётчик:

1. Создаётся с начальным числом (count)
2. Потоки вызывают countDown(), чтобы уменьшить счётчик
3. Потоки вызывают await(), чтобы дождаться, пока счётчик достигнет нуля
4. Когда счётчик = 0, все ожидающие потоки пробуждаются

Простой пример

import java.util.concurrent.CountDownLatch;

public class SimpleCountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);  // Счётчик = 3
        
        // Запускаем 3 рабочих потока
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " работает...");
                    Thread.sleep(2000);  // Имитируем работу
                    System.out.println(Thread.currentThread().getName() + " завершился");
                    latch.countDown();  // Уменьшаем счётчик
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }
        
        System.out.println("Главный поток ждёт завершения других потоков...");
        latch.await();  // Ждём, пока счётчик станет 0
        System.out.println("Все потоки завершились, программа продолжает работу");
    }
}

// Вывод:
// Главный поток ждёт завершения других потоков...
// Thread-0 работает...
// Thread-1 работает...
// Thread-2 работает...
// Thread-0 завершился
// Thread-1 завершился
// Thread-2 завершился
// Все потоки завершились, программа продолжает работу

Реальный пример: параллельная загрузка ресурсов

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ResourceLoader {
    private ExecutorService executorService = Executors.newFixedThreadPool(4);
    
    public void loadResources() throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(4);  // 4 типа ресурсов
        
        executorService.submit(() -> loadUsers(latch));
        executorService.submit(() -> loadProducts(latch));
        executorService.submit(() -> loadCategories(latch));
        executorService.submit(() -> loadConfigurations(latch));
        
        System.out.println("Загрузка ресурсов началась...");
        latch.await();  // Ждём, пока все ресурсы загрузятся
        System.out.println("Все ресурсы загружены, приложение готово!");
    }
    
    private void loadUsers(CountDownLatch latch) {
        try {
            System.out.println("Загружаем пользователей...");
            Thread.sleep(3000);
            System.out.println("Пользователи загружены");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            latch.countDown();
        }
    }
    
    private void loadProducts(CountDownLatch latch) {
        try {
            System.out.println("Загружаем товары...");
            Thread.sleep(2000);
            System.out.println("Товары загружены");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            latch.countDown();
        }
    }
    
    private void loadCategories(CountDownLatch latch) {
        try {
            System.out.println("Загружаем категории...");
            Thread.sleep(1500);
            System.out.println("Категории загружены");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            latch.countDown();
        }
    }
    
    private void loadConfigurations(CountDownLatch latch) {
        try {
            System.out.println("Загружаем конфигурацию...");
            Thread.sleep(1000);
            System.out.println("Конфигурация загружена");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            latch.countDown();
        }
    }
}

Пример: race condition тестирование

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class RaceConditionTester {
    private AtomicInteger counter = new AtomicInteger(0);
    
    public void testConcurrentIncrement() throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        CountDownLatch latch = new CountDownLatch(10);
        
        // 10 потоков, каждый увеличивает счётчик 1000 раз
        for (int i = 0; i < 10; i++) {
            executor.submit(() -> {
                try {
                    for (int j = 0; j < 1000; j++) {
                        counter.incrementAndGet();
                    }
                } finally {
                    latch.countDown();
                }
            });
        }
        
        latch.await();  // Ждём, пока все потоки завершат свою работу
        executor.shutdown();
        
        // Должно быть 10000 (10 потоков * 1000 инкрементов)
        System.out.println("Итоговое значение: " + counter.get());
    }
}

CountDownLatch vs CyclicBarrier

Пример: CyclicBarrier (альтернатива для повторного использования)

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(3, () -> {
            System.out.println("Все потоки достигли барьера!");
        });
        
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " ждёт у барьера");
                    barrier.await();  // Ждём других потоков
                    System.out.println(Thread.currentThread().getName() + " прошёл барьер");
                } catch (Exception e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }
    }
}

Важные методы CountDownLatch

CountDownLatch latch = new CountDownLatch(5);

// Ждём, пока счётчик станет 0
latch.await();

// Ждём с timeout'ом (возвращает true если счётчик = 0, false если timeout)
boolean finished = latch.await(5, TimeUnit.SECONDS);

// Уменьшаем счётчик на 1
latch.countDown();

// Получаем текущее значение счётчика
long currentCount = latch.getCount();

Пример с обработкой ошибок

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

public class TaskExecutorWithErrorHandling {
    private CountDownLatch latch;
    private AtomicInteger errorCount = new AtomicInteger(0);
    
    public void executeTasksWithErrorHandling(int taskCount) throws InterruptedException {
        latch = new CountDownLatch(taskCount);
        
        for (int i = 0; i < taskCount; i++) {
            final int taskId = i;
            new Thread(() -> {
                try {
                    executeTask(taskId);
                } catch (Exception e) {
                    System.err.println("Task " + taskId + " failed: " + e.getMessage());
                    errorCount.incrementAndGet();
                } finally {
                    latch.countDown();
                }
            }).start();
        }
        
        boolean allCompleted = latch.await(30, TimeUnit.SECONDS);
        
        if (!allCompleted) {
            System.out.println("Timeout: некоторые задачи не завершились");
        } else if (errorCount.get() > 0) {
            System.out.println("Завершено с ошибками: " + errorCount.get());
        } else {
            System.out.println("Все задачи успешно завершены");
        }
    }
    
    private void executeTask(int taskId) throws Exception {
        // Имитируем работу
        Thread.sleep((long) (Math.random() * 5000));
        System.out.println("Task " + taskId + " completed");
    }
}

Когда использовать CountDownLatch

1. Ожидание инициализации — ждать, пока все компоненты загрузятся
2. Параллельные задачи — запустить несколько задач параллельно, ждать их завершения
3. Тестирование — синхронизировать потоки в unit тестах
4. Batch обработка — обработать несколько батчей последовательно
5. Стартовые сигналы — все потоки начинают одновременно

Ключевые моменты

• CountDownLatch — инструмент для ожидания завершения других потоков
• Счётчик может только уменьшаться, переиспользовать нельзя
• Безопасен для многопоточной среды
• Для повторного использования используй CyclicBarrier
• Всегда помещай countDown() в finally блок
• Используй timeout версию await() для защиты от бесконечного ожидания