探索JDK7的并发编程——PHASER

JDK7对并发编程提供了更好的支持，包括ThreadLocalRandom、ForkJoinPool、TransferQueue以及本文要讲述的Phaser。

JDK5中引入了CyclicBarrier和CountDownLatch这两个并发控制类，而JDK7中引入的Phaser按照官方的说法是提供了一个功能类似但是更加灵活的实现。接下来我们研究一下Phaser与两个位辈到底有哪些类似，同时带来了哪些灵活性。

CyclicBarrier和CountDownLatch
在使用CyclicBarrier时，需要创建一个CyclicBarrier对象，构造函数需要一个整数作为参数，这个参数是一个“目标”，在CyclicBarrier对象创建后，内部会有一个计数器，初始值为0，CyclicBarrier对象的await方法没被调用一次，这个计数器就会加1，一旦这个计数器的值达到设定的“目标”，所有被CyclicBarrier.await阻塞住的线程都会继续执行。这个目标是固定的，一旦设定便不能修改。
举一个形象的例子，假设有5个人爬山，他们要爬到山顶，等到5个人到期了在同时出发下山，那么我们要在山顶设定一个“目标”，同时还有一个计数器，这个目标就是5，每到山顶一个人，这个人就要等待，同时计数器加1，等到5个人到齐了，也就是计数器达到了这个“目标”，所有等待的人就开始下山了。

使用CountDownLatch时，需要创建一个CountDownLatch对象，构造函数也需要一个整数作为参数，可以把这个参数想象成一个倒计时器，CountDownLatch对象本身是一个发令枪，所有调用CountDownLatch.await方法的线程都会等待发令枪的指令，一旦倒计时器为0，这些线程同时开始执行，而CountDownLatch.countDown方法就是为倒计时器减1。

CyclicBarrier和CountDownLatch的共同点都是有一个目标和一个计数器，等到计数器达到目标后，所有阻塞的线程都将继续执行。他们的不同点是CyclicBarrier.await在等待的同时还修改计数器，而CountDownLatch.await只负责等待，CountDownLatch.countDown才修改计数器，这是他们最本质的区别（千万不要告诉我他们的区别是一个加1一个减1！这就像数数，从1数到10和从10数到1都是数10个数，没有区别的）。

接下来我们分析一下JDK7新加入的Phaser。

Phaser
Phaser同时包含CyclicBarrier和CountDownLatch两个类的功能。Phaser的arrive方法将将计数器加1，awaitAdvance将线程阻塞，直到计数器达到目标，这两个方法与CountDownLatch的countDown和await方法相对应；Phaser的arriveAndAwaitAdvance方法将计数器加1的同时将线程阻塞，直到计数器达到目标后继续执行，这个方法对应CyclicBarrier的await方法。

除了包含以上两个类的功能外，Phaser还提供了更大的灵活性。CyclicBarrier和CountdownLatch在构造函数指定目标后就无法修改，而Phaser提供了register和deregister方法可以对目标进行动态修改。

以下提供一组例子，第一个例子展示了上文中提到的Phaser的方法，第二第三个例子展示了Phaser适用的两个场景，尤其是第三个，这个场景在没有Phaser是比较不容易实现的：

package com.lunarever.study.java7test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Phaser;
import java.util.logging.Level;
import java.util.logging.Logger;

public class PhraseTest {
     
     public static void main(String[] args) throws Exception {
          demo1();
          demo2();
          demo3();
     }
     
     public static void demo1() throws Exception {
          final Phaser phaser = new Phaser(3) {
               @Override
               protected boolean onAdvance(int phase, int registeredParties) {
                    System.out.printf("---> run here: onAdvance --> phase=%d, parties=%d%n", phase, registeredParties);
                    return super.onAdvance(phase, registeredParties);
               }
          };
          new Thread() {
               @Override
               public void run() {
                    System.out.println("---> run here: awaitAdvance 0 start");
                    phaser.awaitAdvance(0);
                    System.out.println("---> run here: awaitAdvance 0 end");
               }
          }.start();
          new Thread() {
               @Override
               public void run() {
                    System.out.println("---> run here: awaitAdvance 1 start");
                    phaser.awaitAdvance(1);
                    System.out.println("---> run here: awaitAdvance 1 end");
               }
          }.start();
          new Thread() {
               @Override
               public void run() {
                    System.out.println("---> run here: arriveAndAwaitAdvance start");
                    phaser.arriveAndAwaitAdvance();
                    System.out.println("---> run here: arriveAndAwaitAdvance end");
               }
          }.start();
          Thread.sleep(200);
          System.out.println("---> arriveAndDeregister 1");
          System.out.println("---> arriveAndDeregister 1 -> " + phaser.arriveAndDeregister());
          Thread.sleep(200);
          System.out.println("---> arrive 1");
          System.out.println("---> arrive 1 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> register 1");
          phaser.register();
          Thread.sleep(200);
          System.out.println("---> arrive 2");
          System.out.println("---> arrive 2 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> arrive 3");
          System.out.println("---> arrive 3 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> arrive 4");
          System.out.println("---> arrive 4 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> arrive 5");
          System.out.println("---> arrive 5 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> arrive 6");
          System.out.println("---> arrive 6 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> arrive 7");
          System.out.println("---> arrive 7 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> arrive 8");
          System.out.println("---> arrive 8 -> " + phaser.arrive());
          Thread.sleep(200);
          System.out.println("---> arrive 9");
          System.out.println("---> arrive 9 -> " + phaser.arrive());
     }

     public static void demo2() {
          final List<Runnable> tasks = new ArrayList<Runnable>();
          for (int i = 0; i < 3; i++) {
               final int counter = i;
               tasks.add(new Runnable() {
                    @Override
                    public void run() {
                         System.out.println("---> " + counter);
                         try {
                              Thread.sleep(200);
                         } catch (InterruptedException ex) {
                              Logger.getLogger(PhraseTest.class.getName()).log(Level.SEVERE, null, ex);
                         }
                    }
               });
          }
          final Phaser phaser = new Phaser(1); // "1" to register self
          // create and start threads
          for (final Runnable task : tasks) {
               phaser.register();
               new Thread() {
                    @Override
                    public void run() {
                         phaser.arriveAndAwaitAdvance(); // await all creation
                         task.run();
                    }
               }.start();
          }
          phaser.arriveAndDeregister(); // allow threads to start and deregister self
     }

     public static void demo3() {
          final List<Runnable> tasks = new ArrayList<Runnable>();
          for (int i = 0; i < 3; i++) {
               final int counter = i;
               tasks.add(new Runnable() {
                    @Override
                    public void run() {
                         System.out.println("---> " + counter);
                         try {
                              Thread.sleep(200);
                         } catch (InterruptedException ex) {
                              Logger.getLogger(PhraseTest.class.getName()).log(Level.SEVERE, null, ex);
                         }
                    }
               });
          }
          final Phaser phaser = new Phaser() {
               @Override
               protected boolean onAdvance(int phase, int registeredParties) {
                    return phase >= 5 || registeredParties == 0;
               }
          };
          phaser.register();
          for (final Runnable task : tasks) {
               phaser.register();
               new Thread() {
                    @Override
                    public void run() {
                         do {
                              task.run();
                              phaser.arriveAndAwaitAdvance();
                         } while (!phaser.isTerminated());
                    }
               }.start();
          }
          phaser.arriveAndDeregister(); // deregister self, don't wait
     }
}

评论

1 楼 deepnighttwo 2011-08-10  

没装Java7，没试用，但是看文章写的还不错，挺清晰明了的，为啥这么多人踩捏。。。