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

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

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

本文主要介紹Phaser的內(nèi)部實(shí)現(xiàn),粗略介紹使用,它的源碼相比于線程池較為簡(jiǎn)單,但最好能對(duì)比其他同步器來了解,讀者最好擁有juc其他同步器,原子類,部分ForkJoin框架的基礎(chǔ).

同時(shí),本文也會(huì)再次提到ForkJoinPool::managedBlock(blocker),之前在ForkJoinPool一文提到了實(shí)現(xiàn)和接口,而在CompletableFuture中見到了一個(gè)blocker的實(shí)現(xiàn).

首先來看一些與Phaser狀態(tài)有關(guān)的簡(jiǎn)單的常量.

//64位整數(shù)表示Phaser的狀態(tài).
private volatile long state;

private static final int  MAX_PARTIES     = 0xffff;//最大parties,后16位表示.
private static final int  MAX_PHASE       = Integer.MAX_VALUE;//最大phase,最大整數(shù)值.
private static final int  PARTIES_SHIFT   = 16;//取parties使用的移位數(shù),16
private static final int  PHASE_SHIFT     = 32;//取phase的移位數(shù),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; //數(shù)量,后32位.
private static final long TERMINATION_BIT = 1L << 63;//終止態(tài),首位.

// 特殊值.
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位.顯然,它表示了一個(gè)ONE_ARRIVAL信息和PARTY信息.
private static final int  EMPTY           = 1;

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

//對(duì)一個(gè)state,取出parties信息,直接取state的17至32位.
private static int partiesOf(long s) {
    return (int)s >>> PARTIES_SHIFT;
}
//對(duì)于一個(gè)state,取出phase信息,直接取前32位.
private static int phaseOf(long s) {
    return (int)(s >>> PHASE_SHIFT);
}
//對(duì)于一個(gè)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);
}

上面都是一些常量,沒什么可分析的,簡(jiǎn)單來個(gè)總結(jié).

Phaser用一個(gè)long型的state保存狀態(tài)信息.

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

下面我們看一些成員變量和有關(guān)函數(shù).

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

//phaser顯然是個(gè)樹的結(jié)果,root代表根,如果當(dāng)前phaser不在樹內(nèi),則root==this
private final Phaser root;


//偶數(shù)隊(duì)列和奇數(shù)隊(duì)列.它們存放等待線程棧的頭,為了減少當(dāng)添加線程與釋放線程的競(jìng)態(tài),
//這里使用了兩個(gè)隊(duì)列并互相切換,子phaser共享root的隊(duì)列以加快釋放.
private final AtomicReference evenQ;
private final AtomicReference oddQ;

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

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

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

為了便于理解,先來看隊(duì)列的實(shí)現(xiàn).

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

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

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

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

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

關(guān)于該方法的執(zhí)行流程,我們結(jié)合幾個(gè)周邊方法一并分析,先來看注冊(cè)方法和onAdvance勾子.

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

    }
    return phase;
}


