摘要：前言在前面的幾篇文章中詳述了框架的若干組分在相應的官方文檔中總會不時地提起同樣的也提到可以用于幫助運行在中的運行時保持有效的執行并行度其實特指其他都在等待一個的前進時熟悉的朋友都知道它的大概組成部分包含支持并發的容器同步器線程池阻塞隊列原子

在前面的幾篇文章中詳述了ForkJoin框架的若干組分,在相應的官方文檔中總會不時地提起"Phaser",同樣的,也提到Phaser可以用于幫助運行在ForkJoinPool中的ForkJoinTask運行時保持有效的執行并行度(其實特指其他task都在等待一個phase的前進時).

熟悉JUC的朋友都知道它的大概組成部分包含:Containers(支持并發的容器),Synchronizers(同步器),Executors(線程池),BlockingQueue(阻塞隊列),Atomic(原子類),Lock?and?Condition(鎖).而Phaser和CyclicBarrier,Semaphore等一樣是一個同步器.

本文主要介紹Phaser的內部實現,粗略介紹使用,它的源碼相比于線程池較為簡單,但最好能對比其他同步器來了解,讀者最好擁有juc其他同步器,原子類,部分ForkJoin框架的基礎.

同時,本文也會再次提到ForkJoinPool::managedBlock(blocker),之前在ForkJoinPool一文提到了實現和接口,而在CompletableFuture中見到了一個blocker的實現.

首先來看一些與Phaser狀態有關的簡單的常量.

//64位整數表示Phaser的狀態.
private volatile long state;

private static final int  MAX_PARTIES     = 0xffff;//最大parties,后16位表示.
private static final int  MAX_PHASE       = Integer.MAX_VALUE;//最大phase,最大整數值.
private static final int  PARTIES_SHIFT   = 16;//取parties使用的移位數,16
private static final int  PHASE_SHIFT     = 32;//取phase的移位數,32
private static final int  UNARRIVED_MASK  = 0xffff;      //未到的,取后16位.
private static final long PARTIES_MASK    = 0xffff0000L; //參加者,17-32位.
private static final long COUNTS_MASK     = 0xffffffffL; //數量,后32位.
private static final long TERMINATION_BIT = 1L << 63;//終止態,首位.

// 特殊值.
private static final int  ONE_ARRIVAL     = 1;
private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;//第1位和17位.顯然,它表示了一個ONE_ARRIVAL信息和PARTY信息.
private static final int  EMPTY           = 1;

//對一個state s計算unarrived的count,
private static int unarrivedOf(long s) {
    //直接取整數位,如果等于EMPTY(1)則返回0,否則取后16位.
    int counts = (int)s;
    return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
}

//對一個state,取出parties信息,直接取state的17至32位.
private static int partiesOf(long s) {
    return (int)s >>> PARTIES_SHIFT;
}
//對于一個state,取出phase信息,直接取前32位.
private static int phaseOf(long s) {
    return (int)(s >>> PHASE_SHIFT);
}
//對于一個state,取出arrived信息
private static int arrivedOf(long s) {
    int counts = (int)s;
    //state的后32位等于1(EMPTY)返回0,否則返回parties(state的17至32位,參考上面的partiesOf方法)和UNARRIVED(state的后16位)的差.
    return (counts == EMPTY) ? 0 :
        (counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK);
}

上面都是一些常量,沒什么可分析的,簡單來個總結.

Phaser用一個long型的state保存狀態信息.

state的前32位表示phase,后16位表示unarrivied,17至32位表示parties,parties減去unarrived即arrived.

下面我們看一些成員變量和有關函數.

//this的父,可以是null表示none
private final Phaser parent;

//phaser顯然是個樹的結果,root代表根,如果當前phaser不在樹內,則root==this
private final Phaser root;


//偶數隊列和奇數隊列.它們存放等待線程棧的頭,為了減少當添加線程與釋放線程的競態,
//這里使用了兩個隊列并互相切換,子phaser共享root的隊列以加快釋放.
private final AtomicReference evenQ;
private final AtomicReference oddQ;

//決定某個phase的等待線程隊列.
private AtomicReference queueFor(int phase) {
    //選擇隊列的方法,如果參數phase是偶數,使用evenQ,否則oddQ.
    return ((phase & 1) == 0) ? evenQ : oddQ;
}

//出現arrive事件時的邊界異常信息.
private String badArrive(long s) {
    return "Attempted arrival of unregistered party for " +
        stateToString(s);
}

//注冊時的邊界異常信息.
private String badRegister(long s) {
    return "Attempt to register more than " +
        MAX_PARTIES + " parties for " + stateToString(s);
}
//他們都用到的stateToString(s),計算參數s對應的phase,parties,arrived.
private String stateToString(long s) {
    return super.toString() +
        "[phase = " + phaseOf(s) +
        " parties = " + partiesOf(s) +
        " arrived = " + arrivedOf(s) + "]";
}

為了便于理解,先來看隊列的實現.

//表示等待隊列的QNode,實現了ManagedBlocker
static final class QNode implements ForkJoinPool.ManagedBlocker {
    //存放所屬phaser
    final Phaser phaser;
    //所屬phase
    final int phase;
    //是否可擾動
    final boolean interruptible;
    //是否定時
    final boolean timed;
    //是否已擾動
    boolean wasInterrupted;
    //計時相關
    long nanos;
    final long deadline;
    //關聯線程,當是null時,取消等待.
    volatile Thread thread; 
    //下一個QNode
    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();
    }
    //isReleasable方法
    public boolean isReleasable() {
        if (thread == null)
            //1.線程已置空(如2),返回true釋放.
            return true;
        if (phaser.getPhase() != phase) {
            //2.發現phaser所處的phase不是構建QNode時的phase了,就置線程為空,返回true.
            thread = null;
            return true;
        }
        if (Thread.interrupted())
            //3.如果當前線程擾動了.
            wasInterrupted = true;
        if (wasInterrupted && interruptible) {
            //4.發現擾動標記,并且QNode配置為可擾動,則置線程null并返回true
            thread = null;
            return true;
        }
        if (timed) {
            //5.定時邏輯,還有nanos,計算新的時長.
            if (nanos > 0L) {
                nanos = deadline - System.nanoTime();
            }
            if (nanos <= 0L) {
                //已經到時間,返回true,線程置空.
                thread = null;
                return true;
            }
        }
        return false;
    }
    //block邏輯
    public boolean block() {
        if (isReleasable())
            return true;
        else if (!timed)
            //不定時的park
            LockSupport.park(this);
        else if (nanos > 0L)
            //定時的情況.
            LockSupport.parkNanos(this, nanos);
        //老規矩
        return isReleasable();
    }
}

前面介紹過CompletableFuture的Singnaller,以及ForkJoinPool中的managedBlock,這一塊的邏輯顯然駕輕就熟.

