Класс Phaser

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

Синхронизация. Как и CyclicBarrier, за Phaser можно неоднократно ожидать. Метод 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 задач с помощью дерева фазеров вы можете использовать код следующей формы, предполагая класс Task с конструктором, принимающим 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. Однако вы можете и должны создавать иерархические фазеры для размещения произвольно больших наборов участников.