Java中的Phaser

JDK1.7新特性

可重用的同步屏障，其功能類似于CyclicBarrier和CountDownLatch，但支持更靈活的用法。

• Registration：與其他障礙不同，注冊在phaser上進行同步的parties數(shù)量可能會隨時間變化。任務(wù)可以隨時進行注冊（使用方法register，bulkRegister或建立初始方數(shù)目的構(gòu)造函數(shù)形式），也可以選擇在任何到達時注銷（使用ArcadeAndDeregister）。與大多數(shù)基本同步結(jié)構(gòu)一樣，registration和deregistration會只影響內(nèi)部計數(shù)。它們不會建立任何進一步的內(nèi)部簿記，因此任務(wù)無法查詢它們是否已注冊。 （但是，可以通過通過子類來引入這種記帳方法。）
• Synchronization：像CyclicBarrier一樣，Phaser可能會反復(fù)等待。arriveAndAwaitAdvance方法的作用類似于CyclicBarrier.await。每一代的phaser都有一個關(guān)聯(lián)的相數(shù)。相數(shù)從零開始，并在所有各方到達移相器時被遞增，在達到Integer.MAX_VALUE后回繞為零。通過使用任何注冊方可以調(diào)用的兩種方法，階段編號的使用可以在到達移相器時以及在等待其他人時獨立控制動作。
• Arrival：arrive和arriveAndDeregister方法記錄到達。這些方法不會阻塞，但是會返回一個關(guān)聯(lián)的到達phase數(shù)。也就是說，到達應(yīng)用到的相位器的phase數(shù)。當給定階段的最終參與者到達時，將執(zhí)行可選操作，并且階段將前進。這些動作由觸發(fā)相位超前的一方執(zhí)行，并由覆蓋方法onAdvance（int，int）安排，該方法也控制終止。重寫此方法類似于為CyclicBarrier提供屏障操作，但比它更具靈活性。
• Waiting：方法awaitAdvance需要一個指示到達階段編號的參數(shù)，并在移相器前進到（或已經(jīng)處于）另一個階段時返回。與使用CyclicBarrier的類似構(gòu)造不同，即使等待線程被中斷，方法awaitAdvance仍繼續(xù)等待。也提供中斷版本和超時版本，但是任務(wù)中斷或超時等待時遇到的異常不會更改相位器的狀態(tài)。如有必要，通?？梢栽谡{(diào)用forceTermination之后，在這些異常的處理程序內(nèi)執(zhí)行任何關(guān)聯(lián)的恢復(fù)。在ForkJoinPool中執(zhí)行的任務(wù)也可以使用相位器，當其他任務(wù)被阻止等待階段前進時，它將確保足夠的并行度來執(zhí)行任務(wù)。
• Termination：移相器可能會進入終止狀態(tài)，可以使用方法isTerminated檢查該狀態(tài)。終止后，所有同步方法將立即返回，而無需等待提前，如負的返回值所示。同樣，終止注冊的嘗試也無效。當onAdvance的調(diào)用返回true時，將觸發(fā)終止。如果注銷導(dǎo)致注冊方的數(shù)量變?yōu)榱悖瑒t默認實現(xiàn)返回true。如下所示，當相位器以固定的迭代次數(shù)控制動作時，通常方便的方法是在當前相位數(shù)達到閾值時覆蓋此方法以終止。方法forceTermination也可用于突然釋放等待線程并允許它們終止。
• Tiering：hasers可以分層（即，以樹形結(jié)構(gòu)構(gòu)建）以減少競爭。相反，可以設(shè)置具有大量參與方的相位器，否則它們將承受沉重的同步競爭成本，以便子相位器組共享一個公共父級。即使這會招致更大的每次操作開銷，也可能會大大提高吞吐量。在分層的相位器樹中，將自動管理子相位器及其父級的注冊和注銷。只要子相位器的注冊方數(shù)量不為零（在Phaser（Phaser，int）構(gòu)造函數(shù)，寄存器或bulkRegister中確定），子相位器就會向其父級注冊。每當調(diào)用到達和取消注冊的結(jié)果而導(dǎo)致注冊方的數(shù)量為零時，子相位器都會從其父級注銷。
• Monitoring：雖然同步方法只能由注冊方調(diào)用，但是相位器的當前狀態(tài)可以由任何調(diào)用者監(jiān)視。在任何給定時刻，總共有g(shù)etRegisteredParties參與者，其中g(shù)etArrivedParties已到達當前階段（getPhase）。當其余（getUnarrivedParties）方到達時，該階段前進。這些方法返回的值可能反映瞬態(tài)，因此通常對于同步控制沒有用。方法toString以便于非正式監(jiān)視的形式返回這些狀態(tài)查詢的快照。