很明顯,如果我們在ForkJoinPool中使用它作為blocker,并在相應的ForkJoinTask的exec或CountedCompleter的compute方法中使用ForkJoinPool::managedBlock(blocker),將每個ForkJoinWorkerThread在阻塞前構建一個QNode進入Phaser的等待隊列(雖然還沒有講到相關內容,但是Phaser顯然不用我們直接操作內部類QNode),那么它將依照上述邏輯進行補償,保障有效的并行度.

前面完成了承前啟后,預熱到此結束,開始看Phaser的核心方法.

//doArrive方法
//它是arrive和arriveAndDeregister方法的主要實現.手動調用這些方法可以加速通過和最小化競態窗口期.
//參數代表要從當前state中減去的調整數值,它的單位依托于業務,當為arrive時減去的單位為ONE_ARRIVAL,
//當為arriveAndDeregister時減去的單位為ONE_DEREGISTER.
private int doArrive(int adjust) {
    final Phaser root = this.root;
    for (;;) {
        //1.變量s初始化,取決于是否當前Phaser是root.不是root將試圖從root同步滯后的state.
        long s = (root == this) ? state : reconcileState();
        //計算phase,前32位.
        int phase = (int)(s >>> PHASE_SHIFT);
        if (phase < 0)
            //2.負數直接返回.說明原來的state首位就是1,前面的TERMINATE_BIT就是64位置1.
            return phase;
        //取count,后32位.
        int counts = (int)s;
        //計算unarrived,和前面一樣的邏輯.
        int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
        if (unarrived <= 0)//2.1
            //沒有unarrived了,說明不應該調用此方法,拋出異常,信息就是前面介紹過的badArrive
            throw new IllegalStateException(badArrive(s));
        //3.嘗試將state減去adjust數.
        if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) {
            //3.1cas成功后,unarrived余1,則前進一個phase
            if (unarrived == 1) {
                //3.1.1取出parties作為下一個state的基礎.
                long n = s & PARTIES_MASK;
                //3.1.2 下一個unarrived,數值上等于parties.
                int nextUnarrived = (int)n >>> PARTIES_SHIFT;
                if (root == this) {
                    //3.1.3當前Phaser是root,onAdvance返回true,則加上終止信號.
                    if (onAdvance(phase, nextUnarrived))
                        n |= TERMINATION_BIT;
                    else if (nextUnarrived == 0)
                        //3.1.4 onAdvance返回false,而計算得出的nextUnarrived是0,即沒有parties,n加上一個empty(1)
                        n |= EMPTY;
                    else
                        //3.1.5nextUnArrived不是0,加到n上.
                        n |= nextUnarrived;
                    //3.1.6前面的流程完成了state的后32位(parties和unarrived),接下來處理前32位.
                    //限定在MAX_PHASE之內,對當前phase加1.
                    int nextPhase = (phase + 1) & MAX_PHASE;
                    //將nextPhase的值加到n的前32位.并用n去cas掉原來的state,因為有3處入口的cas,此處一定能成功
                    n |= (long)nextPhase << PHASE_SHIFT;
                    UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
                    //更新到新的phase,喚醒等待的waiter.
                    releaseWaiters(phase);
                }
                //3.1.7當前Phaser不是root,當nextUnarrived計算得0時,像父傳遞解除注冊,參數ONE_DEREGISTER
                //會同時減去一個unarrived和一個parties.下輪循環正常應進入3.1.8
                else if (nextUnarrived == 0) { 
                    phase = parent.doArrive(ONE_DEREGISTER);
                    //完成傳遞后,將自己的state置empty.
                    UNSAFE.compareAndSwapLong(this, stateOffset,
                                              s, s | EMPTY);
                }
                else
                    //3.1.8,當前Phaser不是root,計算的nextUnarrived非0,像父傳遞一個arrive事件,減去一個unarrived.
                    phase = parent.doArrive(ONE_ARRIVAL);
            }
            //3.2返回當前phase,可能是已進入3.1遞增的.僅有此處可退出循環.
            return phase;
        }
    }
}

關于該方法的執行流程,我們結合幾個周邊方法一并分析,先來看注冊方法和onAdvance勾子.

//注冊和批量注冊.參數代表parties和unarrived字段的增加數,它必須大于0.
private int doRegister(int registrations) {
    // 1.用參數計算一個adjust,同時包含parties和arrive.
    long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
    final Phaser parent = this.parent;
    int phase;
    //循環嘗試更改.
    for (;;) {
        //2.存在parent,則用root的phase調整this的state.
        long s = (parent == null) ? state : reconcileState();
        //取出當前state中保存的counts,parties,unarrived信息.
        int counts = (int)s;
        int parties = counts >>> PARTIES_SHIFT;
        int unarrived = counts & UNARRIVED_MASK;
        if (registrations > MAX_PARTIES - parties)
            //要注冊的數量大于了余量,拋出異常.
            throw new IllegalStateException(badRegister(s));
        //3.計算出phase
        phase = (int)(s >>> PHASE_SHIFT);
        if (phase < 0)
            //phase為負說明state為負,即終止態,終止.
            break;
        //4.當前state表示的參與數非空的邏輯,當前注冊非首次注冊.
        if (counts != EMPTY) {           
            if (parent == null || reconcileState() == s) {
                //this是root或者從root同步的state不變,繼續執行,否則重新循環.
                if (unarrived == 0)  
                    //4.1本輪循環通過原state計算的unarrived為0,說明應等待下一phase,使用root等待      
                    root.internalAwaitAdvance(phase, null);
                else if (UNSAFE.compareAndSwapLong(this, stateOffset,
                                                   s, s + adjust))
                    //4.2本輪循環未發現應等待下一phase,嘗試原子更新,增加adjust到state上.
                    break;
            }
        }
        //5.當前不存在counts,且自身就是root,代表root的首次注冊.
        else if (parent == null) { 
            //5.1計算下一個state,因為沒有參與數,使用phase初始化前32位,并使用adjust做后32位.        
            long next = ((long)phase << PHASE_SHIFT) | adjust;
            if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
                //5.2 cas成功,退出,不成功,下輪循環.
                break;
        }
        //6.是首次注冊,但也不是root的邏輯.代表非root的Phaser的首次注冊.
        else {
            //6.1對當前Phaser加鎖并double check,避免同時調用.加鎖失敗的線程將在后續進入2的邏輯.
            synchronized (this) {  
                //double check state未發生改變.   
                if (state == s) {
                    //6.2首先向父Phaser注冊1.
                    phase = parent.doRegister(1);
                    if (phase < 0)
                        //發現進入終止態,直接停止.
                        break;
                    
                    //6.3向父Phaser注冊成功,循環嘗試cas掉老的state,新state的算法同上,phase加adjust.
                    //在整個while循環中,不再考慮phase進入終止態的情況,因為這些操作處于同一個"事務"中,
                    //且因競態等原因,若某次cas時計入了負數的phase,方法返回后也可以及時發現.
                    while (!UNSAFE.compareAndSwapLong
                           (this, stateOffset, s,
                            ((long)phase << PHASE_SHIFT) | adjust)) {
                        //如果cas不成功,則讀取s為新的state,計算新的phase并重新循環.
                        s = state;
                        phase = (int)(root.state >>> PHASE_SHIFT);
                        // assert (int)s == EMPTY;
                    }
                    //6.4cas成功后退出循環.
                    break;
                }
                //如果if(state==s)判斷失敗,說明有別的線程有當前線程進入synchronized塊前已經加鎖并執行了內部的邏輯且稍后釋放了鎖,
                //這樣當前線程加鎖成功,但if判斷失敗,它會立即釋放鎖并返回到2.
            }
        }

    }
    return phase;
}


