Класс Phaser

public class Phaser
extends Object

Многократно используемый барьер синхронизации, похожий по функциональности на CyclicBarrier и CountDownLatch, но поддерживающий более гибкое использование.

Регистрация. В отличие от других барьеров, количество участников, зарегистрированных для синхронизации на фазер, может меняться со временем. Задачи могут регистрироваться в любое время (с помощью методов 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() возвращает снимки этих запросов состояния в удобной для неформального мониторинга форме.

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

Phaser может использоваться вместо 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()
   }
 }

Лучшее значение TASKS_PER_PHASER в основном зависит от ожидаемых скоростей синхронизации. Значение, такое как четыре, может быть приемлемым для чрезвычайно малых тел задач на каждой фазе (следовательно, высоких скоростей), или до сотен для чрезвычайно больших.

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