CyclicBarrier: как заставить потоки ждать друг друга и не словить дедлок
В прошлом посте мы разобрались, как распараллеливать задачи через ExecutorService. Но что делать, если задача одна, но она состоит из нескольких независимых этапов, которые должны выполняться строго по очереди?
Представим микросервис отчетов. Чтобы собрать один PDF-файл, нам нужно:
1. Потоку А сходить в БД за транзакциями.
2. Потоку Б посчитать налоги.
3. Потоку В подтянуть курсы валют.
Проблема в том, что формировать финальный документ можно только тогда, когда все три шага завершены.
Если поток А закончит первым, он не может просто начать собирать файл — данных от потоков Б и В еще нет.
Для таких сценариев синхронизации в JDK есть CyclicBarrier.
Как это работает
CyclicBarrier — это точка сбора. Мы создаем барьер на определенное количество участников:
// 3 участника + колбэк, который сработает, когда все трое придут к барьеру
CyclicBarrier barrier = new CyclicBarrier(3, () -> {
System.out.println("Все этапы готовы, собираем финальный отчет");
});
Каждый рабочий поток делает свою часть работы и приходит к барьеру:
executor.submit(() -> {
fetchTransactions(); // Шаг 1
try {
barrier.await(); // Ждем остальных
} catch (Exception e) {
Thread.currentThread().interrupt();
}
});
executor.submit(() -> {
calculateTaxes(); // Шаг 2
try {
barrier.await(); // Ждем остальных
} catch (Exception e) {
Thread.currentThread().interrupt();
}
});
executor.submit(() -> {
loadExchangeRates(); // Шаг 3
try {
barrier.await(); // Ждем остальных
} catch (Exception e) {
Thread.currentThread().interrupt();
}
});
Когда третий поток вызывает await(), счетчик достигает нуля. Барьер «ломается», выпускает всех ждущих, сбрасывает свой счетчик обратно до 3 и выполняет переданный в конструктор Runnable. После этого этот же самый экземпляр барьера можно использовать повторно для следующего отчета.
Почему он называется Cyclic (Циклический)
Потому что его можно переиспользовать. В отличие от одноразового CountDownLatch, после того как барьер преодолен и участники освобождены, его внутренний счетчик автоматически возвращается в исходное состояние. Это идеально ложится на пайплайны обработки: собрали первый отчет — тут же запускаем сборку второго на этом же барьере.
Важные нюансы
* Таймауты. Никогда не вызывайте голый barrier.await(). Один зависший поток (например, сетевой тайм-аут при загрузке курсов валют) навсегда заблокирует всю группу. Используйте перегрузку с таймаутом:
barrier.await(5, TimeUnit.SECONDS);
Если время вышло, текущий поток выкинет TimeoutException, а остальные участники, дождавшиеся его, получат BrokenBarrierException.
* Сломанный барьер. Если один из участников вылетел с исключением или не успел в таймаут, барьер переходит в состояние broken. Все последующие вызовы await() будут сразу бросать BrokenBarrierException. Проверить состояние можно через barrier.isBroken(), а сбросить принудительно — через barrier.reset().
* Номера участников. Внутри await() каждый поток может узнать свой порядковый номер прибытия:
int index = barrier.await();
if (index == 0) {
// Этот поток зашел в await последним и "разблокировал" барьер
}
Главный подвох: потерянные исключения
Если таск внутри submit() упал с ошибкой до вызова barrier.await(), об этом никто не узнает. Пул схавает исключение, а остальные потоки так и останутся висеть на барьере вечно (точнее, до таймаута), ожидая мертвого участника.
Чтобы этого избежать, результат вычислений и статус ошибки нужно передавать наружу через потокобезопасный контейнер (например, тот самый ConcurrentHashMap из прошлого поста или AtomicReference), проверяя его перед вызовом await().