//使用root的phase調整this的state,更新滯后的結果.這一般發生在root前進了phase但是
//子phaser還沒有做到這一步,這種情況下,子phaser必須完成這個前進的步驟,這一過程中,phase將
//被置為root的phase,unarrived則會重置為parties,若parties為0,則置為EMPTY.返回結果state.
private long reconcileState() {
    final Phaser root = this.root;
    long s = state;
    //不是root才進行下一步.
    if (root != this) {
        int phase, p;
        //cas,phase采用root,parties不變,unarrived重置為parties或EMPTY.
        while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
               (int)(s >>> PHASE_SHIFT) &&
                //phase滯后于root
                //嘗試cas.
               !UNSAFE.compareAndSwapLong
               (this, stateOffset, s,
                //確定新state的前32位,使用root的phase.
                s = (((long)phase << PHASE_SHIFT) |
                    //新phase<0,后32位直接取this的state表示的counts.
                     ((phase < 0) ? (s & COUNTS_MASK) :
                    //phase有效,this的state表示的parties為0,則后32位使用empty
                      (((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY :
                        //否則,后32位使用parties.
                       ((s & PARTIES_MASK) | p))))))
            s = state;
    }
    return s;
}


//onAdvance勾子方法,參數為當前phase和注冊的parties數.
//默認實現為parties數為0,方法返回true時,調用者會嘗試終止Phaser.(參考前面的doArrive).隨后調用isTerminated方法將返回true.
//執行此方法時拋出的運行時異常或Error將直接上拋給嘗試advance相應的phase的線程,這種情況下不會發生phase的advance.
//方法的入參表示的是Phaser當前的state(未advance前),因此若在onAdvance方法中執行arrive,regist,waiting這三種操作的行為是不確定的也不可靠的.
//如果當前Phaser是一個級聯的成員,那么onAdvance只會由root在每次advance時調用.
//方法的默認實現返回true的場景目前只能是經過數次arriveAndDeregister調用造成parties歸零的結果.我們繼承Phaser可以輕易地重寫此行為,
//比如簡單粗暴地返回false,那么將永遠允許新的注冊.
protected boolean onAdvance(int phase, int registeredParties) {
    return registeredParties == 0;
}

經過前面的代碼分析,已經對Phaser的核心函數doRegister,doArrive有了全面的了解.

兩者都會在一開始同步root的phase,且如果出現落后root的情況,同步了新的phase的同時,也會重新初始化unarrived,并且使用parties的值.

doArrive方法會每次調整unarrived數量(也可包含parties數量,如果使用了解除注冊),當Phaser調用自身的arrive/arriveAndDeregister時,會做出相應的減少,并根據是否為root而決定向上遞歸.

Phaser減少自身unarrived信號(也可能同時有parties信號)后,若發現這已經是最后一個unarrived信號,則進行接下來的判斷:

1.當前Phaser是root,advance并喚醒waiter.(重要的喚醒操作執行點,root一輪完成)

2.當前Phaser不是root,且它已經不具備繼續下一輪的條件(計算nextUnarrived為0,即parties已經被arriveAndDeregister置0),則從父Phaser減少一個unarrived和parties.

3.當前Phaser不是root,但它仍具有parties,滿足進行下一輪的條件(計算nextUnarrived不是0),則從父Phaser減少一個unarrived,但不減少parties.

顯然,子Phaser的最后一個unarrived的消失一定會造成父的unarrived減少,子Phaser不能繼續下一phase的register和arrive時,從父Phaser中卸載.

若不是本Phaser的最后一個unarrived信號,則直接結束,相當于只進行了上面的減少信號操作.

doRegister方法的邏輯大致相反,不同于doArrive,它的參數registrations同時作用于parties和unarrived,即兩個位上同時加上registrations參數.它的大致邏輯:

1.當前注冊并非首次注冊,且出現unarrived==0,即本輪已經完成了arrive,那么本輪將不能注冊,需要等待root更新到下輪.(這也是我們碰到的第一個阻塞)

2.當前注冊并非首次注冊,unarrived也不是0,則在本phase進行注冊,增加相應的parties和unarrived.

3.當前注冊是root的首次注冊,給root的state加上相應的parties和unarrived.

4.當前注冊是非root的首次注冊,加鎖(this),對自己的state加上相應的parties和unarrived(同上,以registrations為單位),而對parent加上一個parties和unarrived單位.

很明顯,對于單Phaser的情況非常好理解,每次減少unarrived數量(先不考慮減少parties),則最終導致Phaser自身進入下一個phase,然后重新初始化unarrived到下一輪,unarrived的新值是前一輪剩下的parties數量.

當我們同時也嘗試減少parties數量,即解除parties的注冊,最終導致沒有parties,那么Phaser將進入終止態.

整個過程中,只要Phaser沒進入終止態,隨時可以進行新的注冊,并增加parties和unarrived的數量.每個arrive可以減少unarrived的數量為任何正整數,不一定是1.

對于多Phaser的情況,有兩個特殊點:

1.對任意Phaser樹中的某一個Phaser調用注冊操作,會令自身加上相應參數個parties和unarrived單位,僅會在該Phaser第一次注冊時增加父Phaser(極端可能,僅從一個葉子節點第一個注冊的情況下可一直遞歸到root)的parties數和unarrived數各1單位(不論參數是多少).

2.對任意Phaser樹中的某一個Phaser調用arrive操作,會令自身減去相應的參數個parties和unarrived單位,同時僅當本Phaser此時是最后一個unarrived時,會減去父Phaser的一個unarrived單位(當前子Phaser仍舊有parties可以構建下一phase),或減去父Phaser一個Parties和unarrived單位.(極端情況下,每一級都是最后一個unarrived時,減少葉子節點的最后一個unarrived會遞歸到root).

每新增一個子Phaser,父Phaser就會增加一個要完成觸發phase的advance前必須要等到arrive的單位;每一個子Phaser中所有的arrive完成,父Phaser都將減少一個要等待advance所必需觸發的arrive.

目前沒有看到await方法,但可以提前說明,等待操作完全依賴于root是否完成本輪.也就是所有子Phaser都完成了同一輪(arrive打滿),才能讓父Phaser本身減去一個所有arrive單位,再觸發父Phaser本輪的完成,此時對任何已完成的Phaser進入注冊,都會進入上述的root.internalAwaitAdvance(phase, null)方法等待root進入下一phase.如果對已經完成所有arrive的Phaser繼續進行arrive操作,因為unarrived已經是0,則會拋出異常.

所以對于使用子Phaser的場景,如果發生很巧妙的情況,Phaser樹上當前子Phaser的arrive結束條件滿足了,使得新來的注冊只能等待下一輪次,而其他分支的子Phaser又偏偏不能完成本輪次,那么新的phaser.doRegister方法將阻塞在此.

好在我們使用Phaser可能會類似CyclicBarrier的使用方式,可對每一輪(phase)進行注冊并等待(也許只等一輪,那么arrive就要帶上deregister),每一輪最后一個線程arrive了,就會停止所有線程的等待,讓所有線程繼續執行,同時開啟了下一輪次,這些線程此時又可以不經注冊直接在新的輪次中進行等待,直到最后一個arrive了,再次喚醒所有線程并繼續執行,同時Phaser再前進一輪,如此往復.中間使用arrive并deregister的線程會從本輪起減少一個unarrive數量(因為parties也減少了,所以再下一輪初始化unarrive數量時也會減少一次).我們可以讓這些線程參與任意的輪次,但要注意的是,如果有線程中途不參加了,一定要解除注冊,否則因為每輪初始化時,要等待arrive的數量都是上一輪剩下的parties數量,有線程停止了執行,卻不減少parties數,那么下輪所有等待的線程將永遠等不到phaser滿足喚醒的條件.

上述的過程中可以明顯的看出,目前已介紹的兩個重要核心函數:注冊和arrive并沒有直接記錄和操作線程的操作,相應的操作在等待方法和喚醒方法中(前面提到過release),我們稍后介紹.

現在假設一個特殊的使用場景,也可以區別于CyclicBarrie和CountDownLatch的使用.還是上面的例子,但是我們準備的線程數與Phaser的parties數/unarrived數不同(一般前者要多些),會發生什么事?

首先創建了Phaser,不指定最初parties數,并用每個線程去注冊(我甚至可以用一個線程去重復注冊,每次的參數registrations還可以不同,注冊的作用并不是將當前線程壓入隊列,而是為本phase設置一個unarrive數量,以控制到達下個phase前必須有多少次arrive的發生),則parties數和unarrived的初值完全與此有關,是一個依托于我們隨意注冊而產生的隨意值.那么假定我們的線程數量大于這個parties數(假定調用注冊方法的線程和arrive及等待的線程無關),并令有的線程執行arrive(完全可以一次arrive減去多個信號量,甚至一個線程多次arrive),有的線程執行await等待信號advance到下一個phase(一個線程在一個周期只能調用一次),有的線程執行了arrive也等待phase前進(這種情況一個線程一周期也只能一次.其實這些分別對應了還未介紹的arrive,waitAdvance,arriveAndWaitAdvance等方法),多帶帶進行await操作的線程可以是任意數量,執行arrive方法的線程加上執行arrive并wait的操作的線程和必須超過unarrived,這才能喚醒等待線程.

目前這些還比較抽象,等到我們看過相應的幾個方法便了然了.

onAdvance的方法默認實現就是判斷本階段注冊的parties數量,如果已經是0則說明沒有parties了,Phaser應該結束.但是我們其實可以重新實現,比如參數中同時傳入了當前的phase,我可以規定上面的例子中phase最多只有3輪次,那么不論什么時候arrive,發現了當前phase已進入3輪,Phaser就被終止.當然,這一過程是由root執行的,但是子Phaser的phase會在每次注冊和arrive發生時同步root,因此本例中對于phase數的判斷可以粗放到所有Phaser,對于parties數則只能作用于root(事實上調用onAdvance的一定是root).

接下來看全量構造方法和若干和上面有關的公有函數.

//初始化一個Phaser,指定parent,指定未到來的參與者數(unarrived parties),但這只是一個初值,
//當我們在任何時候調用注冊方法時,還會相應的增加.
public Phaser(Phaser parent, int parties) {
    if (parties >>> PARTIES_SHIFT != 0)
        //太大了,超過了后16位能表示的整數.
        throw new IllegalArgumentException("Illegal number of parties");
    //初始phase為0.
    int phase = 0;
    this.parent = parent;
    if (parent != null) {
        //1.有parent的情況,共享parent的root,隊列,并向parent中注冊1個parties和unarrived,
        //同時同步一次phase(表面上是同步了parent的,實際上前面已經看過,會同步root).
        final Phaser root = parent.root;
        this.root = root;
        this.evenQ = root.evenQ;
        this.oddQ = root.oddQ;
        if (parties != 0)
            phase = parent.doRegister(1);
    }
    else {
        //2.無parent的情況,root就是this,并初始化奇偶等待隊列.它使用原子引用的形式存放一個QNode,而QNode我們前面已介紹.
        this.root = this;
        this.evenQ = new AtomicReference();
        this.oddQ = new AtomicReference();
    }
    //統一初始化state,后32位的決定依托于parties,如果parties是0則給予EMPTY,直接不管高32位.
    //不為0則給予phase設置為前32位,parties設置parties位和unarrived位.
    this.state = (parties == 0) ? (long)EMPTY :
        ((long)phase << PHASE_SHIFT) |
        ((long)parties << PARTIES_SHIFT) |
        ((long)parties);
}


//注冊方法,就是調用doRegister,參數1.
//它會向this添加一個unarrived的party,如果正巧root正在進行advance,它需要等待下個phase.
//如果this有parent,且它之前沒有過注冊的parties,則首次注冊會觸發自身向parent的注冊.
//如果this已經終止了,那么嘗試注冊將會無效并返回負值.如果注冊的數量大于了最大支持parties(后16位整數),
//會拋出IllegalStateException
public int register() {
    return doRegister(1);
}


//批量注冊指定的信號量,并返回最新的phase.規則基本同上.
public int bulkRegister(int parties) {
    if (parties < 0)
        throw new IllegalArgumentException();
    if (parties == 0)
        //參數0直接查詢最新的phase返回
        return getPhase();
    return doRegister(parties);
}


//arrive一個信號,不等待其他arrive事件,返回最新phase(終止態為負).
//當前Phaser的arrive事件已滿,則對parent來說也會觸發一個arrive.(如果有parent)
public int arrive() {
    return doArrive(ONE_ARRIVAL);
}


//arrive并解除一個注冊parties,也不阻塞等待其他arrive.如果當前Phaser的解除注冊操作
//將parties減至0,且this有parent,這將導致parent也減少一個parties(本phaser解除在parent的注冊).
public int arriveAndDeregister() {
    return doArrive(ONE_DEREGISTER);
}

接下來要看上面已經做足了鋪墊的等待方法了,并結合前面的隊列一塊看.

//令當前線程"到達"此phaser并等待其他parties,它等效于awaitAdvance(arrive()).
//注意,按照道格的注釋,如果你在一個未進行注冊(調用register)的線程里調用此方法其實是一個使用錯誤,
//但是從本方法和前面以及后面有關的方法來看,所有記錄線程的方法均只與arrive和等待有關,與注冊無關.
//因此Phaser本身無法規避這種使用錯誤,我們完全可以使用另一個線程去注冊,而當前線程去arrive,將兩個動作分開.
//方法會返回arrive時最新的phase號.終止時會是負值.
public int arriveAndAwaitAdvance() {
    //記錄root,開始循環.
    final Phaser root = this.root;
    for (;;) {
        //1.預計算,首先同步state
        long s = (root == this) ? state : reconcileState();
        //計算phase
        int phase = (int)(s >>> PHASE_SHIFT);
        if (phase < 0)
            //已終結直接返回最終phase.
            return phase;
        //計算counts,unarrived
        int counts = (int)s;
        int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
        if (unarrived <= 0)
            //已經沒有空余的unarrived信號了,不能再調用arrive,拋出異常.
            throw new IllegalStateException(badArrive(s));
        //2.減余arrive的有關邏輯.嘗試cas減去一個arrive
        if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
                                      s -= ONE_ARRIVAL)) {
            if (unarrived > 1)
                //2.1當前要減的信號不是本Phaser的最后一個信號量,調用root的等待方法.參數2是node,傳空.
                return root.internalAwaitAdvance(phase, null);
            if (root != this)
                //2.2當前要減的信號量是非root的Phaser的最后一個,遞歸給parent(雖然用了return,但是parent也可能在進入2.1后阻塞).
                return parent.arriveAndAwaitAdvance();
            //2.3當前要減的信號量是root的最后一個.
            //2.3.1準備計算下一個狀態,先取出state的parties信息.
            long n = s & PARTIES_MASK; 
            //計算nextUnarrived,它是現在的parties.
            int nextUnarrived = (int)n >>> PARTIES_SHIFT;
            //2.3.2前進phase邏輯.
            if (onAdvance(phase, nextUnarrived))
                //需要終止,給新state的計算基石n加上終止標記.
                n |= TERMINATION_BIT;
            else if (nextUnarrived == 0)
                //計算的nextUnarrived是0,即沒有parties,加上空標記位.
                n |= EMPTY;
            else
                //下一輪能正常進行,加上nextUnarrived位.
                n |= nextUnarrived;
            //2.3.3給n加上下一個phase.
            int nextPhase = (phase + 1) & MAX_PHASE;
            n |= (long)nextPhase << PHASE_SHIFT;
            if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))
                //用n進行cas不成功,將新的phase返回.
                //說明一下,因為方法執行到此前已經執行過2的入口cas,減去了最后一個unarrived,因此在2到此的過程中若有新的注冊,
                //它內部會讀到0個unarrived,就會等待下一個phase(參考前面介紹過的注冊方法),因此cas失敗不會是因為2之后有新的注冊.
                //在整個arrive系列的方法中,最后一次arrive發生后,本Phaser不可能有其他線程再去執行類似2處的減余的情況.
                //故出現這種情況的原因目前來看有二,一是還未介紹的強制關閉Phaser的方法,此時也會突兀地改掉state造成cas恰巧失敗,二是
                //出現一些用戶做出的奇葩行為,比如重寫了其他公有方法.我們自然忽略第二種情況,doug大神也是簡單注釋了一個"terminated".
                return (int)(state >>> PHASE_SHIFT); // terminated
            //cas成功,釋放等待隊列中的線程,返回下一個phase(因為在此過程中的register會等到advance,此時的phase已經是nextPhase了).
            releaseWaiters(phase);
            return nextPhase;
        }
        //3.減余失敗說明出現競態,直接開啟下輪循環重新減余.
    }
}


