Java多线程---Phaser

Phaser是Java7新引入的并发API，我们可以将Phaser的概念看成是一个个的阶段，每个阶段都需要执行的线程任务，任务执行完毕后就进入下一阶段。这里和CyclicBarrier 和CountDownLatch的概念类似，实际上也确实可以用Phaser代替CyclicBarrier和CountDownLatch。

Phaser也是通过计数器来控制，在Phaser中叫parties，我们在指定了parties之后， Phaser可以根据需要动态增加或减少parties的值。

Phaser常用API介绍：

构造方法： public Phaser() {this(null, 0);}public Phaser(int parties) {this(null, parties);}public Phaser(Phaser parent) {this(parent, 0);}public Phaser(Phaser parent, int parties) {if (parties >>> PARTIES_SHIFT != 0)throw new IllegalArgumentException("Illegal number of parties");int phase = 0;this.parent = parent;if (parent != null) {final Phaser root = parent.root;this.root = root;this.evenQ = root.evenQ;this.oddQ = root.oddQ;if (parties != 0)phase = parent.doRegister(1);}else {this.root = this;this.evenQ = new AtomicReference<QNode>();this.oddQ = new AtomicReference<QNode>();}this.state = (parties == 0) ? (long)EMPTY :((long)phase << PHASE_SHIFT) |((long)parties << PARTIES_SHIFT) |((long)parties);}

从上面可以看出， Pahser是存在上下级关系的树状结构，如果不指定parent和parties默认为null和0。后面我们再对parent结构进行说明。先看看它常用API和对应的功能。

public int register() //向此移相器添加一个新的未到达方。public int bulkRegister(int parties)  // 将给定数量的新未到达方添加到此移相器。public int arriveAndDeregister() // 到达此移相器并从其注销，而无需等待其他人到达public int arrive()  // 到达此移相器，无需等待其他人到达public int arriveAndAwaitAdvance()  // 到达这个移相器并等待其他人public int awaitAdvance(int phase)  // 等待此移相器的相位从给定的相位值前进，如果当前相位不等于给定的相位值或此移相器终止，则立即返回。public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)  // 等待此移相器的相位从给定的相位值或给定的超时时间过去，InterruptedException如果在等待时中断则抛出，或者如果当前相位不等于给定的相位值或此移相器终止，则立即返回public final int getPhase() // 返回当前阶段号public int getRegisteredParties()  // 返回在此移相器上注册的参与方数量public int getArrivedParties()  // 返回已到达此移相器当前阶段的已注册方的数量public int getUnarrivedParties()  // 返回尚未到达此移相器当前阶段的已注册方的数量。

这里列出了Phaser的部分API和功能说明。下面我们就看一下这些API的具体用法。

1、Phaser代替CountDownLatch： arrive、awaitAdvance

    public static void main(String[] args) {Phaser phaser = new Phaser(5);for (int i = 0; i < 5; i++) {new Thread(() -> {synchronized (phaser) {System.out.println(Thread.currentThread().getName() + "running");phaser.arrive();}}, "thread" + i).start();}System.out.println(Thread.currentThread().getName() + " wait");phaser.awaitAdvance(phaser.getPhase());System.out.println(Thread.currentThread().getName() + "线程执行完毕");}

这里要注意， arrive方法不是线程安全的，如果不加锁，方法很难结束。

输出结果：

这样无论执行多少次， “ main线程执行完毕 ” 这句话都是最后才输出。我们可以尝试一下，将phaser.arrive() 的调用时机提前，看看方法的执行结果。

2、用phaser代替CyclicBarrier： arriveAndAwaitAdvance

Phaser phaser = new Phaser(5);for(int i=0; i<5; i++){new Thread(()->{System.out.println("=========" + Thread.currentThread().getName());phaser.arriveAndAwaitAdvance();System.out.println("***" + Thread.currentThread().getName());phaser.arriveAndAwaitAdvance();System.out.println("##########" + Thread.currentThread().getName());}).start();}

输出结果：

3.phaser其他API演示：

public static void main(String[] args) {Phaser phaser = new Phaser(5);for(int i=0; i<5; i++){new Thread(()->{synchronized (phaser){phaser.arrive();System.out.println("phaser.getPhase() = "+phaser.getPhase());System.out.println("phaser.getArrivedParties() = "+phaser.getArrivedParties());System.out.println("phaser.getUnarrivedParties() = "+phaser.getUnarrivedParties());}if(Thread.currentThread().getName().equals("thread4")){try {Thread.sleep(1000);System.out.println("=======phaser.getPhase() = "+phaser.getPhase());phaser.arriveAndDeregister();System.out.println("========phaser.arriveAndDeregister() = "+phaser.getPhase());} catch (InterruptedException e) {e.printStackTrace();}}}, "thread"+i).start();}phaser.bulkRegister(3);}

执行结果：