Класс Phaser

Регистрация. В отличие от других барьеров, количество участников, зарегистрированных для синхронизации на фазер, может меняться со временем. Задачи могут регистрироваться в любое время (с помощью методов register(), bulkRegister(int) или форм конструкторов, устанавливающих начальное количество участников), и при необходимости дерегистрироваться при любом прибытии (с помощью arriveAndDeregister()). Как и в большинстве основных синхронизационных конструкций, регистрация и дерегистрация влияют только на внутренние счетчики; они не устанавливают дальнейшее внутреннее ведение записей, поэтому задачи не могут запросить, зарегистрированы ли они. (Однако вы можете ввести такое ведение записей, унаследовав этот класс.)

Синхронизация. Как и CyclicBarrier, фазер может многократно ожидаться. Метод arriveAndAwaitAdvance() имеет эффект, аналогичный CyclicBarrier.await. Каждая генерация фазера имеет связанный номер фазы. Номер фазы начинается с нуля и увеличивается, когда все участники прибывают в фазер, возвращаясь к нулю после достижения Integer.MAX_VALUE. Использование номеров фаз позволяет независимо управлять действиями при прибытии в фазер и при ожидании других, с помощью двух типов методов, которые могут быть вызваны любым зарегистрированным участником:

• Прибытие. Методы arrive() и arriveAndDeregister() регистрируют прибытие. Эти методы не блокируют, но возвращают связанный номер фазы прибытия; то есть номер фазы фазера, к которому относится прибытие. Когда прибывает последний участник для данной фазы, выполняется необязательное действие, и фаза переходит в следующий этап. Эти действия выполняет участник, вызывающий продвижение фазы, и они организованы путем переопределения метода onAdvance(int, int), который также управляет завершением. Переопределение этого метода аналогично, но более гибко, чем предоставление действия барьера для CyclicBarrier.
• Ожидание. Метод awaitAdvance(int) требует аргумента, указывающего номер фазы прибытия, и возвращается, когда фазер переходит к (или уже находится в) другой фазе. В отличие от аналогичных конструкций с использованием CyclicBarrier, метод awaitAdvance продолжает ожидать, даже если поток ожидания прерывается. Также доступны прерываемые и временные версии, но исключения, возникающие во время ожидания задач прерывисто или с таймаутом, не изменяют состояние фазера. При необходимости вы можете выполнить любое связанное восстановление в обработчиках этих исключений, часто после вызова forceTermination. Фазеры также могут использоваться задачами, выполняющимися в ForkJoinPool. Прогресс гарантирован, если уровень параллелизма пула может вместить максимальное количество одновременно заблокированных участников.

Завершение. Фазер может перейти в состояние завершения, которое можно проверить с помощью метода isTerminated(). При завершении все методы синхронизации немедленно возвращаются, не ожидая продвижения, как указано отрицательным возвращаемым значением. Аналогично, попытки регистрации при завершении не имеют эффекта. Завершение запускается, когда вызов onAdvance возвращает true. По умолчанию реализация возвращает true в случае, если дерегистрация привела к тому, что количество зарегистрированных участников стало нулевым. Как показано ниже, когда фазеры управляют действиями с фиксированным числом итераций, часто удобно переопределить этот метод, чтобы вызвать завершение, когда текущий номер фазы достиг порога. Метод forceTermination() также доступен для внезапного освобождения ожидающих потоков и разрешения их завершения.

Иерархия. Фазеры могут быть иерархическими (т. е. построены в виде древовидных структур) для уменьшения конкуренции. Фазеры с большим количеством участников, которые в противном случае столкнулись бы с высокими затратами на конкуренцию при синхронизации, могут быть настроены таким образом, что группы подфазеров делят общего родителя. Это может значительно увеличить пропускную способность, даже если приводит к большей нагрузке на каждую операцию.

В дереве иерархических фазеров регистрация и дерегистрация дочерних фазеров с их родителями управляются автоматически. Всякий раз, когда количество зарегистрированных участников дочернего фазера становится отличным от нуля (как установлено в конструкторе Phaser(Phaser,int), register() или bulkRegister(int)), дочерний фазер регистрируется у своего родителя. Всякий раз, когда количество зарегистрированных участников становится равным нулю в результате вызова arriveAndDeregister(), дочерний фазер дерегистрируется у своего родителя.