//等待當前Phaser從給定的phase前進結束,如果當前phase不等于給定的phase,或者Phaser已終止立即返回.
//1.傳入phase為負,返回它本身.
//2.傳入的phase不是最新的phase,返回最新的.
//3.傳入了最新的phase,等待到advance并返回advance后的phase.
public int awaitAdvance(int phase) {
    final Phaser root = this.root;
    long s = (root == this) ? state : reconcileState();
    int p = (int)(s >>> PHASE_SHIFT);
    if (phase < 0)
        return phase;
    if (p == phase)
        //匹配成功,等root前進.參數node為null
        return root.internalAwaitAdvance(phase, null);
    return p;
}


//參考前面的幾個方法,區別是可擾動.
public int awaitAdvanceInterruptibly(int phase)
    throws InterruptedException {
    final Phaser root = this.root;
    long s = (root == this) ? state : reconcileState();
    int p = (int)(s >>> PHASE_SHIFT);
    if (phase < 0)
        //1.參數phase小于0直接返回它本身.
        return phase;
    if (p == phase) {
        //2.參數phase匹配,回憶一個前面介紹的QNode,匹配當前Phaser和phase,配置為可擾動且不計時.
        QNode node = new QNode(this, phase, true, false, 0L);
        //3.放入root的等待隊列阻塞.
        p = root.internalAwaitAdvance(phase, node);
        if (node.wasInterrupted)
            //4.等待結束,判斷是否是擾動造成的結束,前面介紹過QNode的相關邏輯,
            //它實現了ForkJoinPool.ManagedBlocker,因此在managedBlock方法進行時,
            //會循環調用問詢是否能release,當我們配置了可擾動且擾動了,就會標記這個wasInterrupted,釋放線程引用并返回.
            //發現此種情況拋出異常.
            //同時,當發現等待成功,也會結束,釋放線程引用并返回,但不帶有擾動標記.
            throw new InterruptedException();
    }
    //5.返回1處之前讀取的phase或3處得到的最新phase值.
    return p;
}

