neadam ·移相器 Phaser
1. 前言
本節帶領大家認識第四個常用的 Java 并發工具類之 Phaser。
本節先介紹 Phaser 工具類表達的概念和最基本用法，接著通過一個生活中的例子為大家解釋 Phaser 工具類的使用場合，然后通過簡單的編碼實現此場景，最后帶領大家熟悉 Phaser 工具類的其他重要方法。
下面我們正式開始介紹吧。
2. 概念解釋
Phaser 表示 “階段器”，一個可重用的同步 barrier，與 CyclicBarrier 相比，Phaser 更靈活，而且側重于 “重用”。Phaser 中允許 “注冊的同步者（parties）” 隨時間而變化。Phaser 可以通過構造器初始化 parties 個數，也可以在 Phaser 運行期間隨時加入新的 parties，以及在運行期間注銷 parties。
是不是又強大又抽象，沒關系，我們通過一張圖可以直白了解其提供的邏輯模型。
概念已經了解了，Phaser 工具類最基本的用法是怎樣的呢？看下面。
3. 基本用法
// 創建一個 Phaser 對象
Phaser phaser = new Phaser();
// 將線程 m 作為同步者之一進行同步控制注冊
phaser.register();
// 線程 m 開始處理邏輯
...
// 線程 m 等待同一周期內，其他線程到達，然后進入新的周期，并繼續同步進行
phaser.arriveAndAwaitAdvance();
...
// 線程 m 執行完畢后做同步控制注銷
phaser.arriveAndDeregister();
這個工具類相對而言比較復雜，大家不要著急，結合后面的案例仔細體會。Phaser 應用在哪些場合比較合適呢？下面我們給出最常用的場景說明。
4. 常用場景
Phaser 適合用于具有多階段處理的任務，在每個階段有多個線程并行處理的場景。這樣描述很抽象，我們舉一個生活中的例子：有一個開發小組總共 4 個人，約定一起去旅游。計劃一起出發，先去景點 A 自有活動，3 個小時后去景點 B 自有活動，2 個小時候后活動結束統一集合。這個場景中 4 個人相當于 4 個線程，分了 4 個階段完成了整個計劃。像類似這樣的場景很適合用 Phaser 解決。請看下面代碼。
5. 場景案例
public class PhaserTest {
// 先構建一個階段器對象
private static TravelPhaser travelPhaser = new TravelPhaser();
// 主邏輯
public static void main(String[] args) throws InterruptedException {
// 創建 5 個線程代表每一位同事
for (int i = 1; i < 5; i++) {
// 對每一個需要同步控制的線程進行同步控制注冊
travelPhaser.register();
// 模擬每一位同事開始旅游行動
Thread thread = new Thread(new Colleague(travelPhaser), "同事" + i);
thread.start();
}
}
}
上述代碼在注冊好需要同步控制的所有線程之后，開啟了每一個線程（每位同事）的處理。每一個線程（每位同事）如何行動呢，代碼如下：
import java.util.Random;
/**
* 模擬人以及旅游的各類狀態
*/
public class Colleague implements Runnable {
private TravelPhaser travelPhaser;
public Colleague(TravelPhaser travelPhaser) {
this.travelPhaser = travelPhaser;
}
/**
* 模擬每位同事的動作
*/
@Override
public void run() {
doAnything();
System.out.println(Thread.currentThread().getName() + "到達出發集合地");
travelPhaser.arriveAndAwaitAdvance();
doAnything();
System.out.println(Thread.currentThread().getName() + "已經在景點 A 自由活動結束");
travelPhaser.arriveAndAwaitAdvance();
doAnything();
System.out.println(Thread.currentThread().getName() + "已經在景點 B 自由活動結束");
travelPhaser.arriveAndAwaitAdvance();
doAnything();
System.out.println(Thread.currentThread().getName() + "到達返程集合地");
travelPhaser.arriveAndAwaitAdvance();
}
/**
* 模擬用時
*/
private void doAnything() {
try {
Thread.sleep(new Random().nextInt(10000));
} catch (Exception e) {}
}
}
上述代碼模擬了每位同事的旅游過程。代碼中使用了 arriveAndAwaitAdvance () 進行每個旅游階段的控制。我們再接著看對旅游各個階段的自定義控制：
import java.util.concurrent.Phaser;
/**
* 對每一個階段進行自定義控制
*/
public class TravelPhaser extends Phaser {
protected boolean onAdvance(int phase, int registeredParties) {
switch (phase) {
// 第1階段，旅游前的集合
case 0:
System.out.println("出發前小組人員集合完畢，總人數："+getRegisteredParties());
return false;
// 第2階段，景點 A 游玩
case 1:
System.out.println("景點 A 游玩結束");
return false;
// 第3階段，景點 B 游玩
case 2:
System.out.println("景點 B 游玩結束");
return false;
// 第4階段，旅游結束返程集合
case 3:
System.out.println("所有活動結束后小組人員集合完畢，總人數："+getRegisteredParties());
return true;
default:
return true;
}
}
}
上述代碼只是在各個階段打印了一些描述信息，實際中可以做更多的邏輯控制。運行上面代碼，我們觀察一下運行結果。
同事1到達出發集合地
同事4到達出發集合地
同事2到達出發集合地
同事3到達出發集合地
出發前小組人員集合完畢，總人數：4
同事3已經在景點 A 自由活動結束
同事2已經在景點 A 自由活動結束
同事1已經在景點 A 自由活動結束
同事4已經在景點 A 自由活動結束
景點 A 游玩結束
同事4已經在景點 B 自由活動結束
同事2已經在景點 B 自由活動結束
同事1已經在景點 B 自由活動結束
同事3已經在景點 B 自由活動結束
景點 B 游玩結束
同事2到達返程集合地
同事3到達返程集合地
同事1到達返程集合地
同事4到達返程集合地
所有活動結束后小組人員集合完畢，總人數：4
觀察結果，和我們的預期一致。注意體會 Phaser 提供的多線程共同協作的模型。
6. 其他方法介紹
除過上面代碼中使用的最基本的 register ()、arriveAndAwaitAdvance ()、arriveAndDeregister ()、getRegisteredParties () 方法之外，還有下面幾個方法大家可以了解一下。
-
awaitAdvance (int phase) 方法。
具有阻塞功能，等待 phase 周期數下其他所有的 parties 都到達后返回。如果指定的 phase 與當前的 phase 不一致，則立即返回。 -
awaitAdvanceInterruptibly (int phase) 方法。
同 awaitAdvance 類似，但支持中斷響應，即 waiter 線程如果被外部中斷，則此方法立即返回。 -
forceTermination () 方法。
用于強制終止 phase，此后 Phaser 對象將不可用，即 register 等將不再有效。
7. 小結
本節介紹了 Phaser 基本概念原理，并且通過一個簡單的例子，介紹了其使用場景和基本用法。希望大家在學習過程中，多思考勤練習，早日掌握之。
2025 imooc.com All Rights Reserved |