前面的文章中我們講到了CyclicBarrier、CountDownLatch的使用，這里再回顧一下CountDownLatch主要用在一個(gè)線程等待多個(gè)線程執(zhí)行完畢的情況，而CyclicBarrier用在多個(gè)線程互相等待執(zhí)行完畢的情況。

Phaser是java 7 引入的新的并發(fā)API。他引入了新的Phaser的概念，我們可以將其看成一個(gè)一個(gè)的階段，每個(gè)階段都有需要執(zhí)行的線程任務(wù)，任務(wù)執(zhí)行完畢就進(jìn)入下一個(gè)階段。所以Phaser特別適合使用在重復(fù)執(zhí)行或者重用的情況。

基本使用

在CyclicBarrier、CountDownLatch中，我們使用計(jì)數(shù)器來(lái)控制程序的順序執(zhí)行，同樣的在Phaser中也是通過(guò)計(jì)數(shù)器來(lái)控制。在Phaser中計(jì)數(shù)器叫做parties， 我們可以通過(guò)Phaser的構(gòu)造函數(shù)或者register()方法來(lái)注冊(cè)。

通過(guò)調(diào)用register()方法，我們可以動(dòng)態(tài)的控制phaser的個(gè)數(shù)。如果我們需要取消注冊(cè)，則可以調(diào)用arriveAndDeregister()方法。

我們看下arrive：

 public int arrive() {
  return doArrive(ONE_ARRIVAL);
 }

Phaser中arrive實(shí)際上調(diào)用了doArrive方法，doArrive接收一個(gè)adjust參數(shù)，ONE_ARRIVAL表示arrive，ONE_DEREGISTER表示arriveAndDeregister。

Phaser中的arrive()、arriveAndDeregister()方法，這兩個(gè)方法不會(huì)阻塞，但是會(huì)返回相應(yīng)的phase數(shù)字，當(dāng)此phase中最后一個(gè)party也arrive以后，phase數(shù)字將會(huì)增加，即phase進(jìn)入下一個(gè)周期，同時(shí)觸發(fā)（onAdvance）那些阻塞在上一phase的線程。這一點(diǎn)類(lèi)似于CyclicBarrier的barrier到達(dá)機(jī)制；更靈活的是，我們可以通過(guò)重寫(xiě)onAdvance方法來(lái)實(shí)現(xiàn)更多的觸發(fā)行為。

下面看一個(gè)基本的使用：

 void runTasks(List<Runnable> tasks) {
  final Phaser phaser = new Phaser(1); // "1" to register self
  // create and start threads
  for (final Runnable task : tasks) {
   phaser.register();
   new Thread() {
    public void run() {
     phaser.arriveAndAwaitAdvance(); // await all creation
     task.run();
    }
   }.start();
  }

  // allow threads to start and deregister self
  phaser.arriveAndDeregister();
 }

上面的例子中，我們?cè)趫?zhí)行每個(gè)Runnable之前調(diào)用register（）來(lái)注冊(cè)， 然后調(diào)用arriveAndAwaitAdvance()來(lái)等待這一個(gè)Phaser周期結(jié)束。最后我們調(diào)用 phaser.arriveAndDeregister();來(lái)取消注冊(cè)主線程。

多個(gè)Phaser周期

Phaser的值是從0到Integer.MAX_VALUE，每個(gè)周期過(guò)后該值就會(huì)加一，如果到達(dá)Integer.MAX_VALUE則會(huì)繼續(xù)從0開(kāi)始。

如果我們執(zhí)行多個(gè)Phaser周期，則可以重寫(xiě)onAdvance方法：

 protected boolean onAdvance(int phase, int registeredParties) {
  return registeredParties == 0;
 }

onAdvance將會(huì)在最后一個(gè)arrive（）調(diào)用的時(shí)候被調(diào)用，如果這個(gè)時(shí)候registeredParties為0的話，該P(yáng)haser將會(huì)調(diào)用isTerminated方法結(jié)束該P(yáng)haser。

如果要實(shí)現(xiàn)多周期的情況，我們可以重寫(xiě)這個(gè)方法：

protected boolean onAdvance(int phase, int registeredParties) {
    return phase >= iterations || registeredParties == 0;
   }

上面的例子中，如果phase次數(shù)超過(guò)了指定的iterations次數(shù)則就會(huì)自動(dòng)終止。

我們看下實(shí)際的例子：

 void startTasks(List<Runnable> tasks, final int iterations) {
  final Phaser phaser = new Phaser() {
   protected boolean onAdvance(int phase, int registeredParties) {
    return phase >= iterations || registeredParties == 0;
   }
  };
  phaser.register();
  for (final Runnable task : tasks) {
   phaser.register();
   new Thread() {
    public void run() {
     do {
      task.run();
      phaser.arriveAndAwaitAdvance();
     } while (!phaser.isTerminated());
    }
   }.start();
  }
  phaser.arriveAndDeregister(); // deregister self, don't wait
 }

上面的例子將會(huì)執(zhí)行iterations次。

本文的例子請(qǐng)參考https://github.com/ddean2009/learn-java-concurrency/tree/master/Phaser

到此這篇關(guān)于java多線程之Phaser的使用的文章就介紹到這了,更多相關(guān)java多線程Phaser內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論