//同上方法,但帶有計時.
public int awaitAdvanceInterruptibly(int phase,
                                     long timeout, TimeUnit unit)
    throws InterruptedException, TimeoutException {
    long nanos = unit.toNanos(timeout);
    final Phaser root = this.root;
    long s = (root == this) ? state : reconcileState();
    int p = (int)(s >>> PHASE_SHIFT);
    if (phase < 0)
        return phase;
    if (p == phase) {
        //不同于上面方法的地方,建立的QNode帶有計時和等待時長.
        QNode node = new QNode(this, phase, true, true, nanos);
        p = root.internalAwaitAdvance(phase, node);
        if (node.wasInterrupted)
            //被擾動的情況.
            throw new InterruptedException();
        else if (p == phase)
            //時間到了phase沒有前進,超時.
            throw new TimeoutException();
    }
    return p;
}

前面的幾個核心方法粗略過完,補充一些重要內容.

首先在前面曾分析過有線程阻塞等待下一個phase的情況,并沒有加上定時等待的考慮.在超時的情況下,阻塞的線程可能會收到異常并退出.

建立QNode可以限定是否定時和可擾動,這取決于我們使用哪個方法去await.

除最后一個線程arrive外,所有線程調用這些方法都會減少一個arrive并加入等待隊列,直到(1)配置了定時且超時,(2)當前是可擾動等待且使用了Thread.interrupt(),(3)最后一個線程使用上述方法或arrive方法,使得Phaser前進了一個輪次,internalWaitAdvance結束.其中(1)(2)均會遷成arrive線程拋出異常,只有(3)才是正常的情況.

