4 Phaser¶
Phaser¶
Una de las funcionalidades más potentes de la API de concurrencia de Java es la posibilidad de ejecutar tareas concurrentes que se sincronicen en distintas fases, para lo que utiliza la clase Phaser (secuenciador por fases).
Phaser
Barrera de sincronización de un número variable de hilos que se reinicia automáticamente
Este mecanismo es útil cuando las tareas concurrentes se dividen en varias fases (pasos o etapas). La clase Phaser nos proporciona el mecanismo para sincronizar los hilos al final de cada fase, de manera que ningún hilo comienza una nueva fase hasta que todos los hilos hayan concluido la fase anterior.
Internamente, un objeto Phaser mantiene cuatro datos muy importantes para su funcionamiento:
- phase: Número de fase en el que se encuentra. Inicialmente vale 0.
- registered parties: Número de participantes registrados para participar en la siguiente fase.
- unarrived: Número de participantes que aún no han concluido la fase actual. Al comenzar una fase se reinicializa su valor al correspondiente a registered parties.
- terminated: Indicador booleano de si el proceso de sincronización a través del Phaser se puede considerar como terminado o no. Su valor inicial es false.
A diferencia de con otros tipos de barreras, como CyclicBarrier, el número de hilos participantes en una sincronización con Phaser puede variar durante el proceso de sincronización.
Para indicar el número inicial de hilos participantes registrados podemos pasar un valor entero al constructor Phaser(parties) de la clase Phaser. Si le pasamos un valor negativo se lanzará la excepción IllegalArgumentException. Dicho valor será establecido como valor inicial de unarrived y de registered parties.
Si usamos el constructor vacío Phaser(), el valor de unarrived y de registered parties será inicializado a 0.
A diferencia de los CyclicBarrier, los propios hilos pueden registrarse como participantes en la sincronización, para lo que llamarán al método register(), que registra un nuevo participante en la fase actual, o al método bulkRegister(parties), que registra el número de hilos indicados como participantes de la fase actual. Estos métodos retornan el número de fase en el que se han registrado (el valor actual de phase).
En ambos casos, los hilos registrados son contabilizados como hilos que aún no han concluido la fase actual, incrementando el valor de unarrived, y como hilos registrados para la próxima fase, incrementando el valor de registered parties.
Si el proceso de sincronización ya ha sido marcada como terminado para cuando el hilo se registra, es decir, si el valor de terminated es true para cuando un hilo llama a register() o bulkRegister(parties), estos métodos no tendrán ningún efecto sobre los hilos ni sobre el Phaser, y retornarán un valor negativo como número de fase.
Si un hilo trata de registrarse justo en el momento de un cambio de fase, es decir, como veremos más adelante, mientras se está ejecutando el método onAdvance() del Phaser, el registro de los nuevos hilo esperará hasta que dicho método concluya su ejecución, registrando a los hilos como participantes de la nueva fase.
El hecho de que los propios hilos puedan registrarse en el proceso de sincronización otorga una flexibilidad adicional que no posee CyclicBarrier. Gracias a ello, si usamos Phaser no tenemos por qué saber de antemano el número de hilos a sincronizar, ya que cada uno de ellos puede registrarse personalmente, algo que no era posible en el caso de CyclicBarrier.
Cada vez que un hilo concluye una fase debe informar de ello al Phaser. Dicha operación se denomina arrive. Por otra parte, si queremos que el hilo espere a que el resto de hilos concluya la fase actual del Phaser deberá de comunicárselo expresamente a través de una operación conocida como await advance. Esta es otra de las diferencias principales respecto a otro tipos de barreras, en las que ambas operaciones, la indicación de haber llegado a la barrera y la operación de esperar al resto de hilos que deben llegar a la barrera correspondía a una única operación indivisible, como en el caso del método await() de CyclicBarrier.
La clase Phaser nos ofrece distintos métodos para llevar a cabo las operaciones arrive y await advance. El primero de ellos es el método arrive(), mediante el que un hilo informa al Phaser de que ha concluido la fase actual del mismo, sin ponerse a esperar a que el resto de hilos lo haga. Como consecuencia el Phaser decrementará el valor de unarrived. Este método retorna la fase actual del Phaser a la que el hilo ha llegado, o un valor negativo si el Phaser ya ha sido marcado como terminado, es decir, si terminated es true.
Gracias a esta separación entre las operaciones arrive y await advance, el número de participantes no tiene por qué corresponder obligatoriamente con el número de hilos a sincronizar, aunque lo habitual es que coincidan.
Como complemento a arrive(), tenemos el método awaitAdvance(phase), que permite al hilo indicar que desea esperar a que los hilos concluyan la fase pasada como argumento, es decir, a que el valor de unarrived sea 0. Si el Phaser no se encuentra en ese momento en dicha fase, o ya ha sido marcado como terminado (terminated es true), el método no bloqueará al hilo que lo ejecuta, retornando inmediatamente.
El valor proporcionado como argumento a awaitAdvance(phase) provendrá normalmente del valor retornado por una llamada anterior al método arrive(). De hecho, si queremos realizar ambas operaciones arrive y await advance seguidas una detrás de la otra haríamos awaitAdvance(arrive()).
Como esta opción de hacer ambas operaciones una seguida de la otra es la más habitual, la clase Phaser proporciona un método para ejecutarlas con una única llamada, a través del método arriveAndAwaitAdvance(), que informa de que el hilo la concluido la fase actual del Phaser (arrive), decrementando unarrived, y lo suspende hasta que todos los hilos participantes en dicha fase de la sincronización hayan informado de que la han concluido (awaitAdvance). Como vemos, el funcionamiento del método arriveAndAwaitAdvance() es similar al del método await() de un CyclicBarrier.
Otra opción disponible para los hilos es indicar que se ha concluido la fase actual y que ya no se quiere participar en el resto de fases de la sincronización, por lo que ni siquiera va a esperar al resto de hilos. Para ello hará uso del método arriveAndDeregister(), que decrementa el valor de unarrived y además hace que en las siguientes fases de la sincronización haya un participante menos, decrementando el valor de registered parties.
Este método es usado habitualmente por el hilo que controla la inicialización del Phaser, por ejemplo el hilo principal, que desea coordinador el inicio del proceso de ejecución de los otros hilos, como veremos más adelante en un ejemplo en el que se usa el Phaser como barrera inicial.
Es importante que cuando un hilo ya no quiera participar en las siguientes fases de la sincronización haga arriveAndDeregister() ya que si tan sólo hace arrive() no decrementará registered parties, afectando al correcto funcionamiento del Phaser y del resto de hilos que continúen en la sincronización, que estarán esperando a que dicho hilo haga el arrive en la fase siguiente para poder continuar.
Los métodos arrive(), awaitAdvance(), arriveAndAwaitAdvance() y arriveAndDeregister() no se ven afectados por el hecho de que el hilo ya estuviera marcado para interrupción.
Además, los métodos arriveAndAwaitAdvance() y awaitAdvance() ni siquiera se ven afectados por el hecho de que el hilo sea interrumpido mientras está esperando a que el resto de hilos concluyan la fase actual del Phaser. Si queremos que se lance la interrupción InterruptedException en este caso, deberemos usar el método awaitAdvanceInterruptibly(phase) o la versión con timeout explicado en el siguiente párrafo.
El método awaitAdvanceInterruptibly(phase, timeout, timeUnit) se encuentra sobrecargado para indicar un tiempo máximo que deseamos que el hilo sea suspendido en espera de que el resto de hilos concluyan la fase actual del Phaser, transcurrido el cuál se lanzará la excepción TimeoutException y el hilo será reactivado inmediatamente.
A diferencia de en el caso de CyclicBarrier, en el que la generación de la excepción InterruptedException o de TimeoutException en el método await() producía que la barrera fuera marcada como rota, la generación de la excepción InterruptedException en los métodos anteriores del Phaser no produce ningún cambio de estado en el mismo.
Todos los métodos vistos con anterioridad para realizar las operaciones de arrive y await advance retornan el número de fase en el que se encuentra el Phaser cuando se llaman, o un valor negativo si la sincronización ya había concluido para cuando se llama a estos métodos, es decir, cuando el Phaser ya había sido marcado como terminado.
Cuando el hecho de que un hilo ejecute la operación arrive hace el valor de unarrived pase a ser 0, se considera que la fase actual ha concluido, y en dicho hilo se ejecuta el método onAdvance(phase, registeredParties) del Phaser, que recibe el número de fase que se acaba de concluir y el número de hilos que están registrados para la próxima fase. Al cambiar de fase, unarrived se reinicializa al valor de registered parties y el valor de phase es incrementado en una unidad.
El método onAdvance() debe retornar un valor booleano indicativo de si el proceso de sincronización o ha concluido o no y por tanto si el Phaser debe ser marcado como terminado, activando el valor terminated. La implementación por defecto de este método retornará false, a no ser que el número de participantes registrados para la nueva fase sea 0:
// Default implementation of onAdvance
protected boolean onAdvance(int phase, int registeredParties) {
return registeredParties == 0;
}
Si se produce una excepción en este método, dicha excepción será propagada al hilo que propició su ejecución al concluir la fase y el Phaser no avanzará a la siguiente fase.
Para saber si un Phaser ha sido marcado como terminado podemos usar el método isTerminated(). Si queremos forzar el marcado como terminado del Phaser podemos usar su método forceTermination(), que establece el valor true en terminated, independientemente del número de participantes que hayan ejecutado el método arriveAndDeregister(), o de que hayamos sobrescrito el método onAdvance(). Si había hilos esperando que otros finalizaran la fase actual, son reactivados.
Cuando un objeto Phaser entra en este estado de terminación, es decir cuando terminated es true, los métodos relacionados con la operación await advance retornan un valor negativo, en vez del valor positivo correspondiente al número de fase actual que retornan habitualmente, algo que deberemos comprobar cuando el objeto Phaser pueda ser terminado abruptamente.
La clase Phaser define además una serie de métodos informativos que nos permiten hacer un seguimiento del estado del mismo:
getPhase(): Retorna la fase en la que se encuentra elPhaser.getRegisteredParties(): Retorna el número de participantes registrados en la fase actual delPhaser.getArrivedParties(): Retorna el número de participantes que han concluido la fase actual delPhaser.getUnarrivedParties(): Retorna el número de participantes registrados que no han concluido aún la fase actual delPhaser.
Veamos distintos casos de uso de Phaser.
Phaser como barrera de sincronización¶
void runTasks(List<Runnable> tasks) {
// The thread than calls runTasks creates the Phaser and registers itself.
final Phaser phaser = new Phaser(1);
// Create and starts the tasks
for (final Runnable task : tasks) {
// Register the task. It must be done before running the tasks, so they
// can wait each other.
phaser.register();
new Thread() {
public void run() {
// The first thing each task does is awaiting creation of all tasks.
phaser.arriveAndAwaitAdvance();
task.run();
}
}.start();
}
// After registering the task the thread that calls runTasks can continue
// without any later synchronization
phaser.arriveAndDeregister();
}
Como vemos el código anterior es bastante parecido al funcionamiento de una CyclicBarrier que actúe como barrera inicial. Sin embargo el uso de Phaser lo hace más flexible, porque no tenemos por qué saber de antemano el número de hilos a sincronizar, ya que cada uno de ellos puede registrarse, algo que no era posible en el caso de CyclicBarrier.
Phaser como barrera cíclica con un número de ciclos¶
void startTasks(List<Runnable> tasks, final int iterations) {
// The thread that calls startTasks creates and custom phaser is marked
// as terminated when the number of iterations is done or no
// parties are registered.
final Phaser phaser = new Phaser() {
protected boolean onAdvance(int phase, int registeredParties) {
return phase >= iterations || registeredParties == 0;
}
};
// The thread that calls startTasks register itself and each task before
// stating them
phaser.register();
for (final Runnable task : tasks) {
phaser.register();
new Thread() {
public void run() {
do {
// Each task runs and then await to the rest to advance
// to the next iteration until the phaser is terminated.
task.run();
phaser.arriveAndAwaitAdvance();
} while (!phaser.isTerminated());
}
}.start();
}
// The thread that calls startTasks can continue without any later synchronization
phaser.arriveAndDeregister();
}
Proyecto Phaser¶
En este proyecto desarrollaremos una aplicación que simula una conjunto de amigos que han quedado para tomar unas cervezas en un pub. El plan consiste en salir de casa y quedar en el pub, esperando a que lleguen todos los amigos para tomar la primera cerveza. Cuando todos hayan terminado de beber la primera cerveza, se tomarán una segunda cerveza, y cuando todos se la hayan tomado, se despedirán y se irán a casa.
En esta demo tendremos diferentes tipos de amigos:
- Tres amigos que realizan al completo todas las fases cumpliendo con las esperas de rigor (friend #0, #1 y #2, representados por la clase Friend).
- Un amigo impaciente que realizar al completo todas la fases pero que no piensa esperar a nadie (friend #3, representado por la clase ImpacientFriend).
- Un amigo que sólo se tomará una cerveza, pero que esperará a los amigos en el pub y esperará a que se terminen la primera cerveza antes de irse a casa (friend #4, representado por la clase OneBeerFriend).
- Un amigo que va a unirse más tarde, y que se tomará las cervezas que pueda dependiendo que cuando haya informado a los amigos de su incorporación. Los amigos respetuosos lo esperarán en la fase en la que se encuentren (friend #5, representado por la clase TardyFriend).
- Un amigo que para cuando trata de unirse es demasiado tarde porque el resto de amigos ya no están tomando cervezas (friend #6, representado también por la clase TardyFriend, pero que será lanzado mucho más tarde, cuando el phaser ya se encuentre terminado).
Para la sincronización se creará una clase que extienda de Phaser en la que sobrescribiremos el método onAdvance() para mostrar mensajes personalizados cuando se cambie de fase, y que de por terminada la secuenciación cuando los amigos se terminen la segunda cerveza.
import java.util.concurrent.TimeUnit;
public class Main {
private static final int NORMAL_FRIENDS = 3;
public static void main(String[] args) throws InterruptedException {
int i;
FriendsPhaser phaser = new FriendsPhaser();
for (i = 0; i < NORMAL_FRIENDS; i++) {
new Thread(new Friend("Friend #" + i, phaser), "Friend #" + i).start();
}
// This friend won't wait for anybody.
new Thread(new ImpacientFriend("Friend #" + i, phaser), "Friend #" + i).start();
i++;
// This friend will drink just one beer and will go home.
new Thread(new OneBeerFriend("Friend #" + i, phaser), "Friend #" + i).start();
i++;
// This friend will join be late, maybe when other friends are drinking.
TimeUnit.SECONDS.sleep(9);
new Thread(new TardyFriend("Friend #" + i, phaser), "Friend #" + i).start();
i++;
// This friend will join very very late.
TimeUnit.SECONDS.sleep(12);
new Thread(new TardyFriend("Friend #" + i, phaser), "Friend #" + i).start();
}
}
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.Phaser;
public class FriendsPhaser extends Phaser {
public static final int ARRIVE_TO_PUB_PHASE = 0;
public static final int FINISH_FIST_BEER_PHASE = 1;
public static final int FINISH_SECOND_BEER_PHASE = 2;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
@Override
protected boolean onAdvance(int phase, int registeredParties) {
switch (phase) {
case ARRIVE_TO_PUB_PHASE:
System.out.printf("%s -> All %d friends arrived to pub (executed in %s)\n",
LocalTime.now().format(dateTimeFormatter), registeredParties,
Thread.currentThread().getName());
break;
case FINISH_FIST_BEER_PHASE:
System.out.printf("%s -> All %d friends finished their first beer (executed in %s)\n",
LocalTime.now().format(dateTimeFormatter), registeredParties,
Thread.currentThread().getName());
break;
case FINISH_SECOND_BEER_PHASE:
System.out.printf("%s -> All %d friends finished their second beer (executed in %s)\n",
LocalTime.now().format(dateTimeFormatter), registeredParties,
Thread.currentThread().getName());
// The phaser is marked as terminated.
return true;
}
return super.onAdvance(phase, registeredParties);
}
}
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.concurrent.Phaser;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
public class Friend implements Runnable {
private final String name;
private final Phaser phaser;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
public Friend(String name, Phaser phaser) {
Objects.requireNonNull(name);
Objects.requireNonNull(phaser);
this.name = name;
this.phaser = phaser;
}
@Override
public void run() {
phaser.register();
try {
goToPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going to the pub\n", name);
return;
}
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends in the pub\n", name);
return;
}
try {
firstBeerInPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while drinking the first beer\n", name);
return;
}
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends to finish their fist beer\n", name);
return;
}
try {
secondBeerInPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while drinking the second beer\n", name);
return;
}
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends to finish their second beer\n", name);
return;
}
try {
goHome();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going back home\n", name);
}
}
private void goToPub() throws InterruptedException {
System.out.printf("%s -> %s is leaving home\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has arrived in the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void firstBeerInPub() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has finished the first beer\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void secondBeerInPub() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has finished the second beer\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void goHome() throws InterruptedException {
System.out.printf("%s -> %s is leaving the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s is at home\n",
LocalTime.now().format(dateTimeFormatter), name);
}
}
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.concurrent.Phaser;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
public class ImpacientFriend implements Runnable {
private final String name;
private final Phaser phaser;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
public ImpacientFriend(String name, Phaser phaser) {
Objects.requireNonNull(name);
Objects.requireNonNull(phaser);
this.name = name;
this.phaser = phaser;
}
@Override
public void run() {
phaser.register();
try {
goToPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going to the pub\n", name);
return;
}
phaser.arriveAndDeregister();
try {
firstBeerInPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while drinking the first beer\n", name);
return;
}
try {
secondBeerInPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while drinking the second beer\n", name);
return;
}
try {
goHome();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going back home\n", name);
}
}
private void goToPub() throws InterruptedException {
System.out.printf("%s -> %s is leaving home\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has arrived in the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void firstBeerInPub() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has finished the first beer\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void secondBeerInPub() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has finished the second beer\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void goHome() throws InterruptedException {
System.out.printf("%s -> %s is leaving the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s is at home\n",
LocalTime.now().format(dateTimeFormatter), name);
}
}
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.concurrent.Phaser;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
public class OneBeerFriend implements Runnable {
private final String name;
private final Phaser phaser;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
public OneBeerFriend(String name, Phaser phaser) {
Objects.requireNonNull(name);
Objects.requireNonNull(phaser);
this.name = name;
this.phaser = phaser;
}
@Override
public void run() {
phaser.register();
try {
goToPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going to the pub\n", name);
return;
}
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends in the pub\n", name);
return;
}
try {
firstBeerInPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while drinking the first beer\n", name);
return;
}
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends to finish their fist beer\n", name);
return;
}
// No more beers.
phaser.arriveAndDeregister();
try {
goHome();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going back home\n", name);
}
}
private void goToPub() throws InterruptedException {
System.out.printf("%s -> %s is leaving home\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has arrived in the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void firstBeerInPub() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has finished the first beer\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void goHome() throws InterruptedException {
System.out.printf("%s -> %s is leaving the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s is at home\n",
LocalTime.now().format(dateTimeFormatter), name);
}
}
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.concurrent.Phaser;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
public class TardyFriend implements Runnable {
private final String name;
private final Phaser phaser;
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss");
public TardyFriend(String name, Phaser phaser) {
Objects.requireNonNull(name);
Objects.requireNonNull(phaser);
this.name = name;
this.phaser = phaser;
}
@Override
public void run() {
if (!phaser.isTerminated()) {
int joinPhase = phaser.register();
System.out.printf("%s -> %s has joined friends in phase #%d\n",
LocalTime.now().format(dateTimeFormatter), name, joinPhase);
try {
goToPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going to the pub\n", name);
return;
}
// Tardy friends shouln't do arrive on previous phases in order not to
// interfere with the synchonization process or the current phase of the
// phaser.
if (joinPhase <= FriendsPhaser.ARRIVE_TO_PUB_PHASE) {
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends in the pub\n", name);
return;
}
}
try {
firstBeerInPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while drinking the first beer\n", name);
return;
}
if (joinPhase <= FriendsPhaser.FINISH_FIST_BEER_PHASE) {
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends to finish their fist beer\n", name);
return;
}
}
try {
secondBeerInPub();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while drinking the second beer\n", name);
return;
}
if (joinPhase <= FriendsPhaser.FINISH_SECOND_BEER_PHASE) {
try {
phaser.awaitAdvanceInterruptibly(phaser.arrive());
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while waiting for friends to finish their second beer\n", name);
return;
}
}
try {
goHome();
} catch (InterruptedException e) {
System.out.printf("%s has been interrupted while going back home\n", name);
}
} else {
System.out.printf("%s -> %s called his friends too late\n",
LocalTime.now().format(dateTimeFormatter), name);
}
}
private void goToPub() throws InterruptedException {
System.out.printf("%s -> %s is leaving home\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has arrived in the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void firstBeerInPub() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has finished the first beer\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void secondBeerInPub() throws InterruptedException {
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s has finished the second beer\n",
LocalTime.now().format(dateTimeFormatter), name);
}
private void goHome() throws InterruptedException {
System.out.printf("%s -> %s is leaving the pub\n",
LocalTime.now().format(dateTimeFormatter), name);
TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(5) + 1);
System.out.printf("%s -> %s is at home\n",
LocalTime.now().format(dateTimeFormatter), name);
}
@SuppressWarnings("unused")
private void awaitPhase(Phaser phaser, int currentPhase, int expectedPhase) {
while (currentPhase < expectedPhase && !phaser.isTerminated()) {
System.out.printf("%s -> %s is waiting phase #%d to finish\n",
LocalTime.now().format(dateTimeFormatter), name, currentPhase);
currentPhase = phaser.arriveAndAwaitAdvance();
}
}
}
Si ejecutamos el programa veremos el comportamiento de los distintos tipos de amigos respecto al phaser.