//使用root的phase調(diào)整this的state,更新滯后的結(jié)果.這一般發(fā)生在root前進(jìn)了phase但是
//子phaser還沒有做到這一步,這種情況下,子phaser必須完成這個(gè)前進(jìn)的步驟,這一過程中,phase將
//被置為root的phase,unarrived則會(huì)重置為parties,若parties為0,則置為EMPTY.返回結(jié)果state.
private long reconcileState() {
    final Phaser root = this.root;
    long s = state;
    //不是root才進(jìn)行下一步.
    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勾子方法,參數(shù)為當(dāng)前phase和注冊(cè)的parties數(shù).
//默認(rèn)實(shí)現(xiàn)為parties數(shù)為0,方法返回true時(shí),調(diào)用者會(huì)嘗試終止Phaser.(參考前面的doArrive).隨后調(diào)用isTerminated方法將返回true.
//執(zhí)行此方法時(shí)拋出的運(yùn)行時(shí)異?；駿rror將直接上拋給嘗試advance相應(yīng)的phase的線程,這種情況下不會(huì)發(fā)生phase的advance.
//方法的入?yún)⒈硎镜氖荘haser當(dāng)前的state(未advance前),因此若在onAdvance方法中執(zhí)行arrive,regist,waiting這三種操作的行為是不確定的也不可靠的.
//如果當(dāng)前Phaser是一個(gè)級(jí)聯(lián)的成員,那么onAdvance只會(huì)由root在每次advance時(shí)調(diào)用.
//方法的默認(rèn)實(shí)現(xiàn)返回true的場(chǎng)景目前只能是經(jīng)過數(shù)次arriveAndDeregister調(diào)用造成parties歸零的結(jié)果.我們繼承Phaser可以輕易地重寫此行為,
//比如簡(jiǎn)單粗暴地返回false,那么將永遠(yuǎn)允許新的注冊(cè).
protected boolean onAdvance(int phase, int registeredParties) {
    return registeredParties == 0;
}

經(jīng)過前面的代碼分析,已經(jīng)對(duì)Phaser的核心函數(shù)doRegister,doArrive有了全面的了解.

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

doArrive方法會(huì)每次調(diào)整unarrived數(shù)量(也可包含parties數(shù)量,如果使用了解除注冊(cè)),當(dāng)Phaser調(diào)用自身的arrive/arriveAndDeregister時(shí),會(huì)做出相應(yīng)的減少,并根據(jù)是否為root而決定向上遞歸.

Phaser減少自身unarrived信號(hào)(也可能同時(shí)有parties信號(hào))后,若發(fā)現(xiàn)這已經(jīng)是最后一個(gè)unarrived信號(hào),則進(jìn)行接下來的判斷:

1.當(dāng)前Phaser是root,advance并喚醒waiter.(重要的喚醒操作執(zhí)行點(diǎn),root一輪完成)

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

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

顯然,子Phaser的最后一個(gè)unarrived的消失一定會(huì)造成父的unarrived減少,子Phaser不能繼續(xù)下一phase的register和arrive時(shí),從父Phaser中卸載.

若不是本Phaser的最后一個(gè)unarrived信號(hào),則直接結(jié)束,相當(dāng)于只進(jìn)行了上面的減少信號(hào)操作.

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

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

2.當(dāng)前注冊(cè)并非首次注冊(cè),unarrived也不是0,則在本phase進(jìn)行注冊(cè),增加相應(yīng)的parties和unarrived.

3.當(dāng)前注冊(cè)是root的首次注冊(cè),給root的state加上相應(yīng)的parties和unarrived.

4.當(dāng)前注冊(cè)是非root的首次注冊(cè),加鎖(this),對(duì)自己的state加上相應(yīng)的parties和unarrived(同上,以registrations為單位),而對(duì)parent加上一個(gè)parties和unarrived單位.

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

當(dāng)我們同時(shí)也嘗試減少parties數(shù)量,即解除parties的注冊(cè),最終導(dǎo)致沒有parties,那么Phaser將進(jìn)入終止態(tài).

整個(gè)過程中,只要Phaser沒進(jìn)入終止態(tài),隨時(shí)可以進(jìn)行新的注冊(cè),并增加parties和unarrived的數(shù)量.每個(gè)arrive可以減少unarrived的數(shù)量為任何正整數(shù),不一定是1.

對(duì)于多Phaser的情況,有兩個(gè)特殊點(diǎn):

1.對(duì)任意Phaser樹中的某一個(gè)Phaser調(diào)用注冊(cè)操作,會(huì)令自身加上相應(yīng)參數(shù)個(gè)parties和unarrived單位,僅會(huì)在該P(yáng)haser第一次注冊(cè)時(shí)增加父Phaser(極端可能,僅從一個(gè)葉子節(jié)點(diǎn)第一個(gè)注冊(cè)的情況下可一直遞歸到root)的parties數(shù)和unarrived數(shù)各1單位(不論參數(shù)是多少).

2.對(duì)任意Phaser樹中的某一個(gè)Phaser調(diào)用arrive操作,會(huì)令自身減去相應(yīng)的參數(shù)個(gè)parties和unarrived單位,同時(shí)僅當(dāng)本Phaser此時(shí)是最后一個(gè)unarrived時(shí),會(huì)減去父Phaser的一個(gè)unarrived單位(當(dāng)前子Phaser仍舊有parties可以構(gòu)建下一phase),或減去父Phaser一個(gè)Parties和unarrived單位.(極端情況下,每一級(jí)都是最后一個(gè)unarrived時(shí),減少葉子節(jié)點(diǎn)的最后一個(gè)unarrived會(huì)遞歸到root).

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

目前沒有看到await方法,但可以提前說明,等待操作完全依賴于root是否完成本輪.也就是所有子Phaser都完成了同一輪(arrive打滿),才能讓父Phaser本身減去一個(gè)所有arrive單位,再觸發(fā)父Phaser本輪的完成,此時(shí)對(duì)任何已完成的Phaser進(jìn)入注冊(cè),都會(huì)進(jìn)入上述的root.internalAwaitAdvance(phase, null)方法等待root進(jìn)入下一phase.如果對(duì)已經(jīng)完成所有arrive的Phaser繼續(xù)進(jìn)行arrive操作,因?yàn)閡narrived已經(jīng)是0,則會(huì)拋出異常.

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