QNode前面已介紹,它是一個blocker,需要調用ForkJoinPool::managedBlock才會起作用(顯然root的internalAwaitAdvance必然與此方法有關聯).當然這個作用與任務是否運行在ForkJoinPool無關,如果等待phaser前進的線程是運行在ForkJoinPool中的ForkJoinWorkerThread,顯然會在internalAwaitAdvance期間進行補償.這一塊可參考前面的"CompletableFuture與響應式編程"和"ForkJoin框架之ForkJoinPool"兩篇文章.

另外,這些代碼也再次說明了root的作用: (1)對一切非root的Phaser進行等待都會用root的internalAwaitAdvance;(2)每次注冊或arrive一定會同步root的最新phase.

其中(1)也間接說明了為什么構建Phaser時只有root創建等待隊列,所有子Phaser共享.

上面還保留了一個疑問,提到了"強制關閉Phaser"造成arriveAndAwaitAdvance出現cas失敗的問題,doug大神直接注釋了一個terminated,我們馬上來看這一塊,以及一些周邊的公共函數,加深理解,然后再來解決關于等待隊列最后的一些問題.

//強制關閉Phaser,讓Phaser進入終止態,但是這個過程不影響它已注冊的parties,如果此Phaser是
//一個Phaser樹中的成員,那么所有phaser集中的Phaser都會關閉,如果它已經關閉,此方法無效.此方法可以
//用于若干任務出現意料之外異常的情況下的協調恢復.
public void forceTermination() {
    // Only need to change root state
    final Phaser root = this.root;
    long s;
    //已是終止態直接忽略.
    while ((s = root.state) >= 0) {
        //直接嘗試給root的state加上終止位.顯然加上了它,子Phaser在注冊和arrive等方法同步回新的phase就是個負數,
        //因此更改root的phase為負相當于判了所有Phaser的死刑.唯一需要解決的是已經阻塞在root.internalAwaitAdvandce的線程.
        if (UNSAFE.compareAndSwapLong(root, stateOffset,
                                      s, s | TERMINATION_BIT)) {
            // 加上終止位成功,先后喚醒偶數等待隊列和奇數等待隊列.
            releaseWaiters(0); // Waiters on evenQ
            releaseWaiters(1); // Waiters on oddQ
            //返回
            return;
        }
    }
}


//返回當前phase,直接用root的state去取.
public final int getPhase() {
    return (int)(root.state >>> PHASE_SHIFT);
}

//查詢注冊的parties數量.調用前面介紹過的partiesOf
public int getRegisteredParties() {
    return partiesOf(state);
}

//查詢已經arrived的parties數量.調用介紹過的arriveOf

public int getArrivedParties() {
    return arrivedOf(reconcileState());
}

//查詢未arrive的parties數量,調用前面介紹過的unarrivedOf
public int getUnarrivedParties() {
    return unarrivedOf(reconcileState());
}

//返回parent
public Phaser getParent() {
    return parent;
}

//返回root
public Phaser getRoot() {
    return root;
}

//判斷當前Phaser是否終止,直接取root的state是否為負,可見,終止態完全取決于root.
public boolean isTerminated() {
    return root.state < 0L;
}

這些方法都比較簡單,只有forceTermination需要再強調一翻,前面介紹arrayAndAwaitAdvance時曾提過在減去最后一個unarrived信號后去cas到下一個phase失敗的情況,doug大神簡單注釋了一句terminated,直接返回了當前的phase(顯然只能是負),在周邊方法重重加鎖的前提下,那一次cas的失敗唯一一處就是強制關閉,因為它只改關閉標記位,相當于動了phase,而沒有動unarrived標記位和parties標記位.所以重寫Phaser的方法要謹慎,很可能不小心打破了這個封裝.

從上面的有關方法可以看出,子Phaser的終止態嚴重依賴于root,目前可以確定的是root的phase一旦表現出終止態,所有新來的注冊,arrive,arrive并await將會立即返回,唯一需要關注的就是root被設置了終止標記后,正陷入等待的線程怎么辦的問題.

我們下面就來看Phaser的等待機制,這里面又能見到道格大神非常有趣的玩法.

//工具方法,移除某個phase的等待者.
private void releaseWaiters(int phase) {
    QNode q;  //保存隊列中的隊首
    Thread t;  // 保存線程引用.
    //取隊列,用phase的奇偶決定,phase是偶數就取偶數隊列,否則取奇數隊列.而這個phase其實只用來取隊列了,后續的操作與它無關.
    AtomicReference head = (phase & 1) == 0 ? evenQ : oddQ;
    //循環,找出所有phase不等于root的phase的(其實root是最大的,所以就是找出非最新phase加入進來的waiter QNode)
    while ((q = head.get()) != null &&
           q.phase != (int)(root.state >>> PHASE_SHIFT)) {
        //找出了,利用原子引用將head指向next.
        if (head.compareAndSet(q, q.next) &&
            (t = q.thread) != null) {
            //發現阻塞者,喚醒線程.回憶下前面實現blocker方法中的isReleaseble和block方法都有將線程置空的操作.(三種情況,喚醒擾動超時都會置空)
            //但是那些方法并沒有將代表該阻塞線程的QNode移除隊列,因此可能會發現thread已經是null(代表無阻塞者)的情況,只需要移除隊列即可.
            q.thread = null;
            LockSupport.unpark(t);
        }
    }
}


//上面releaseWaiters方法的一個變種,但它只會處理遍歷過程中位于頭部的元素,出現正常的等待節點就會立即返回.
//此方法在這一塊可以有效的減少內存的占用.退出時返回當前的phase.
private int abortWait(int phase) {
    //同樣,參數phase只是用來選擇要處理的隊列.
    AtomicReference head = (phase & 1) == 0 ? evenQ : oddQ;
    for (;;) {
        Thread t;
        QNode q = head.get();
        //計算最新phase的值p
        int p = (int)(root.state >>> PHASE_SHIFT);
        if (q == null || ((t = q.thread) != null && q.phase == p))
            //1.出現q為null代表整隊列元素已出隊,直接返回p;
            //或者在出隊過程中head(q)記錄的線程引用還在,說明未超時或擾動,且是本phase的等待節點,終止循環并返回最新phase.
            return p;
        if (head.compareAndSet(q, q.next) && t != null) {
            //進入條件,參考1的條件,因為1會直接返回.故進入2的條件其實是q非空且處于舊的phase.只有這種情況才可以出隊.
            //2.將q出隊,置空線程引用,釋放線程.
            q.thread = null;
            LockSupport.unpark(t);
        }
    }
}

