Java中的Phaser

JDK1.7新特性

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

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

在確保相位永遠(yuǎn)不會環(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()     //當(dāng)前線程到達(dá)屏障后等待滿足條件后再向下一個屏障執(zhí)行，當(dāng)計數(shù)不滿足時線程會一直阻塞
int arriveAndDeregister()       //使當(dāng)前線程退出，并使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ù)加一，并且當(dāng)前線程不用等待其他線程到達(dá)屏障；當(dāng)其他線程計數(shù)不足，依然會處于等待狀態(tài)
int awaitAdvance(int phase)     //如果入?yún)hase值和當(dāng)前getPhase()方法值一致，則等待屏障，否則忽略屏障，當(dāng)前線程不可被中斷
int awaitAdvanceInterruptibly(int phase) //當(dāng)前線程可被中斷，當(dāng)前的parties不滿足入?yún)r跳過屏障
int awaitAdvanceInterruptibly(int phase,long timeout, TimeUnit unit) //當(dāng)前線程可被中斷，當(dāng)前的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)文章希望大家以后多多支持腳本之家！

最新評論