好在我們使用Phaser可能會(huì)類似CyclicBarrier的使用方式,可對(duì)每一輪(phase)進(jìn)行注冊(cè)并等待(也許只等一輪,那么arrive就要帶上deregister),每一輪最后一個(gè)線程arrive了,就會(huì)停止所有線程的等待,讓所有線程繼續(xù)執(zhí)行,同時(shí)開啟了下一輪次,這些線程此時(shí)又可以不經(jīng)注冊(cè)直接在新的輪次中進(jìn)行等待,直到最后一個(gè)arrive了,再次喚醒所有線程并繼續(xù)執(zhí)行,同時(shí)Phaser再前進(jìn)一輪,如此往復(fù).中間使用arrive并deregister的線程會(huì)從本輪起減少一個(gè)unarrive數(shù)量(因?yàn)閜arties也減少了,所以再下一輪初始化unarrive數(shù)量時(shí)也會(huì)減少一次).我們可以讓這些線程參與任意的輪次,但要注意的是,如果有線程中途不參加了,一定要解除注冊(cè),否則因?yàn)槊枯喅跏蓟瘯r(shí),要等待arrive的數(shù)量都是上一輪剩下的parties數(shù)量,有線程停止了執(zhí)行,卻不減少parties數(shù),那么下輪所有等待的線程將永遠(yuǎn)等不到phaser滿足喚醒的條件.

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

現(xiàn)在假設(shè)一個(gè)特殊的使用場(chǎng)景,也可以區(qū)別于CyclicBarrie和CountDownLatch的使用.還是上面的例子,但是我們準(zhǔn)備的線程數(shù)與Phaser的parties數(shù)/unarrived數(shù)不同(一般前者要多些),會(huì)發(fā)生什么事?

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

目前這些還比較抽象,等到我們看過相應(yīng)的幾個(gè)方法便了然了.

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

接下來看全量構(gòu)造方法和若干和上面有關(guān)的公有函數(shù).

//初始化一個(gè)Phaser,指定parent,指定未到來的參與者數(shù)(unarrived parties),但這只是一個(gè)初值,
//當(dāng)我們?cè)谌魏螘r(shí)候調(diào)用注冊(cè)方法時(shí),還會(huì)相應(yīng)的增加.
public Phaser(Phaser parent, int parties) {
    if (parties >>> PARTIES_SHIFT != 0)
        //太大了,超過了后16位能表示的整數(shù).
        throw new IllegalArgumentException("Illegal number of parties");
    //初始phase為0.
    int phase = 0;
    this.parent = parent;
    if (parent != null) {
        //1.有parent的情況,共享parent的root,隊(duì)列,并向parent中注冊(cè)1個(gè)parties和unarrived,
        //同時(shí)同步一次phase(表面上是同步了parent的,實(shí)際上前面已經(jīng)看過,會(huì)同步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,并初始化奇偶等待隊(duì)列.它使用原子引用的形式存放一個(gè)QNode,而QNode我們前面已介紹.
        this.root = this;
        this.evenQ = new AtomicReference();
        this.oddQ = new AtomicReference();
    }
    //統(tǒng)一初始化state,后32位的決定依托于parties,如果parties是0則給予EMPTY,直接不管高32位.
    //不為0則給予phase設(shè)置為前32位,parties設(shè)置parties位和unarrived位.
    this.state = (parties == 0) ? (long)EMPTY :
        ((long)phase << PHASE_SHIFT) |
        ((long)parties << PARTIES_SHIFT) |
        ((long)parties);
}


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


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


//arrive一個(gè)信號(hào),不等待其他arrive事件,返回最新phase(終止態(tài)為負(fù)).
//當(dāng)前Phaser的arrive事件已滿,則對(duì)parent來說也會(huì)觸發(fā)一個(gè)arrive.(如果有parent)
public int arrive() {
    return doArrive(ONE_ARRIVAL);
}


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

接下來要看上面已經(jīng)做足了鋪墊的等待方法了,并結(jié)合前面的隊(duì)列一塊看.

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


//等待當(dāng)前Phaser從給定的phase前進(jìn)結(jié)束,如果當(dāng)前phase不等于給定的phase,或者Phaser已終止立即返回.
//1.傳入phase為負(fù),返回它本身.
//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前進(jìn).參數(shù)node為null
        return root.internalAwaitAdvance(phase, null);
    return p;
}