//計算有效cpu,控制自旋.
private static final int NCPU = Runtime.getRuntime().availableProcessors();

//常量,每輪arrive等待的字旋數,取決于NCPU,小于2則取1,不小于2取2的8次冪.
static final int SPINS_PER_ARRIVAL = (NCPU < 2) ? 1 : 1 << 8;


//珊珊來遲的內部等待方法.它可能會一直阻塞到phase的advance發生(除非取消了等待).
//此方法僅限root調用.參數phase表示當前的phase,參數node表示等待節點,用于追蹤節點的擾動或超時.
//如果是null,表示是一次不可擾動的等待.返回值為當前最新的phase.
private int internalAwaitAdvance(int phase, QNode node) {
    // 1.調用releaseWaiters,傳入參數phase的前一個phase,顯然這只是決定釋放哪一個隊列.參數絕對實時準確的情況下會先將老的隊列釋放掉.
    releaseWaiters(phase-1); 
    //節點入隊標記,入隊了就會變為true
    boolean queued = false;   
    //記錄每一輪循環的unarrived數量,用于決定是否擴增自旋等待次數.
    int lastUnarrived = 0; 
    //自旋數,參考上面的計算邏輯.
    int spins = SPINS_PER_ARRIVAL;
    long s;
    int p;
    //開啟循環,直到phase前進為止或內部判斷已取消等待.
    while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
        //2.傳入node是null,即非可擾動的模式邏輯.只有非可擾動模式才有自旋.
        if (node == null) {  
            //2.1每輪自讀進入都會嘗試計算新的unarrived,如果發現出現了變動(變大或者變小),
            //會將它保存到前面的lastUnarrived.        
            int unarrived = (int)s & UNARRIVED_MASK;
            if (unarrived != lastUnarrived &&
                (lastUnarrived = unarrived) < NCPU)
                //發現新變化的unarrived head = (phase & 1) == 0 ? evenQ : oddQ;
            QNode q = node.next = head.get();
            //這一行不起眼的if條件代碼真的是一個悄無聲息解決了一個大暗坑的地方,后面說.
            if ((q == null || q.phase == phase) &&
                (int)(state >>> PHASE_SHIFT) == phase) 
                //double check避免臟入隊,入隊條件是(1)無頭,(2)或者頭元素的phase等于參數phase(因為相鄰的兩個phase絕對不會入同一個隊).
                //滿足(1)(2)的同時,還要滿足(3),參數phase就是當前的state表示的phase(因為此方法只能root使用,故為root表示的最新phase).
                //條件滿足,入隊,取代原來的head,原來head代表的node成為node的next.而條件不滿足進入下一循環,很可能while條件就不滿足了退出循環.
                queued = head.compareAndSet(q, node);
        }
        //5.已經在某一輪循環入隊了,使用ForkJoinPool的managedBlock管理block,其間可能會釋放線程引用.
        else {
            try {
                //5.1它內部也有循環,且會調用前面看到過的isReleasable和block實現,顯然它一旦結束(包含擾動),一定會造成下輪外循環終止于3處.
                ForkJoinPool.managedBlock(node);
            } catch (InterruptedException ie) {
                //5.2出現擾動異常catch住,并保存.下輪循環也會終止在3處.
                node.wasInterrupted = true;
            }
        }
    }
    //6.走出上面的while循環,可能是root已經advance到下一個phase(2前的循環),也可能是傳入node的情況下出現了擾動或超時(5)造成(3)滿足
    if (node != null) {
        //6.1node存在代表可能已經壓入隊列,結果要么是已出現擾動或超時(方法結束后會拋出異常),要么是已正常完成.
        //顯然,代碼執行到此處就要返回了,阻塞的線程會拋出異常結束(超時或擾動)或繼續執行(正常advance),
        //沒有必要去嘗試喚醒能執行出前面while循環到達6馬上要返回的線程.
        if (node.thread != null)
            //6.2取消node中的線程引用,避免外面的線程嘗試喚醒.
            node.thread = null;       // avoid need for unpark()
        if (node.wasInterrupted && !node.interruptible)
            //6.3如果node本身設置了不可被擾動,但5.2處判斷線程本身拋出了擾動異常,卻被catch住了,此處擾動本線程.
            Thread.currentThread().interrupt();
        if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
            //6.4發現phase并未前進.還是參數傳入的pahse,說明一定是擾動或超時的結果,abortWait對本phase使用的隊列進行清理,
            //而清理的目標前面已論述過,是本隊列頭部開始的早于本phase的元素.(發現一個不滿足條件的就停止了清理).
            return abortWait(phase); // possibly clean up on abort
    }
    //7.退出上面的while循環一定會到此幫助釋放早于最新階段的waiter.注意,是早于最新phase的,參數phase只是決定了選哪個隊列(奇偶).
    //如果是6.4代表的那種擾動超時情況,此處其實釋放的是舊的結果.被喚醒的線程其實一般是執行在5.1處阻塞的.當前線程能運行到此絕對不需要喚醒.
    releaseWaiters(phase);
    return p;
}

到此Phaser的代碼解析已完畢,我們來分析關于隊列,等待和喚醒的問題.

1.Phaser維護了兩個"隊列",不論加入等待隊列還是彈出等待隊列,都是從頭部進行,新加入的成員會成功隊列的新頭,原來的頭會成為它的next,彈出時next成為新頭.所以相當于一個對頭部的"后進先出",考慮官方起名和注釋,我們依舊保持隊列這個稱呼.

2.喚醒時,會從隊列的頭部依次彈出node?的phase早于root的最新phase的node,.

3.等待時,入隊的node成為新的頭.

4.當輪次增加時,會使用和本輪不同的隊列增加元素,同時也會喚醒本輪中等待的node.

因為喚醒和等待同時進行,且各自操作各自的隊列(不同的phase),因此彼此之間沒有競態(盡管一個是頭入一個是頭出),可以說設計巧妙,下面我們來腦洞大開,思考一個極端情況.

我們假設一種極端的phase切換場景,奇數phase大量等待入隊,偶數phase則迅速完成.假設當前phase對應的隊列是奇數對列,輪次提升完成后,它去釋放當前的隊列元素,結果未等這個釋放操作執行完畢,偶數隊列的輪次很快執行完,奇數隊列中積壓了成千上萬個node未能釋放,輪次卻又切回到了奇數隊列,會出現什么事?

