Java 7的并发包中推出了Phaser，其功能跟CyclicBarrier和CountDownLatch有些重叠，但是提供了更灵活的用法，例如支持动态调整注册任务的数量等。本文在Phaser自带的示例代码基础上进行一下简单的分析。

注册(Registration)

Phaser支持通过register()和bulkRegister(int parties)方法来动态调整注册任务的数量，此外也支持通过其构造函数进行指定初始数量。在适当的时机，Phaser支持减少注册任务的数量，例如 arriveAndDeregister()。单个Phaser实例允许的注册任务数的上限是65535。

到达(Arrival)

正如Phaser类的名字所暗示，每个Phaser实例都会维护一个phase number，初始值为0。每当所有注册的任务都到达Phaser时，phase number累加，并在超过Integer.MAX_VALUE后清零。arrive()和arriveAndDeregister()方法用于记录到 达，arriveAndAwaitAdvance()方法用于记录到达，并且等待其它未到达的任务。

终止(Termination)

Phaser支持终止。Phaser终止之后，调用register()和bulkRegister(int parties)方法没有任何效果，arriveAndAwaitAdvance()方法也会立即返回。触发终止的时机是在protected boolean onAdvance(int phase, int registeredParties)方法返回时，如果该方法返回true，那么Phaser会被终止。默认实现是在注册任务数为0时返回true（即 return registeredParties == 0;）。此外，forceTermination()方法用于强制终止，isTerminated()方法用于判断是否已经终止。

层次结构(Tiering)

Phaser支持层次结构，即通过构造函数Phaser(Phaser parent)和Phaser(Phaser parent, int parties)构造一个树形结构。这有助于减轻因在单个的Phaser上注册过多的任务而导致的竞争，从而提升吞吐量，代价是增加单个操作的开销。

使用案例

样例1

在有些场景下，我们希望控制多个线程的启动时机：例如在并发相关的单元测试中，有时需要控制线程的启动时机，以期获得最大程度的并发，通常我们会使用CountDownLatch，以下是使用Phaser的版本。

Java代码  

1. import java.util.concurrent.Phaser;
2. public class PhaserTest1 {
3. public static void main(String args[]) {
4. //
5. final int count = 5;
6. final Phaser phaser = new Phaser(count);
7. for(int i = 0; i < count; i++) {
8. System.out.println("starting thread, id: " + i);
9. final Thread thread = new Thread(new Task(i, phaser));
10. thread.start();
11. }
12. }
13. public static class Task implements Runnable {
14. //
15. private final int id;
16. private final Phaser phaser;
17. public Task(int id, Phaser phaser) {
18. this.id = id;
19. this.phaser = phaser;
20. }
21. @Override
22. public void run() {
23. phaser.arriveAndAwaitAdvance();
24. System.out.println("in Task.run(), phase: " + phaser.getPhase() + ", id: " + this.id);
25. }
26. }
27. }

以上例子中，由于线程是在一个循环中start，因此start的时机有一定的间隔。本例中这些线程实际开始工作的时机是在所有的线程都调用了phaser.arriveAndAwaitAdvance()之后。

此外，如果留心arriveAndAwaitAdvance()方法的签名，会发现它并没有抛出InterruptedException，实际上，即使 当前线程被中断，arriveAndAwaitAdvance()方法也不会返回，而是继续等待。如果在等待时希望可中断，或者可超时，那么需要使用以下 方法：

Java代码  

1. awaitAdvance(arrive())  // 等效于arriveAndAwaitAdvance()
2. awaitAdvanceInterruptibly(int phase)
3. awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)

样例2

有些时候我们希望只有在某些外部条件满足时，才真正开始任务的执行，例如：

Java代码  

1. import java.io.BufferedReader;
2. import java.io.InputStreamReader;
3. import java.util.concurrent.Phaser;
4. public class PhaserTest2 {
5. public static void main(String args[]) throws Exception {
6. //
7. final Phaser phaser = new Phaser(1);
8. for(int i = 0; i < 5; i++) {
9. phaser.register();
10. System.out.println("starting thread, id: " + i);
11. final Thread thread = new Thread(new Task(i, phaser));
12. thread.start();
13. }
14. //
15. System.out.println("Press ENTER to continue");
16. BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
17. reader.readLine();
18. phaser.arriveAndDeregister();
19. }
20. public static class Task implements Runnable {
21. //
22. private final int id;
23. private final Phaser phaser;
24. public Task(int id, Phaser phaser) {
25. this.id = id;
26. this.phaser = phaser;
27. }
28. @Override
29. public void run() {
30. phaser.arriveAndAwaitAdvance();
31. System.out.println("in Task.run(), phase: " + phaser.getPhase() + ", id: " + this.id);
32. }
33. }
34. }

以上例子中，只有当用户按下回车之后，任务才真正开始执行。需要注意的是，arriveAndDeregister()方法不会被阻塞，并且返回到达时的phase number（arrive方法也是如此）。

样例3

CyclicBarrier支持barrier action, Phaser同样也支持。不同之处是Phaser的barrier action需要改写onAdvance方法来进行定制。

Java代码  