//參考前面的幾個(gè)方法,區(qū)別是可擾動(dòng).
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.參數(shù)phase小于0直接返回它本身.
        return phase;
    if (p == phase) {
        //2.參數(shù)phase匹配,回憶一個(gè)前面介紹的QNode,匹配當(dāng)前Phaser和phase,配置為可擾動(dòng)且不計(jì)時(shí).
        QNode node = new QNode(this, phase, true, false, 0L);
        //3.放入root的等待隊(duì)列阻塞.
        p = root.internalAwaitAdvance(phase, node);
        if (node.wasInterrupted)
            //4.等待結(jié)束,判斷是否是擾動(dòng)造成的結(jié)束,前面介紹過QNode的相關(guān)邏輯,
            //它實(shí)現(xiàn)了ForkJoinPool.ManagedBlocker,因此在managedBlock方法進(jìn)行時(shí),
            //會(huì)循環(huán)調(diào)用問詢是否能release,當(dāng)我們配置了可擾動(dòng)且擾動(dòng)了,就會(huì)標(biāo)記這個(gè)wasInterrupted,釋放線程引用并返回.
            //發(fā)現(xiàn)此種情況拋出異常.
            //同時(shí),當(dāng)發(fā)現(xiàn)等待成功,也會(huì)結(jié)束,釋放線程引用并返回,但不帶有擾動(dòng)標(biāo)記.
            throw new InterruptedException();
    }
    //5.返回1處之前讀取的phase或3處得到的最新phase值.
    return p;
}

//同上方法,但帶有計(jì)時(shí).
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帶有計(jì)時(shí)和等待時(shí)長(zhǎng).
        QNode node = new QNode(this, phase, true, true, nanos);
        p = root.internalAwaitAdvance(phase, node);
        if (node.wasInterrupted)
            //被擾動(dòng)的情況.
            throw new InterruptedException();
        else if (p == phase)
            //時(shí)間到了phase沒有前進(jìn),超時(shí).
            throw new TimeoutException();
    }
    return p;
}

前面的幾個(gè)核心方法粗略過完,補(bǔ)充一些重要內(nèi)容.

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

建立QNode可以限定是否定時(shí)和可擾動(dòng),這取決于我們使用哪個(gè)方法去await.

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

QNode前面已介紹,它是一個(gè)blocker,需要調(diào)用ForkJoinPool::managedBlock才會(huì)起作用(顯然root的internalAwaitAdvance必然與此方法有關(guān)聯(lián)).當(dāng)然這個(gè)作用與任務(wù)是否運(yùn)行在ForkJoinPool無關(guān),如果等待phaser前進(jìn)的線程是運(yùn)行在ForkJoinPool中的ForkJoinWorkerThread,顯然會(huì)在internalAwaitAdvance期間進(jìn)行補(bǔ)償.這一塊可參考前面的"CompletableFuture與響應(yīng)式編程"和"ForkJoin框架之ForkJoinPool"兩篇文章.

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

其中(1)也間接說明了為什么構(gòu)建Phaser時(shí)只有root創(chuàng)建等待隊(duì)列,所有子Phaser共享.

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

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


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

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

//查詢已經(jīng)arrived的parties數(shù)量.調(diào)用介紹過的arriveOf

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

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

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

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

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

這些方法都比較簡(jiǎn)單,只有forceTermination需要再?gòu)?qiáng)調(diào)一翻,前面介紹arrayAndAwaitAdvance時(shí)曾提過在減去最后一個(gè)unarrived信號(hào)后去cas到下一個(gè)phase失敗的情況,doug大神簡(jiǎn)單注釋了一句terminated,直接返回了當(dāng)前的phase(顯然只能是負(fù)),在周邊方法重重加鎖的前提下,那一次cas的失敗唯一一處就是強(qiáng)制關(guān)閉,因?yàn)樗桓年P(guān)閉標(biāo)記位,相當(dāng)于動(dòng)了phase,而沒有動(dòng)unarrived標(biāo)記位和parties標(biāo)記位.所以重寫Phaser的方法要謹(jǐn)慎,很可能不小心打破了這個(gè)封裝.

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

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

//工具方法,移除某個(gè)phase的等待者.
private void releaseWaiters(int phase) {
    QNode q;  //保存隊(duì)列中的隊(duì)首
    Thread t;  // 保存線程引用.
    //取隊(duì)列,用phase的奇偶決定,phase是偶數(shù)就取偶數(shù)隊(duì)列,否則取奇數(shù)隊(duì)列.而這個(gè)phase其實(shí)只用來取隊(duì)列了,后續(xù)的操作與它無關(guān).
    AtomicReference head = (phase & 1) == 0 ? evenQ : oddQ;
    //循環(huán),找出所有phase不等于root的phase的(其實(shí)root是最大的,所以就是找出非最新phase加入進(jìn)來的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) {
            //發(fā)現(xiàn)阻塞者,喚醒線程.回憶下前面實(shí)現(xiàn)blocker方法中的isReleaseble和block方法都有將線程置空的操作.(三種情況,喚醒擾動(dòng)超時(shí)都會(huì)置空)
            //但是那些方法并沒有將代表該阻塞線程的QNode移除隊(duì)列,因此可能會(huì)發(fā)現(xiàn)thread已經(jīng)是null(代表無阻塞者)的情況,只需要移除隊(duì)列即可.
            q.thread = null;
            LockSupport.unpark(t);
        }
    }
}


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

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

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


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

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

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

