Что такое CyclicBarrier?

CyclicBarrier: Синхронизация потоков

Что это такое

CyclicBarrier — это класс синхронизации в Java (java.util.concurrent), который позволяет группе потоков ждать друг друга в определённой точке перед продолжением выполнения. Это циклический барьер — он может быть переиспользован несколько раз.

Ключевые отличия от CountDownLatch

Основной концепт

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {
    public static void main(String[] args) throws InterruptedException {
        // Создаём барьер для 3 потоков
        CyclicBarrier barrier = new CyclicBarrier(3);
        
        for (int i = 1; i <= 3; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " ждёт");
                    barrier.await();  // Ждём остальные потоки
                    System.out.println(Thread.currentThread().getName() + " продолжает");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "Поток-" + i).start();
        }
        
        Thread.sleep(3000);
    }
}

// Вывод:
// Поток-1 ждёт
// Поток-2 ждёт
// Поток-3 ждёт
// Поток-3 продолжает
// Поток-1 продолжает
// Поток-2 продолжает

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

1. Создаёшь CyclicBarrier с количеством потоков
2. Каждый поток вызывает await() — блокируется
3. Когда все потоки дошли до await() — барьер открывается
4. Все потоки продолжают выполнение одновременно
5. Барьер автоматически переустанавливается для следующего цикла

Пример: параллельная обработка данных

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.BrokenBarrierException;

public class DataProcessor {
    static class Worker implements Runnable {
        private String name;
        private int[] data;
        private CyclicBarrier barrier;
        
        Worker(String name, int[] data, CyclicBarrier barrier) {
            this.name = name;
            this.data = data;
            this.barrier = barrier;
        }
        
        @Override
        public void run() {
            try {
                // Фаза 1: загрузка данных
                System.out.println(name + " загружает данные...");
                Thread.sleep(1000);
                System.out.println(name + " загрузил данные");
                
                // Все ждут друг друга перед обработкой
                barrier.await();
                
                // Фаза 2: обработка
                System.out.println(name + " обрабатывает данные...");
                Thread.sleep(500);
                System.out.println(name + " обработал данные");
                
                // Ждём перед сохранением
                barrier.await();
                
                // Фаза 3: сохранение
                System.out.println(name + " сохраняет результат");
                
            } catch (InterruptedException | BrokenBarrierException e) {
                Thread.currentThread().interrupt();
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        int[] data = new int[100];
        CyclicBarrier barrier = new CyclicBarrier(3);
        
        for (int i = 1; i <= 3; i++) {
            new Thread(new Worker("Worker-" + i, data, barrier)).start();
        }
        
        Thread.sleep(5000);
    }
}

CyclicBarrier с действием

Можно выполнить действие после того, как все потоки достигли барьера:

public class BarrierWithAction {
    public static void main(String[] args) throws InterruptedException {
        // Второй параметр — действие после барьера
        CyclicBarrier barrier = new CyclicBarrier(2, () -> {
            System.out.println("[ДЕЙСТВИЕ] Все потоки готовы, начинаем!");
        });
        
        new Thread(() -> {
            try {
                System.out.println("Поток 1 готов");
                barrier.await();
                System.out.println("Поток 1 продолжает");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
        
        new Thread(() -> {
            try {
                System.out.println("Поток 2 готов");
                barrier.await();
                System.out.println("Поток 2 продолжает");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
        
        Thread.sleep(2000);
    }
}

// Вывод:
// Поток 1 готов
// Поток 2 готов
// [ДЕЙСТВИЕ] Все потоки готовы, начинаем!
// Поток 1 продолжает
// Поток 2 продолжает

Обработка исключений

public class BarrierExceptionHandling {
    public static void main(String[] args) throws InterruptedException {
        CyclicBarrier barrier = new CyclicBarrier(3);
        
        for (int i = 1; i <= 3; i++) {
            final int threadNum = i;
            new Thread(() -> {
                try {
                    if (threadNum == 2) {
                        throw new RuntimeException("Ошибка в потоке 2");
                    }
                    System.out.println("Поток " + threadNum + " ждёт");
                    barrier.await();
                } catch (RuntimeException e) {
                    System.out.println("Ошибка: " + e.getMessage());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
        
        Thread.sleep(2000);
    }
}

// Когда один поток выбросит исключение,
// BrokenBarrierException выбросится у остальных

Практические примеры использования

1. Модель Producer-Consumer

public class ProducerConsumer {
    static class Stage {
        private CyclicBarrier barrier;
        
        Stage(CyclicBarrier barrier) {
            this.barrier = barrier;
        }
        
        void waitForOthers() throws Exception {
            barrier.await();
        }
    }
}

2. Map-Reduce паттерн

// Map фаза: все потоки обрабатывают данные
// await() на барьере
// Reduce фаза: все потоки объединяют результаты
// await() на барьере
// Результат выведен

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

CyclicBarrier barrier = new CyclicBarrier(2);

// Основной метод — ждёт остальные потоки
barrier.await();

// С timeout
barrier.await(5, TimeUnit.SECONDS);

// Получить количество ожидающих потоков
int waiting = barrier.getNumberWaiting();

// Получить количество партий
int parties = barrier.getParties();

// Сбросить барьер (в случае ошибки)
barrier.reset();

CyclicBarrier vs CountDownLatch

// CountDownLatch — один раз
CountDownLatch latch = new CountDownLatch(3);
// После countDown() трижды, нельзя переиспользовать

// CyclicBarrier — переиспользуется
CyclicBarrier barrier = new CyclicBarrier(3);
// После await() трижды, автоматически переустанавливается

Заключение

CyclicBarrier идеально подходит для:

• Многофазных алгоритмов с синхронизацией фаз
• Параллельной обработки данных в несколько этапов
• Симуляций, где потоки должны развиваться в lock-step
• Координации работы нескольких потоков перед дорогостоящей операцией