顯然奇數隊列如果一直保持這種極端場景,它會越來越龐大,逼近撐爆內存的同時,大量線程也會得不到釋放,甚至于老一輪的線程需要等待新一輪的線程去釋放.為什么老一輪的線程會去等待新一輪的線程釋放呢?

releaseWaiter的方法我們已經看出,它只會釋放phase早于最新的node,此時最新壓入的元素屬于當前最新的phase,顯然不滿足條件,那么會造成奇數隊列中兩輪前壓入的元素不能得到清除,兩輪前就在釋放當時積壓node的線程(那一輪最后一個arrive)發現不符合清理條件,就直接return并終止了,只能等待本輪最后一個arrive出現后繼續進行釋放.如果本輪最后一個arrive出現很晚,在下一輪依舊保持如此極端,往返數輪,確實會導致奇數隊列中積壓大量node,且第一輪就在等待該輪次結束的線程早就滿足了釋放條件(升到了2輪),事實上可能是第n輪才得到釋放,這還符合Phaser的定義嗎?我們使用它,就是要保證每一輪多帶帶使用,每一輪次達到條件,線程釋放并執行,下一輪次是下一輪次.

然而doug的代碼就是這個樣子,想遍各種極端,覺得可能找到了bug,那么就需要仔細思考了.作者來簡述一下這個趟坑的分析過程.

這個問題確實已經得到了極大的規避了,畢竟是個極端情況.

1.線程的喚醒真的很快,盡管此處除了喚醒還包含了原子引用的更新(每次出隊都要cas).

2.如果沒有注冊,顯然就沒有arrive相關的情況,盡管可以多帶帶調用,但必須保證在arrive時傳入的數量此時已經注冊了,因此每一輪次(phase)中可能積壓等待喚醒的線程的操作一定是在注冊之后,但是我們回憶一下,注冊方法的第一步就是要等待完成advance,而且傳給internalAwaitAdvance的node會是null,即不能擾動和超時,所以當本輪次阻塞了一定數量的線程后,如果不去arrive,也不考慮超時和擾動的情況,那么線程將一直阻塞.我們不可能在輪次advance前進行注冊,也就不可能在advance之前進行新一phase的arrive.

3.當本輪次的最后一個arrive線程觸發了輪次的更新后,才可以開啟注冊以及新輪次的arrive,但是此時使用了另一個等待隊列,而觸發了輪次更新的上一輪的arrive線程將會立即進行前一個隊列中積壓的線程的喚醒操作.只有該喚醒操作足夠慢,且新的輪次極快就完成了的情況,才可能造成在原arrive線程未能及時釋放奇數隊列的情況下,新一輪次再次向其中添加元素.

4.最重要的還在上面的internalAwaitAdvance方法,那一段被作者標上了入隊條件的注釋處,要想入隊,必須if ((q == null || q.phase == phase) &&加上后面的條件,而這兩個條件的限定已經很明顯,要想入隊,必須滿足該等待隊列沒有元素或者隊首是本輪的元素,而該方法又是下一輪首次注冊時必須等待完成的,下一輪的arrive又必須發生在下一輪的首次注冊之后,因此根本不會出現本輪wait的線程還要等下一輪甚至下N輪的線程去釋放的極端情況,哪怕真的去做一個極端測試:讓奇數輪大量積壓線程,讓偶數輪快速切換,然后測試第一輪壓入的線程到底是不是本輪釋放的.(作者差點就要寫單元測試去做這個極端測試了!)

這一段不經意的if,一個小小的條件,如果不注意真的忽略了,小代碼大功效,誰能想到,這么深的暗坑就這樣被規避了.

前面已經詳述了Phaser的源碼以及若干趟坑辛路.其實已經沒什么好總結的了,就在此順便對比常見同步器CyclicBarrier,CountDownLatch,Semaphore的特征和實現.

從使用特征上看:

1.CountDownLatch是一次性的,只能初始化決定parties數量,等待者可以是多個,每次釋放都會減少一個信號量,直到歸0時為止,最后一個釋放者將喚醒其他等待的線程.它也不能繼續使用.

2.CyclicBarrier是可重用的,分代的,每一代之間彼此獨立,但是每一代的初始parties是相同的,不可在運行期內動態調整,每一代最后一個線程會去開啟一下代,并可以在此時運行一個用戶指定的action,與此同時喚醒其他線程繼續執行.它可以在運行完一代后繼續被使用.并且它還支持重置.

3.Semaphore是一個資源量的典型,如果說CountDownLatch和CyclicBarrier或者Phaser都是等到"人夠了"再放行,Semaphore卻是起到限流的作用,它控制了有限的令牌數,這個數量不可以動態地更改,在不能acquire到足夠的令牌數時,線程將阻塞,直到其他線程釋放了足量的令牌數并喚醒它為止.每一個持有了令牌的線程都可以喚醒阻塞等待獲取的線程.

4.Phaser的功能上不同很明顯,首先它的參與者數量幾乎時刻可變(除了正在進入下一phase期間),隨時可以增加減少parties數量,每一phase等待者可以是多個,每一phase中,每個能從internalAwaitAdvance方法中走出循環的線程都可以幫助喚醒,當然最終能進入喚醒操作還是要歸功于最后一個arrive的線程(盡管它arrive后其他線程醒來后也會幫助喚醒).Phaser的喚醒者不一定是參與者.

從實現來看:

1.CountDownLatch借助了aqs來實現parties的釋放,它使用cas+park的方式,不使用Lock.

2.CyclicBarrier需要借助重入鎖和condition,每一個await的線程都要全局加鎖,阻塞時await在condition上.

3.Semaphore在實現上類似CountDownLatch,也是基于aqs,只不過它允許獲取和釋放,對state有增有減,總量不變.也是cas+park的方式阻塞,也不使用Lock

4.Phaser因為功能的要求,不基于AQS(它不能有構建時就固定的state,盡管可以初始化一個state,但它必須支持改變),它依托于原子引用實現了一個內部的隊列,相應的等待/入隊/喚醒等操作通過cas自旋+park的方式,同樣不使用Lock.并利用雙隊列的方式規避了前一輪的釋放和后一輪的響醒的阻塞.

此外還有兩點結合前面的推理和自測驗證的結論:

1.Phaser中的每一個phase是保證了可見性的,經作者自測,在任何使用Phaser的代碼中await前后,不會出現串phase讀出的亂序情況(側面說明每個phase不會依賴后一個或幾個phase的釋放).

2.Phaser需要對await的線程進行阻塞時,是將它打包成一個node(blocker),利用ForkJoinPool來block的.如果使用Phaser同步的任務是運行在ForkJoinPool中的,它將會利用到相應的補償機制,經作者自測,這將保證Phaser中block的每一個任務必然得到執行,每一個阻塞的線程必然得到釋放.