用法示例

可以使用Phaser代替CountDownLatch來控制為可變方提供服務(wù)的單發(fā)操作。典型的習(xí)慣用法是將方法設(shè)置為首先注冊，然后開始操作，然后注銷，如下所示：

 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();
 }

使一組線程針對給定的迭代次數(shù)重復(fù)執(zhí)行操作的一種方法是重寫onAdvance：

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
 }

如果主要任務(wù)以后必須等待終止，則它可以重新注冊，然后執(zhí)行類似的循環(huán)：

// ...
phaser.register();
while (!phaser.isTerminated())
  phaser.arriveAndAwaitAdvance();

在確保相位永遠不會環(huán)繞Integer.MAX_VALUE的情況下，可以使用相關(guān)的構(gòu)造來等待特定的相位編號。例如：

 void awaitPhase(Phaser phaser, int phase) {
   int p = phaser.register(); // assumes caller not already registered
   while (p < phase) {
     if (phaser.isTerminated())
       // ... deal with unexpected termination
     else
       p = phaser.arriveAndAwaitAdvance();
   }
   phaser.arriveAndDeregister();
 }

要使用移相器樹創(chuàng)建一組n個任務(wù)，可以使用以下形式的代碼，假設(shè)Task類的構(gòu)造函數(shù)接受在構(gòu)造時注冊的Phaser。在調(diào)用build（new Task [n]，0，n，new Phaser（））之后，可以啟動這些任務(wù)，例如通過提交到池中：

 void build(Task[] tasks, int lo, int hi, Phaser ph) {
   if (hi - lo > TASKS_PER_PHASER) {
     for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
       int j = Math.min(i + TASKS_PER_PHASER, hi);
       build(tasks, i, j, new Phaser(ph));
     }
   } else {
     for (int i = lo; i < hi; ++i)
       tasks[i] = new Task(ph);
       // assumes new Task(ph) performs ph.register()
   }
 }

TASKS_PER_PHASER的最佳值主要取決于預(yù)期的同步速率。對于極小的每階段任務(wù)主體（因此是高比率），低至4的值可能合適，而對于極大型的任務(wù)階段，則最高可能為數(shù)百。實施注意事項：此實施將參與方的最大數(shù)量限制為65535。嘗試注冊其他參與方會導(dǎo)致IllegalStateException。但是，您可以并且應(yīng)該創(chuàng)建分層的相位器，以容納任意數(shù)量的參與者。

構(gòu)造方法和公共方法

int arriveAndAwaitAdvance()     //當前線程到達屏障后等待滿足條件后再向下一個屏障執(zhí)行，當計數(shù)不滿足時線程會一直阻塞
int arriveAndDeregister()       //使當前線程退出，并使parties值減1
int getPhase()                  //獲取已滿足多少次屏障
protected boolean onAdvance(int phase, int registeredParties)  //通過新的屏障時被調(diào)用，在創(chuàng)建Phaser時子類實現(xiàn)邏輯，返回true使phaser失效
int getRegisteredParties()      //獲取注冊的parties數(shù)量
int register()                 // 每一次調(diào)用就動態(tài)遞增一個parties
int bulkRegister(int parties)   //多個添加parties
int getArrivedParties()        // 獲取已被使用的parties個數(shù)
 int getUnarrivedParties()     // 獲取未被使用的parties個數(shù)
 
int arrive()                    //使parties個數(shù)加一，并且當前線程不用等待其他線程到達屏障；當其他線程計數(shù)不足，依然會處于等待狀態(tài)
int awaitAdvance(int phase)     //如果入?yún)hase值和當前getPhase()方法值一致，則等待屏障，否則忽略屏障，當前線程不可被中斷
int awaitAdvanceInterruptibly(int phase) //當前線程可被中斷，當前的parties不滿足入?yún)r跳過屏障
int awaitAdvanceInterruptibly(int phase,long timeout, TimeUnit unit) //當前線程可被中斷，當前的parties不滿足入?yún)r跳過屏障或者超過指定時間
void forceTermination()         //使phaser屏障功能失效
 boolean isTerminated()         //判斷phaser對象是否呈銷毀狀態(tài)  

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

最新評論