1. import java.util.concurrent.Phaser;
2. public class PhaserTest3 {
3. public static void main(String args[]) throws Exception {
4. //
5. final int count = 5;
6. final int phaseToTerminate = 3;
7. final Phaser phaser = new Phaser(count) {
8. @Override
9. protected boolean onAdvance(int phase, int registeredParties) {
10. System.out.println("====== " + phase + " ======");
11. return phase >= phaseToTerminate || registeredParties == 0;
12. }
13. };
14. //
15. for(int i = 0; i < count; i++) {
16. System.out.println("starting thread, id: " + i);
17. final Thread thread = new Thread(new Task(i, phaser));
18. thread.start();
19. }
20. }
21. public static class Task implements Runnable {
22. //
23. private final int id;
24. private final Phaser phaser;
25. public Task(int id, Phaser phaser) {
26. this.id = id;
27. this.phaser = phaser;
28. }
29. @Override
30. public void run() {
31. do {
32. try {
33. Thread.sleep(500);
34. } catch(InterruptedException e) {
35. // NOP
36. }
37. System.out.println("in Task.run(), phase: " + phaser.getPhase() + ", id: " + this.id);
38. phaser.arriveAndAwaitAdvance();
39. } while(!phaser.isTerminated());
40. }
41. }
42. }

本例中的barrier action只是简单地打印了一条信息，此外在超过指定的迭代次数后终止了Phaser。

样例4

在Smaple 3的例子中，主线程在其它工作线程结束之前已经终止。如果希望主线程等待这些工作线程结束，除了使用Thread.join()之外，也可以尝试以下的方式：

Java代码  

1. import java.util.concurrent.Phaser;
2. public class PhaserTest4 {
3. public static void main(String args[]) throws Exception {
4. //
5. final int count = 5;
6. final int phaseToTerminate = 3;
7. final Phaser phaser = new Phaser(count) {
8. @Override
9. protected boolean onAdvance(int phase, int registeredParties) {
10. System.out.println("====== " + phase + " ======");
11. return phase == phaseToTerminate || registeredParties == 0;
12. }
13. };
14. //
15. for(int i = 0; i < count; i++) {
16. System.out.println("starting thread, id: " + i);
17. final Thread thread = new Thread(new Task(i, phaser));
18. thread.start();
19. }
20. //
21. phaser.register();
22. while (!phaser.isTerminated()) {
23. phaser.arriveAndAwaitAdvance();
24. }
25. System.out.println("done");
26. }
27. public static class Task implements Runnable {
28. //
29. private final int id;
30. private final Phaser phaser;
31. public Task(int id, Phaser phaser) {
32. this.id = id;
33. this.phaser = phaser;
34. }
35. @Override
36. public void run() {
37. while(!phaser.isTerminated()) {
38. try {
39. Thread.sleep(500);
40. } catch(InterruptedException e) {
41. // NOP
42. }
43. System.out.println("in Task.run(), phase: " + phaser.getPhase() + ", id: " + this.id);
44. phaser.arriveAndAwaitAdvance();
45. }
46. }
47. }
48. }

如果希望主线程在特定的phase结束之后终止，那么可以在主线程中调用下述方法：

Java代码  

1. public static void awaitPhase(Phaser phaser, int phase) {
2. int p = phaser.register(); // assumes caller not already registered
3. while (p < phase) {
4. if (phaser.isTerminated()) {
5. break; // ... deal with unexpected termination
6. } else {
7. p = phaser.arriveAndAwaitAdvance();
8. }
9. }
10. phaser.arriveAndDeregister();
11. }

需要注意的是，awaitPhase方法中的if (phaser.isTerminated()) 分支里需要能够正确处理Phaser终止的情况。否则由于在Phaser终止之后， phaser.register()和arriveAndAwaitAdvance()方法均返回负值，那么上述方法可能陷入死循环。

样例5

以下对Phaser进行分层的例子：

Java代码  

1. import java.util.concurrent.Phaser;
2. public class PhaserTest6 {
3. //
4. private static final int TASKS_PER_PHASER = 4;
5. public static void main(String args[]) throws Exception {
6. //
7. final int phaseToTerminate = 3;
8. final Phaser phaser = new Phaser() {
9. @Override
10. protected boolean onAdvance(int phase, int registeredParties) {
11. System.out.println("====== " + phase + " ======");
12. return phase == phaseToTerminate || registeredParties == 0;
13. }
14. };
15. //
16. final Task tasks[] = new Task[10];
17. build(tasks, 0, tasks.length, phaser);
18. for (int i = 0; i < tasks.length; i++) {
19. System.out.println("starting thread, id: " + i);
20. final Thread thread = new Thread(tasks[i]);
21. thread.start();
22. }
23. }
24. public static void build(Task[] tasks, int lo, int hi, Phaser ph) {
25. if (hi - lo > TASKS_PER_PHASER) {
26. for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
27. int j = Math.min(i + TASKS_PER_PHASER, hi);
28. build(tasks, i, j, new Phaser(ph));
29. }
30. } else {
31. for (int i = lo; i < hi; ++i)
32. tasks[i] = new Task(i, ph);
33. }
34. }
35. public static class Task implements Runnable {
36. //
37. private final int id;
38. private final Phaser phaser;
39. public Task(int id, Phaser phaser) {
40. this.id = id;
41. this.phaser = phaser;
42. this.phaser.register();
43. }
44. @Override
45. public void run() {
46. while (!phaser.isTerminated()) {
47. try {
48. Thread.sleep(200);
49. } catch (InterruptedException e) {
50. // NOP
51. }
52. System.out.println("in Task.run(), phase: " + phaser.getPhase()    + ", id: " + this.id);
53. phaser.arriveAndAwaitAdvance();
54. }
55. }
56. }
57. }

需要注意的是，TASKS_PER_PHASER的值取决于具体的Task实现。对于Task执行时间很短的场景（也就是竞争相对激烈），可以考虑使用较小的TASKS_PER_PHASER值，例如4。反之可以适当增大TASKS_PER_PHASER。