2.喚醒時(shí),會(huì)從隊(duì)列的頭部依次彈出node?的phase早于root的最新phase的node,.

3.等待時(shí),入隊(duì)的node成為新的頭.

4.當(dāng)輪次增加時(shí),會(huì)使用和本輪不同的隊(duì)列增加元素,同時(shí)也會(huì)喚醒本輪中等待的node.

因?yàn)閱拘押偷却瑫r(shí)進(jìn)行,且各自操作各自的隊(duì)列(不同的phase),因此彼此之間沒有競(jìng)態(tài)(盡管一個(gè)是頭入一個(gè)是頭出),可以說設(shè)計(jì)巧妙,下面我們來腦洞大開,思考一個(gè)極端情況.

我們假設(shè)一種極端的phase切換場(chǎng)景,奇數(shù)phase大量等待入隊(duì),偶數(shù)phase則迅速完成.假設(shè)當(dāng)前phase對(duì)應(yīng)的隊(duì)列是奇數(shù)對(duì)列,輪次提升完成后,它去釋放當(dāng)前的隊(duì)列元素,結(jié)果未等這個(gè)釋放操作執(zhí)行完畢,偶數(shù)隊(duì)列的輪次很快執(zhí)行完,奇數(shù)隊(duì)列中積壓了成千上萬個(gè)node未能釋放,輪次卻又切回到了奇數(shù)隊(duì)列,會(huì)出現(xiàn)什么事?

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

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

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

這個(gè)問題確實(shí)已經(jīng)得到了極大的規(guī)避了,畢竟是個(gè)極端情況.

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

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

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

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

這一段不經(jīng)意的if,一個(gè)小小的條件,如果不注意真的忽略了,小代碼大功效,誰(shuí)能想到,這么深的暗坑就這樣被規(guī)避了.

前面已經(jīng)詳述了Phaser的源碼以及若干趟坑辛路.其實(shí)已經(jīng)沒什么好總結(jié)的了,就在此順便對(duì)比常見同步器CyclicBarrier,CountDownLatch,Semaphore的特征和實(shí)現(xiàn).

從使用特征上看:

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

2.CyclicBarrier是可重用的,分代的,每一代之間彼此獨(dú)立,但是每一代的初始parties是相同的,不可在運(yùn)行期內(nèi)動(dòng)態(tài)調(diào)整,每一代最后一個(gè)線程會(huì)去開啟一下代,并可以在此時(shí)運(yùn)行一個(gè)用戶指定的action,與此同時(shí)喚醒其他線程繼續(xù)執(zhí)行.它可以在運(yùn)行完一代后繼續(xù)被使用.并且它還支持重置.

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

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

從實(shí)現(xiàn)來看:

1.CountDownLatch借助了aqs來實(shí)現(xiàn)parties的釋放,它使用cas+park的方式,不使用Lock.

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

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

4.Phaser因?yàn)楣δ艿囊?不基于AQS(它不能有構(gòu)建時(shí)就固定的state,盡管可以初始化一個(gè)state,但它必須支持改變),它依托于原子引用實(shí)現(xiàn)了一個(gè)內(nèi)部的隊(duì)列,相應(yīng)的等待/入隊(duì)/喚醒等操作通過cas自旋+park的方式,同樣不使用Lock.并利用雙隊(duì)列的方式規(guī)避了前一輪的釋放和后一輪的響醒的阻塞.

此外還有兩點(diǎn)結(jié)合前面的推理和自測(cè)驗(yàn)證的結(jié)論:

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

2.Phaser需要對(duì)await的線程進(jìn)行阻塞時(shí),是將它打包成一個(gè)node(blocker),利用ForkJoinPool來block的.如果使用Phaser同步的任務(wù)是運(yùn)行在ForkJoinPool中的,它將會(huì)利用到相應(yīng)的補(bǔ)償機(jī)制,經(jīng)作者自測(cè),這將保證Phaser中block的每一個(gè)任務(wù)必然得到執(zhí)行,每一個(gè)阻塞的線程必然得到釋放.