原始碼分析：Phaser 之更靈活的同步屏障
阿新 • • 發佈：2021-02-22
## 簡介
`Phaser` 是 JDK 1.7 開始提供的一個可重複使用的同步屏障，功能類似於`CyclicBarrier`和`CountDownLatch`，但使用更靈活，支援對任務的動態調整，並支援分層結構來達到更高的吞吐量。
### Registration（註冊）
與其他屏障的情況不同，在 `Phaser` 上註冊同步的參與方的數量可能隨時間而變化。任務可以在任何時候註冊（使用方法`register`、`bulkRegister`或建立初始參與方數量的建構函式），可以在任何到達時取消註冊（使用`arriveAndDeregister`），註冊和登出隻影響內部計數，任務無法查詢它們是否已註冊。
### Synchronization（同步）
像`CyclicBarrier`，`Phaser`也可以重複`await`。方法`arriveAndAwaitAdvance()`有效果類似於`CyclicBarrier.await` 。phaser的每一代都有一個相關的phase number，初始值為0，當所有註冊的任務都到達phaser時phase+1，到達最大值(Integer.MAX_VALUE)之後清零。使用phase number可以獨立控制到達phaser 和 等待其他執行緒 的動作，通過下面兩種型別的方法:
- Arrival(到達機制)
`arrive`和`arriveAndDeregister`方法記錄到達狀態。
這些方法不會阻塞，但是會返回一個相關的arrival phase number；也就是說，phase number用來確定到達狀態。當所有任務都到達給定phase時，可以執行一個可選的函式，這個函式通過重寫`onAdvance`方法實現，通常可以用來控制終止狀態。
重寫此方法類似於為`CyclicBarrier`提供一個`barrierAction`（執行的命令執行緒），但比它更靈活。
- Waiting(等待機制)
`awaitAdvance`方法需要一個表示 arrival phase number 的引數，並且在phaser前進到與給定phase不同的phase時返回。和`CyclicBarrier`不同，即使等待執行緒已經被中斷，`awaitAdvance`方法也會一直等待。中斷狀態和超時時間同樣可用，但是當任務等待中斷或超時後未改變phaser的狀態時會遭遇異常。如果有必要，在方法`forceTermination`之後可以執行這些異常的相關的handler進行恢復操作，Phaser也可能被ForkJoinPool中的任務使用，這樣在其他任務阻塞等待一個phase時可以保證足夠的並行度來執行任務。
### Termination(終止機制)
可以用`isTerminated`方法檢查`phaser`的終止狀態。
在終止時，所有同步方法立刻返回一個負值。
在終止時嘗試註冊也沒有效果。當呼叫`onAdvance`返回true時Termination被觸發。當deregistration操作使已註冊的parties變為0時，onAdvance的預設實現就會返回true。也可以重寫onAdvance方法來定義終止動作。forceTermination方法也可以釋放等待執行緒並且允許它們終止。
### **Tiering(分層結構)**
`Phaser`支援分層結構(樹狀構造)來減少競爭。
註冊了大量parties的Phaser可能會因為同步競爭消耗很高的成本， 因此可以設定一些子Phaser來共享一個通用的parent。這樣的話即使每個操作消耗了更多的開銷，但是會提高整體吞吐量。
在一個分層結構的phaser裡，子節點phaser的註冊和取消註冊都通過父節點管理。
子節點phaser通過構造或方法`register`、`bulkRegister`進行首次註冊時，在其父節點上註冊。子節點phaser通過呼叫`arriveAndDeregister`進行最後一次取消註冊時，也在其父節點上取消註冊。
### **Monitoring(狀態監控)**
由於同步方法可能只被已註冊的parties呼叫，所以phaser的當前狀態也可能被任何呼叫者監控。在任何時候，可以通過getRegisteredParties獲取parties數，其中`getArrivedParties`方法返回已經到達當前phase的parties數。當剩餘的parties(通過方法getUnarrivedParties獲取)到達時，phase進入下一代。這些方法返回的值可能只表示短暫的狀態，所以一般來說在同步結構裡並沒有啥卵用。
### 分層執行示意圖
## 使用示例
```java
void runTasks(List tasks) throws InterruptedException{
// "1" to register self
final Phaser phaser = new Phaser(1);
// create and start threads
for (final Runnable task : tasks) {
phaser.register();
new Thread() {
@Override
public void run() {
// await all creation
// 類似 CountDownLatch.await() 和 CyclicBarrier.await()
System.out.println("等待所有的任務+1");
phaser.arriveAndAwaitAdvance();
task.run();
}
}.start();
}
// allow threads to start and deregister self
TimeUnit.SECONDS.sleep(1);
System.out.println("jinglingwang.cn 放行。。。。。。");
// 類似 CountDownLatch.countDown() 減到了0 和 CyclicBarrier 中的最後一個執行緒呼叫了await()
phaser.arriveAndDeregister();
}
```
### 多階段執行示例
這裡的階段有點類似多次使用CyclicBarrier，並不是Phaser的分層
```java
void runTasks2() {
// 定義階段數
int phases = 3;
// 進入下一個階段需要的參與數（執行緒數）
int parties = 5;
// 自定義onAdvance https://jinglingwang.cn
Phaser phaser = new Phaser(parties){
@Override
protected boolean onAdvance(int phase,int registeredParties){
System.out.println("階段phase: "+(phase +1) +" 執行完畢");
return phase > phases || registeredParties == 0;
}
};
for(int i = 1; i <= parties; i++){
new Thread(()->{
for(int j = 1; j <= phases; j++){
System.out.println(Thread.currentThread().getName() + " doing 階段："+ j);
phaser.arriveAndAwaitAdvance();
}
},"Thread-"+i).start();
}
}
```
執行結果：
```java
Thread-1 doing 階段：1
Thread-4 doing 階段：1
Thread-3 doing 階段：1
Thread-2 doing 階段：1
Thread-5 doing 階段：1
階段phase: 1 執行完畢
Thread-5 doing 階段：2
Thread-3 doing 階段：2
Thread-4 doing 階段：2
Thread-1 doing 階段：2
Thread-2 doing 階段：2
階段phase: 2 執行完畢
Thread-2 doing 階段：3
Thread-3 doing 階段：3
Thread-1 doing 階段：3
Thread-4 doing 階段：3
Thread-5 doing 階段：3
階段phase: 3 執行完畢
```
## 原始碼分析
### 內部類QNode
內部等待佇列，用於在阻塞時記錄等待執行緒及相關資訊
```java
static final class QNode implements ForkJoinPool.ManagedBlocker {
final Phaser phaser;
final int phase;
final boolean interruptible;
final boolean timed;
boolean wasInterrupted;
long nanos;
final long deadline;
volatile Thread thread; // nulled to cancel wait
QNode next; // 由此看出是一個單向列表
QNode(Phaser phaser, int phase, boolean interruptible,
boolean timed, long nanos) {
this.phaser = phaser;
this.phase = phase;
this.interruptible = interruptible;
this.nanos = nanos;
this.timed = timed;
this.deadline = timed ? System.nanoTime() + nanos : 0L;
thread = Thread.currentThread();
}
... 部分程式碼省略 ...
}
```
### 主要的屬性
```java
// 狀態變數，用於儲存當前階段phase、參與者數parties、未完成的參與者數unarrived_count
// 低0-15位表示未到達parties數，中16-31位表示等待的parties數，中32-62位表示當前階段phase
private volatile long state;
// 最多可以有多少個參與者，即每個階段最多有多少個任務，十進位制表示為65535
private static final int MAX_PARTIES = 0xffff;
// 最多可以有多少階段，2的31次方-1，十進位制：2147483647
private static final int MAX_PHASE = Integer.MAX_VALUE;
// 參與者數量的偏移量
private static final int PARTIES_SHIFT = 16;
// 階段的偏移量
private static final int PHASE_SHIFT = 32;
// 未完成的參與者數的掩碼，低16位，二進位制：1111 1111 1111 1111
private static final int UNARRIVED_MASK = 0xffff; // to mask ints
// 參與者數，中間16位，二進位制：1111 1111 1111 1111 0000 0000 0000 0000
private static final long PARTIES_MASK = 0xffff0000L; // to mask longs
// counts的掩碼，counts等於參與者數和未完成的參與者數的'|'操作
// 二進位制：1111 1111 1111 1111 1111 1111 1111 1111
private static final long COUNTS_MASK = 0xffffffffL;
// 二進位制位第64位為1，終止位
private static final long TERMINATION_BIT = 1L << 63;
// 一些特殊的值
// 一次一個參與者完成
private static final int ONE_ARRIVAL = 1;
// 增加減少參與者時使用，1左移16位，二進位制：0001 0000 0000 0000 0000
private static final int ONE_PARTY = 1 << PARTIES_SHIFT;
// 減少參與者時使用，二進位制：0001 0000 0000 0000 0001
private static final int ONE_DEREGISTER = ONE_ARRIVAL|ONE_PARTY;
// 沒有參與者時使用
private static final int EMPTY = 1;
// 當前Phaser的父級；如果沒有，則為null
private final Phaser parent;
/** phaser的根。如果不在樹中則等於phaser */
private final Phaser root;
/** 兩個佇列連結串列，在偶數和奇數階段交替使用 */
private final AtomicReference evenQ; // 偶數
private final AtomicReference oddQ; // 奇數
```
### 構造方法
```java
public Phaser() {
this(null, 0);
}
public Phaser(int parties) {
this(null, parties);
}
public Phaser(Phaser parent) {
this(parent, 0);
}
public Phaser(Phaser parent, int parties) {
if (parties >>> PARTIES_SHIFT != 0)
throw new IllegalArgumentException("Illegal number of parties");
int phase = 0;
this.parent = parent;
if (parent != null) { // 有設定parent
final Phaser root = parent.root;
this.root = root;
this.evenQ = root.evenQ;
this.oddQ = root.oddQ;
if (parties != 0)
phase = parent.doRegister(1);
}
else {
this.root = this; // root是當前phaser
// 初始化兩個佇列
this.evenQ = new AtomicReference();
this.oddQ = new AtomicReference();
}
// 確定state，先是一個三目運算
// parties 為 0 時，state為 1
//
this.state = (parties == 0) ? (long)EMPTY :
((long)phase << PHASE_SHIFT) | // 當前階段左移32位
((long)parties << PARTIES_SHIFT) | // 等待的parties數，左移16位
((long)parties); // 未到達parties數，就存低16位
}
```
整個構造方法最重要的就是最後`state`值的確認，也可以看出低0-15位表示未到達parties數，中16-31位表示等待的parties數，中32-62位表示當前階段phase。
比如入參為5的話，初始化的state值的二進位制表示為：0101 0000 0000 0000 0101
### register()方法
方法說明：向當前phaser新增一個新的unarrived（未到達）的party，如果`onAdvance`正在執行，那麼這個方法會等待它執行結束再返回結果。如果當前phaser有父節點，並且當前phaser上沒有已註冊的party，那麼就會交給父節點註冊。
程式碼分析：
```java
public int register() {
return doRegister(1);
}
private int doRegister(int registrations) {
// 調整的狀態，等待的parties數和unarrived（未到達）parties數同時增加
long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
final Phaser parent = this.parent;
int phase;
for (;;) { //自旋
long s = (parent == null) ? state : reconcileState(); // 取state值
// 轉換成int，state的低32位，也就是parties和unarrived的值
int counts = (int)s;
// 取等待的parties數
int parties = counts > head = (phase & 1) == 0 ? evenQ : oddQ;
QNode q = node.next = head.get();
// 再次判斷
if ((q == null || q.phase == phase) && (int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq
queued = head.compareAndSet(q, node); // CAS修改入隊
} else {
try {
ForkJoinPool.managedBlock(node); // 阻塞node,等待被喚醒
} catch (InterruptedException ie) {
node.wasInterrupted = true;
}
}
}
// 執行緒已經被喚醒，並且phase已經有變化了才會退出上面的自旋,或者完成終止，退出自旋
if (node != null) {
if (node.thread != null)
node.thread = null; // 避免 unpark()
if (node.wasInterrupted && !node.interruptible)
Thread.currentThread().interrupt();
if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
return abortWait(phase); // possibly clean up on abort
}
// 喚醒當前phaser階段的執行緒
releaseWaiters(phase);
return p;
}
/** 從佇列中刪除執行緒，喚醒當前phaser階段的執行緒 */
private void releaseWaiters(int phase) {
QNode q; // 佇列的第一個元素
Thread t; // its thread
// 再次根據當前phaser選擇對應的佇列
AtomicReference head = (phase & 1) == 0 ? evenQ : oddQ;
while ((q = head.get()) != null && q.phase != (int)(root.state >>> PHASE_SHIFT)) {
if (head.compareAndSet(q, q.next) && (t = q.thread) != null) {
// 刪掉q節點，喚醒q節點中的執行緒
q.thread = null;
LockSupport.unpark(t); // 喚醒執行緒
}
}
}
```
register()方法總結：
1. register方法為phaser新增一個新的party，如果onAdvance正在執行，那麼這個方法會等待它執行結束再返回結果。
2. register和bulkRegister都由doRegister實現，bulkRegister是批量註冊新增
3. 使用了自旋 + CAS 技術來保證更新成功
4. 如果前階段正在執行onAdvance()方法，則需要阻塞等待（根據phase入相應佇列）其執行完後再進行註冊
5. 當前phaser如果有父節點，需要交由父節點來完成註冊
### arrive()方法
使當前執行緒到達phaser，不等待其他任務到達。返回arrival phase number。
```java
public int arrive() {
// 一次一個參與者完成
return doArrive(ONE_ARRIVAL); // 特殊的屬性值 ONE_ARRIVAL: 1
}
private int doArrive(int adjust) {
final Phaser root = this.root;
for (;;) { // 自旋
long s = (root == this) ? state : reconcileState(); // 確定state值
int phase = (int)(s >>> PHASE_SHIFT); //位運算，得到當前階段phaser
if (phase < 0)
return phase;
int counts = (int)s; // 表示parties和unarrived的值
int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK); // 計算未到達數
if (unarrived <= 0)
throw new IllegalStateException(badArrive(s)); // 到達時邊界異常
if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) { // CAS直接修改state
if (unarrived == 1) { // == 1 表示當前為最後一個未到達的任務
long n = s & PARTIES_MASK; // 掩碼計算當前parties， 保留了16-32位的部分
int nextUnarrived = (int)n >