Мониторинг. Хотя методы синхронизации могут вызываться только зарегистрированными участниками, текущее состояние фазера может быть отслежено любым вызывающим лицом. В любой момент времени существует getRegisteredParties() участников в общей сложности, из которых getArrivedParties() прибыли в текущую фазу (getPhase()). Когда оставшиеся (getUnarrivedParties()) участники прибывают, фаза переходит в следующий этап. Значения, возвращаемые этими методами, могут отражать временные состояния и, таким образом, в целом не пригодны для управления синхронизацией. Метод toString() возвращает снимки этих запросов состояния в форме, удобной для неформального мониторинга.

Эффекты согласованности памяти: Действия перед любым типом метода arrive предшествуют соответствующему продвижению фазы и действиям onAdvance (если они есть), которые в свою очередь предшествуют действиям после продвижения фазы.

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

Фазер может быть использован вместо CountDownLatch для управления одноразовым действием, обслуживающим переменное количество участников. Типичная идиома заключается в том, что метод, устанавливающий это, сначала регистрирует, затем запускает все действия, затем дерегистрирует, как показано ниже:

 
 void runTasks(List<Runnable> tasks) {
   Phaser startingGate = new Phaser(1); // "1" to register self
   // create and start threads
   for (Runnable task : tasks) {
     startingGate.register();
     new Thread(() -> {
       startingGate.arriveAndAwaitAdvance();
       task.run();
     }).start();
   }

   // deregister self to allow threads to proceed
   startingGate.arriveAndDeregister();
 }

Один из способов заставить набор потоков многократно выполнять действия для заданного числа итераций — переопределить onAdvance:

 
 void startTasks(List<Runnable> tasks, int iterations) {
   Phaser phaser = new Phaser() {
     protected boolean onAdvance(int phase, int registeredParties) {
       return phase >= iterations - 1 || registeredParties == 0;
     }
   };
   phaser.register();
   for (Runnable task : tasks) {
     phaser.register();
     new Thread(() -> {
       do {
         task.run();
         phaser.arriveAndAwaitAdvance();
       } while (!phaser.isTerminated());
     }).start();
   }
   // allow threads to proceed; don't wait for them
   phaser.arriveAndDeregister();
 }

 
   // ...
   phaser.register();
   while (!phaser.isTerminated())
     phaser.arriveAndAwaitAdvance();

Связанные конструкции могут быть использованы для ожидания определенных номеров фаз в контекстах, где вы уверены, что фаза никогда не будет циклически повторяться Integer.MAX_VALUE. Например:

 
 void awaitPhase(Phaser phaser, int phase) {
   int p = phaser.register(); // assumes caller not already registered
   while (p < phase) {
     if (phaser.isTerminated())
       // ... deal with unexpected termination
     else
       p = phaser.arriveAndAwaitAdvance();
   }
   phaser.arriveAndDeregister();
 }

Для создания набора n задач с использованием дерева фазеров можно использовать код следующего вида, предполагая класс задачи с конструктором, принимающим Phaser, с которым он регистрируется при создании. После вызова build(new Task[n], 0, n, new Phaser()), эти задачи можно запустить, например, отправив их в пул:

 
 void build(Task[] tasks, int lo, int hi, Phaser ph) {
   if (hi - lo > TASKS_PER_PHASER) {
     for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
       int j = Math.min(i + TASKS_PER_PHASER, hi);
       build(tasks, i, j, new Phaser(ph));
     }
   } else {
     for (int i = lo; i < hi; ++i)
       tasks[i] = new Task(ph);
       // assumes new Task(ph) performs ph.register()
   }
 }

Примечания к реализации: В этой реализации максимальное количество участников ограничено 65535. Попытки зарегистрировать дополнительных участников приводят к IllegalStateException. Однако вы можете и должны создавать иерархические фазеры, чтобы вместить произвольно большие наборы участников.