建站
咨詢
售后
建站咨詢
新聞中心
前言
成都創(chuàng)新互聯(lián)公司專注于企業(yè)營(yíng)銷型網(wǎng)站建設(shè)、網(wǎng)站重做改版、久治網(wǎng)站定制設(shè)計(jì)、自適應(yīng)品牌網(wǎng)站建設(shè)、H5頁(yè)面制作、商城網(wǎng)站開(kāi)發(fā)、集團(tuán)公司官網(wǎng)建設(shè)、外貿(mào)營(yíng)銷網(wǎng)站建設(shè)、高端網(wǎng)站制作、響應(yīng)式網(wǎng)頁(yè)設(shè)計(jì)等建站業(yè)務(wù)，價(jià)格優(yōu)惠性價(jià)比高，為久治等各大城市提供網(wǎng)站開(kāi)發(fā)制作服務(wù)。
JDK5中引入了CyclicBarrier和CountDownLatch這兩個(gè)并發(fā)控制類，而JDK7中引入的Phaser按照官方的說(shuō)法是提供了一個(gè)功能類似但是更加靈活的實(shí)現(xiàn)。接下來(lái)我們帶著幾個(gè)問(wèn)題來(lái)研究一下Phaser與(CountDownLath、CyclicBarrier)到底有哪些類似，同時(shí)帶來(lái)了哪些靈活性?
- Phaser 是什么?
- Phaser 具有哪些特性?
- Phaser相對(duì)于 CyclicBarrier 和 CountDownLatch的優(yōu)勢(shì)?
CyclicBarrier和CountDownLatch
CyclicBarrier介紹
在使用CyclicBarrier時(shí)，需要?jiǎng)?chuàng)建一個(gè)CyclicBarrier對(duì)象，構(gòu)造函數(shù)需要一個(gè)整數(shù)作為參數(shù)，這個(gè)參數(shù)是一個(gè)“目標(biāo)”，在CyclicBarrier對(duì)象創(chuàng)建后，內(nèi)部會(huì)有一個(gè)計(jì)數(shù)器，初始值為0，CyclicBarrier對(duì)象的await方法每被調(diào)用一次，這個(gè)計(jì)數(shù)器就會(huì)加1，一旦這個(gè)計(jì)數(shù)器的值達(dá)到設(shè)定的“目標(biāo)”，所有被CyclicBarrier.await阻塞住的線程都會(huì)繼續(xù)執(zhí)行。這個(gè)目標(biāo)是固定的，一旦設(shè)定便不能修改。
舉一個(gè)例子，假設(shè)有5個(gè)人爬香山，他們要爬到山頂，等到5個(gè)人到齊了再同時(shí)出發(fā)下山，那么我們要在山頂設(shè)定一個(gè)“目標(biāo)”，同時(shí)還有一個(gè)計(jì)數(shù)器，這個(gè)目標(biāo)就是5，每一個(gè)人到山頂后，這個(gè)人就要等待，同時(shí)計(jì)數(shù)器加1，等到5個(gè)人到齊了，也就是計(jì)數(shù)器達(dá)到了這個(gè)“目標(biāo)”，所有等待的人就開(kāi)始下山了。 更多內(nèi)容請(qǐng)閱讀《并發(fā)編程之CyclicBarrier原理與使用》
CountDownLathch介紹
使用CountDownLatch時(shí)，需要?jiǎng)?chuàng)建一個(gè)CountDownLatch對(duì)象，構(gòu)造函數(shù)也需要一個(gè)整數(shù)作為參數(shù)，可以把這個(gè)參數(shù)想象成一個(gè)倒計(jì)時(shí)器，CountDownLatch對(duì)象本身是一個(gè)發(fā)令槍，所有調(diào)用CountDownLatch.await方法的線程都會(huì)等待發(fā)令槍的指令，一旦倒計(jì)時(shí)器為0，這些線程同時(shí)開(kāi)始執(zhí)行，而CountDownLatch.countDown方法就是為倒計(jì)時(shí)器減1。
更多內(nèi)容請(qǐng)閱讀《并發(fā)編程之CountDownLatch原理與使用》
對(duì)比分析
CyclicBarrier和CountDownLatch的共同點(diǎn)都是有一個(gè)目標(biāo)和一個(gè)計(jì)數(shù)器，等到計(jì)數(shù)器達(dá)到目標(biāo)后，所有阻塞的線程都將繼續(xù)執(zhí)行。它們的不同點(diǎn)是CyclicBarrier.await在等待的同時(shí)還修改計(jì)數(shù)器，而CountDownLatch.await只負(fù)責(zé)等待，CountDownLatch.countDown才修改計(jì)數(shù)器。
CountDownLatch和CyclicBarrier都能夠?qū)崿F(xiàn)線程之間的等待，只不過(guò)它們側(cè)重點(diǎn)不同：
- CountDownLatch一般用于一個(gè)或多個(gè)線程，等待其他線程執(zhí)行完任務(wù)后，再才執(zhí)行;
- CyclicBarrier一般用于一組線程互相等待至某個(gè)狀態(tài)，然后這一組線程再同時(shí)執(zhí)行;
- CountDownLatch 是一次性的，CyclicBarrier 是可循環(huán)利用的;
- CountDownLathch是一個(gè)計(jì)數(shù)器，線程完成一個(gè)記錄一個(gè)，計(jì)數(shù)器遞減，只能用一次。如下圖：
CyclicBarrier的計(jì)數(shù)器更像一個(gè)閥門，需要所有線程都到達(dá)，然后繼續(xù)執(zhí)行，計(jì)數(shù)器遞減，提供reset功能，可以多次使用。如下圖：
Phaser是什么?
Phaser，翻譯為移相器(階段)，它適用于這樣一種場(chǎng)景，一個(gè)大任務(wù)可以分為多個(gè)階段完成，且每個(gè)階段的任務(wù)可以多個(gè)線程并發(fā)執(zhí)行，但是必須上一個(gè)階段的任務(wù)都完成了才可以執(zhí)行下一個(gè)階段的任務(wù)。
這種場(chǎng)景雖然使用CyclicBarrier 或者 CountDownLatch 也可以實(shí)現(xiàn)，但是要復(fù)雜的多，首先，具體需要多少個(gè)階段是可能變的，其次，每個(gè)階段的任務(wù)數(shù)也可能會(huì)變的。相比于CyclicBarrier 和 CountDownLath ，Phaser更加靈活更加方便。
Phaser使用方法
Phaser同時(shí)包含CyclicBarrier和CountDownLatch兩個(gè)類的功能。
- Phaser的arrive方法將將計(jì)數(shù)器加1，awaitAdvance將線程阻塞，直到計(jì)數(shù)器達(dá)到目標(biāo)，這兩個(gè)方法與CountDownLatch的countDown和await方法相對(duì)應(yīng);
- Phaser的arriveAndAwaitAdvance方法將計(jì)數(shù)器加1的同時(shí)將線程阻塞，直到計(jì)數(shù)器達(dá)到目標(biāo)后繼續(xù)執(zhí)行，這個(gè)方法對(duì)應(yīng)CyclicBarrier的await方法。
除了包含以上兩個(gè)類的功能外，Phaser還提供了更大的靈活性。CyclicBarrier和CountdownLatch在構(gòu)造函數(shù)指定目標(biāo)后就無(wú)法修改，而Phaser提供了register和deregister方法可以對(duì)目標(biāo)進(jìn)行動(dòng)態(tài)修改。
下面看一個(gè)最簡(jiǎn)單的使用案例：
- package com.niuh.tools;
- import java.util.concurrent.Phaser;
- /**
- *
- * Phaser示例
- *
- */
- public class PhaserRunner {
- // 定義每個(gè)階段需要執(zhí)行3個(gè)小任務(wù)
- public static final int PARTIES = 3;
- // 定義需要4個(gè)階段完成的大任務(wù)
- public static final int PHASES = 4;
- public static void main(String[] args) {
- Phaser phaser = new Phaser(PARTIES) {
- @Override
- protected boolean onAdvance(int phase, int registeredParties) {
- System.out.println("==phase: " + phase + " finished==");
- return super.onAdvance(phase, registeredParties);
- }
- };
- for (int i = 0; i < PARTIES; i++) {
- new Thread(() -> {
- for (int j = 0; j < PHASES; j++) {
- System.out.println(String.format("%s: phase: %d", Thread.currentThread().getName(), j));
- phaser.arriveAndAwaitAdvance();
- }
- }, "Thread " + i).start();
- }
- }
- }
這里我們定義個(gè)需要4個(gè)階段完成的大任務(wù)，每個(gè)階段需要3個(gè)小任務(wù)，針對(duì)這些小任務(wù)，我們分別起3個(gè)線程來(lái)執(zhí)行這些小任務(wù)，查看輸出結(jié)果為：
- Thread 2: phase: 0
- Thread 0: phase: 0
- Thread 1: phase: 0
- ==phase: 0 finished==
- Thread 2: phase: 1
- Thread 1: phase: 1
- Thread 0: phase: 1
- ==phase: 1 finished==
- Thread 1: phase: 2
- Thread 2: phase: 2
- Thread 0: phase: 2
- ==phase: 2 finished==
- Thread 1: phase: 3
- Thread 0: phase: 3
- Thread 2: phase: 3
- ==phase: 3 finished==
可以看到，每個(gè)階段都是三個(gè)線程都完成來(lái)才進(jìn)入下一個(gè)階段。這是怎么實(shí)現(xiàn)的呢?
Phaser原理猜測(cè)
結(jié)合AQS的原理，大概猜測(cè)一下Phaser的實(shí)現(xiàn)原理：
- 首先，需要存儲(chǔ)當(dāng)前階段phase、當(dāng)前階段的任務(wù)數(shù)(參與者)parties、未完成參與者的數(shù)量，這三個(gè)變量我們可以放在一個(gè)變量state中存儲(chǔ)。
- 其次，需要一個(gè)隊(duì)列存儲(chǔ)先完成的參與者，當(dāng)最后一個(gè)參與者完成任務(wù)時(shí)，需要喚醒隊(duì)列中的參與者。
結(jié)合上面的案例帶入：初始時(shí)當(dāng)前階段為0，參與者為3個(gè)，未完成參與者數(shù)為3;
- 第一個(gè)線程執(zhí)行到 phaser.arriveAndAwaitAdvance(); 時(shí)進(jìn)入隊(duì)列;
- 第二個(gè)線程執(zhí)行到 phaser.arriveAndAwaitadvance(); 時(shí)進(jìn)入隊(duì)列;
- 第三個(gè)線程執(zhí)行到 phaser.arriveAndAwaitadvance(); 時(shí)先執(zhí)行這個(gè)階段的總結(jié) onAdvance(), 再喚醒簽名兩個(gè)線程繼續(xù)執(zhí)行下一個(gè)階段的任務(wù)。
基于這樣的一個(gè)思路，整體能說(shuō)的通，至于是不是這樣?讓我們一起來(lái)看源碼吧。
Phaser源碼分析
主要API
- register()，增加一個(gè)參與者，需要同時(shí)增加parties和unarrived兩個(gè)數(shù)值，也就是state中的16位和低16位
- onAdvance(int phase, int registeredParties)，當(dāng)前階段所有線程完成時(shí)，會(huì)調(diào)用OnAdvance()
- bulkRegister(int parties)，指定參與者數(shù)目注冊(cè)到Phaser中，同時(shí)增加parties和unarrived兩個(gè)數(shù)值
- arrive()，作用使parties值加1，并且不在屏障處等待，直接運(yùn)行下面的代碼
- awaitAdvance(int phase)，如果傳入的參數(shù)與當(dāng)前階段一致，這個(gè)方法會(huì)將當(dāng)前線程置于休眠，直到這個(gè)階段的參與者都完成運(yùn)行。如果傳入的階段參數(shù)與當(dāng)前階段不一致，立即返回
- arriveAndAwaitAdvance()，當(dāng)前線程當(dāng)前階段執(zhí)行完畢，等待其它線程完成當(dāng)前階段
- arriveAndDeregister()，當(dāng)一個(gè)線程調(diào)用來(lái)此方法時(shí)，parties將減1，并且通知這個(gè)線程已經(jīng)完成來(lái)當(dāng)前預(yù)警，不會(huì)參加到下一個(gè)階段中，因此Phaser對(duì)象在開(kāi)始下一個(gè)階段時(shí)不會(huì)等待這個(gè)線程。
- awaitAdvanceInterruptibly(int phase)，這個(gè)方法跟awaitAdvance(int phase)一樣，不同之處是，如果這個(gè)方法中休眠的線程被中斷，它將拋出InterruptedException異常。
- getPhase()，當(dāng)前階段
- getRegisteredParties()，總數(shù)
- getArrivedParties()，到達(dá)總數(shù)
- getUnarrivedParties()，未到達(dá)總數(shù)
內(nèi)部類QNode
QNode用來(lái)跟蹤當(dāng)前線程的信息的。QNode被組織成單向鏈表的形式。用來(lái)管理是否阻塞或者被中斷。
QNode繼承自ForkJoinPool.ManagedBlocker。ForkJoinPool來(lái)管理是否阻塞和中斷狀態(tài)。這里只需要重寫isReleasable和block。
- isReleaseable用于判斷是否釋放當(dāng)前節(jié)點(diǎn)。
- block用于阻塞。
- 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();
- }
- public boolean isReleasable() {
- if (thread == null)
- return true;
- if (phaser.getPhase() != phase) {
- thread = null;
- return true;
- }
- if (Thread.interrupted())
- wasInterrupted = true;
- if (wasInterrupted && interruptible) {
- thread = null;
- return true;
- }
- if (timed) {
- if (nanos > 0L) {
- nanos = deadline - System.nanoTime();
- }
- if (nanos <= 0L) {
- thread = null;
- return true;
- }
- }
- return false;
- }
- public boolean block() {
- if (isReleasable())
- return true;
- else if (!timed)
- LockSupport.park(this);
- else if (nanos > 0L)
- LockSupport.parkNanos(this, nanos);
- return isReleasable();
- }
- }
整體代碼比較簡(jiǎn)單。要注意的是在isReleasable中使用了thread=null來(lái)使得避免解鎖任務(wù)。使用方法類似于internalAwaitAdvance中的用法。先完成的參與者放入隊(duì)列中的節(jié)點(diǎn)，這里我們只需要關(guān)注 thread 和 next兩個(gè)屬性即可，很明顯這是一個(gè)單鏈表，存儲(chǔ)這入隊(duì)的線程。
主要屬性
- /*
- * unarrived -- 還沒(méi)有抵達(dá)屏障的參與者的個(gè)數(shù) (bits 0-15)
- * parties -- 需要等待的參與者的個(gè)數(shù) (bits 16-31)
- * phase -- 屏障所處的階段 (bits 32-62)
- * terminated -- 屏障的結(jié)束標(biāo)記 (bit 63 / sign)
- */
- // 狀態(tài)變量，用于存儲(chǔ)當(dāng)前階段phase、參與者數(shù)parties、未完成的參與者數(shù)unarrived_count
- private volatile long state;
- // 最多可以有多少個(gè)參與者，即每個(gè)階段最多有多少個(gè)任務(wù)
- private static final int MAX_PARTIES = 0xffff;
- // 最多可以有多少階段
- private static final int MAX_PHASE = Integer.MAX_VALUE;
- // 參與者數(shù)量的偏移量
- private static final int PARTIES_SHIFT = 16;
- // 當(dāng)前階段的偏移量
- private static final int PHASE_SHIFT = 32;
- // 未完成的參與者數(shù)的掩碼，低16位
- private static final int UNARRIVED_MASK = 0xffff; // to mask ints
- // 參與者數(shù)，中間16位
- private static final long PARTIES_MASK = 0xffff0000L; // to mask longs
- // counts的掩碼，counts等于參與者數(shù)和未完成的參與者數(shù)的 '｜' 操作
- private static final long COUNTS_MASK = 0xffffffffL;
- private static final long TERMINATION_BIT = 1L << 63;
- // 一次一個(gè)參與者完成
- private static final int ONE_ARRIVAL = 1;
- // 增加減少參與者時(shí)使用
- private static final int ONE_PARTY = 1 << PARTIES_SHIFT;
- // 減少參與者時(shí)使用
- private static final int ONE_DEREGISTER = ONE_ARRIVAL|ONE_PARTY;
- // 沒(méi)有參與者使用
- private static final int EMPTY = 1;
- // 用于求未完成參與者數(shù)量
- private static int unarrivedOf(long s) {
- int counts = (int)s;
- return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
- }
- // 用于求參與者數(shù)量（中間16位），注意int的為止
- private static int partiesOf(long s) {
- return (int)s >>> PARTIES_SHIFT;
- }
- // 用于求階段數(shù)（高32位），注意int的位置
- private static int phaseOf(long s) {
- return (int)(s >>> PHASE_SHIFT);
- }
- // 已完成參與者數(shù)量
- private static int arrivedOf(long s) {
- int counts = (int)s;
- return (counts == EMPTY) ? 0 :
- (counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK);
- }
- /**
- * The parent of this phaser, or null if none
- */
- private final Phaser parent;
- /**
- * The root of phaser tree. Equals this if not in a tree.
- */
- private final Phaser root;
- // 用于存儲(chǔ)已經(jīng)=完成參與者所在的線程，根據(jù)當(dāng)前階段的奇偶性選擇不同的隊(duì)列
- private final AtomicReference
evenQ; - private final AtomicReference
oddQ;
主要屬性位 state 和 evenQ 及 oddQ
- state，volatile的long來(lái)表示狀態(tài)變量，高32位存儲(chǔ)當(dāng)前階段phase，中間16位存儲(chǔ)參與者的數(shù)量，低16位存儲(chǔ)未完成參與者的數(shù)量。
- unarrived -- 還沒(méi)有抵達(dá)屏障的參與者的個(gè)數(shù) (bits 0-15)
- parties -- 需要等待的參與者的個(gè)數(shù) (bits 16-31)
- phase -- 屏障所處的階段 (bits 32-62)
- terminated -- 屏障的結(jié)束標(biāo)記 (bit 63 / sign)
如果是空狀態(tài)，也就是沒(méi)有子階段注冊(cè)的初始階段。這里用一個(gè)EMPTY狀態(tài)表示，也就是0個(gè)子階段和一個(gè)未到達(dá)階段。
所有的狀態(tài)變化都是通過(guò)CAS操作執(zhí)行的，唯一例外是注冊(cè)一個(gè)子相移器(sub-Phaser)，用于構(gòu)成樹(shù)的，也就是Phaser的父Phaser非空。這個(gè)子相移器的分階段是通過(guò)一個(gè)內(nèi)置鎖來(lái)設(shè)置。
- evenQ 和 oddQ，是根據(jù)phaser的奇偶狀態(tài)來(lái)設(shè)置的，用來(lái)存儲(chǔ)等待的線程。為了避免競(jìng)爭(zhēng)，這里使用了Phaser的數(shù)值奇偶來(lái)存儲(chǔ)，此外對(duì)于子相移器，它與其根相移器使用同一個(gè)evenQ或者oddQ，以加速釋放。
構(gòu)造方法
- 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) { // 父phaser不為空
- final Phaser root = parent.root;
- this.root = root; // 指向root phaser
- this.evenQ = root.evenQ; // 兩個(gè)棧，整個(gè)phaser鏈只有一份
- this.oddQ = root.oddQ;
- if (parties != 0)
- phase = parent.doRegister(1); // 向父phaser注冊(cè)當(dāng)前線程
- }
- else {
- this.root = this; // 否則，自己是root phaser
- this.evenQ = new AtomicReference
(); // 負(fù)責(zé)創(chuàng)建兩個(gè)棧（QNode鏈） - this.oddQ = new AtomicReference
(); - }
- // 狀態(tài)變量state的存儲(chǔ)分為三段
- this.state = (parties == 0) ? (long)EMPTY :
- ((long)phase << PHASE_SHIFT) |
- ((long)parties << PARTIES_SHIFT) |
- ((long)parties);
- }
構(gòu)造函數(shù)中還有一個(gè)parent和root，這是用來(lái)構(gòu)造多層級(jí)階段的，用于構(gòu)成樹(shù)的。
重點(diǎn)還是還是看state的賦值方式，高32位存儲(chǔ)當(dāng)前階段phase，中間16位存儲(chǔ)參與者的數(shù)量，低16位存儲(chǔ)未完成參與者的數(shù)量。
主要方法
下面我們一起來(lái)看看幾個(gè)主要方法的源碼，重點(diǎn)是三個(gè)private的核心方法：doArrive、doRegister、reconcileState
register方法
增加一個(gè)參與者，需要同時(shí)增加parties和unarrived兩個(gè)數(shù)值，也就是state中的16位和低16位(中間16位存儲(chǔ)參與者的數(shù)量，低16位存儲(chǔ)未完成參與者的數(shù)量)
- public int register() {
- return doRegister(1);
- }
這里主要調(diào)用的是doRegister方法，我們往下看。
doRegister方法
- private int doRegister(int registrations) {
- // adjustment to state
- // state應(yīng)該加的值，注意這里是相當(dāng)于同時(shí)增加parties和unarrived
- long adjust = ((long)registrations << PARTIES_SHIFT) | registrations; //計(jì)算出需要調(diào)整的量
- final Phaser parent = this.parent; //查看可能存在的相移器
- int phase;
- for (;;) {
- // state的值
- long s = (parent == null) ? state : reconcileState(); // reconcileState()方法是調(diào)整當(dāng)前phaser的狀態(tài)與root的一致
- // state的低32未，也就是parties和unarrived的值
- int counts = (int)s;
- // parties的值
- int parties = counts >>> PARTIES_SHIFT;
- // unarrived的值
- int unarrived = counts & UNARRIVED_MASK;
- // 檢查是否溢出
- if (registrations > MAX_PARTIES - parties) //如果需要注冊(cè)的數(shù)量超過(guò)運(yùn)行注冊(cè)的最大值，則拋出異常狀態(tài)異常
- throw new IllegalStateException(badRegister(s));
- // 當(dāng)前階段phase
- phase = (int)(s >>> PHASE_SHIFT);
- if (phase < 0) //如果當(dāng)前狀態(tài)為終止?fàn)顟B(tài)則跳出循環(huán)直接退出
- break;
- // 不是第一個(gè)參與者
- if (counts != EMPTY) { // not 1st registration //如果當(dāng)前狀態(tài)不是第一次注冊(cè)線程
- if (parent == null || reconcileState() == s) { //如果當(dāng)相移器的父相移器為空，則直接信息CAS，如果當(dāng)前相移器部位空則調(diào)用reconcileState處理，這個(gè)稍后再看。reconcileState這里主要為了防止出現(xiàn)同步性錯(cuò)誤。
- // unarrived等于0說(shuō)明當(dāng)前階段正在執(zhí)行onAdvance()方法，等待其執(zhí)行完畢
- if (unarrived == 0) // wait out advance
- root.internalAwaitAdvance(phase, null);
- // 否則就修改state的值，增加adjust，如果成功就跳出循環(huán)
- else if (UNSAFE.compareAndSwapLong(this, stateOffset,
- s, s + adjust))
- break;
- }
- }
- // 是第一個(gè)參與者，當(dāng)前狀態(tài)是第一次注冊(cè)。如果如果當(dāng)前相移器沒(méi)有父相移器。則直接進(jìn)行CAS
- else if (parent == null) { // 1st root registration
- // 計(jì)算state的值
- long next = ((long)phase << PHASE_SHIFT) | adjust;
- // 修改state的值，如果成功就跳出循環(huán)
- if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
- break;
- }
- else { // 如果當(dāng)前是第一次設(shè)置，并且該相移器被組織在一個(gè)樹(shù)中則需要考慮一下，則需要使用內(nèi)置鎖來(lái)進(jìn)如
- // 多層級(jí)階段的處理方式
- synchronized (this) { // 1st sub registration
- if (state == s) { // recheck under lock 這里有可能發(fā)生競(jìng)爭(zhēng)。所以這里還需要檢查一下，如果失敗則需退出同步區(qū)重新嘗試進(jìn)入。
- phase = parent.doRegister(1); // 調(diào)用其父相移器的注冊(cè)方法
- if (phase < 0)
- break;
- // finish registration whenever parent registration
- // succeeded, even when racing with termination,
- // since these are part of the same "transaction".
- while (!UNSAFE.compareAndSwapLong
- (this, stateOffset, s,
- ((long)phase << PHASE_SHIFT) | adjust)) {
- s = state;
- phase = (int)(root.state >>> PHASE_SHIFT);
- // assert (int)s == EMPTY;
- }
- break;
- }
- }
- }
- }
- return phase;
- }
增加一個(gè)參與者的總體的邏輯為：
- 增加一個(gè)參與者，需要同時(shí)增加parties和unarrived兩個(gè)數(shù)值，也就是state中的16位和低16位;
- 如果是第一個(gè)參與者，則嘗試原子更新state的值，如果成功了就退出;
- 如果不是第一個(gè)參與者，則檢查是不是在執(zhí)行onAdvance() , 如果是等待onAdvance() 執(zhí)行完成，如果否則嘗試原子更新state的值，直到成功退出;
- 等待onAdvance() 完成是采用先自旋后進(jìn)入隊(duì)列排隊(duì)的方式等待，減少線程上下文切換;
arriveAndAwaitAdvance()方法
當(dāng)前線程當(dāng)前階段執(zhí)行完畢，等待其他線程完成當(dāng)前階段。 如果當(dāng)前線程是該階段最后一個(gè)到達(dá)的，則當(dāng)前線程會(huì)執(zhí)行onAdvance()方法，并喚醒其它線程進(jìn)入下一個(gè)階段。
- public int arriveAndAwaitAdvance() {
- // Specialization of doArrive+awaitAdvance eliminating some reads/paths
- final Phaser root = this.root;
- for (;;) {
- // state的值
- long s = (root == this) ? state : reconcileState();
- // 當(dāng)前階段
- int phase = (int)(s >>> PHASE_SHIFT);
- if (phase < 0)
- return phase;
- // parties 和 unarrived的值
- int counts = (int)s;
- // unarrived的值（state的低16位）
- int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
- if (unarrived <= 0)
- throw new IllegalStateException(badArrive(s));
- // 修改state的值
- if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
- s -= ONE_ARRIVAL)) {
- // 如果不是最后一個(gè)到達(dá)的，則調(diào)用internalAwaitAdvance()方法自旋或進(jìn)入隊(duì)列等待
- if (unarrived > 1)
- // 這里是直接返回了，internalAwaitAdvance()方法的源碼見(jiàn)register()方法解析
- return root.internalAwaitAdvance(phase, null);
- // 到這里說(shuō)明是最后一個(gè)到達(dá)的參與者
- if (root != this)
- return parent.arriveAndAwaitAdvance();
- // n 只保留了state中parties的部分，也就是中16位
- long n = s & PARTIES_MASK; // base of next state
- // parties的值，即下一次需要到達(dá)的參與者數(shù)量
- int nextUnarrived = (int)n >>> PARTIES_SHIFT;
- // 執(zhí)行onAdvance()方法，返回true表示下一階段參與者數(shù)量為0了，也就是結(jié)束了
- if (onAdvance(phase, nextUnarrived))
- n |= TERMINATION_BIT;
- else if (nextUnarrived == 0)
- n |= EMPTY;
- else
- n |= nextUnarrived; // n加上unarrived的值
- // 下階段等待當(dāng)前階段加1
- int nextPhase = (phase + 1) & MAX_PHASE;
- // n 加上下一個(gè)階段的值
- n |= (long)nextPhase << PHASE_SHIFT;
- // 修改state的值為n
- if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))
- return (int)(state >>> PHASE_SHIFT); // terminated
- // 喚醒其它參與者并進(jìn)入下一個(gè)階段
- releaseWaiters(phase);
- // 返回下一階段的值
- return nextPhase;
- }
- }
- }
arriveAndAwaitAdvance的大致邏輯為：
- 修改state中unarrived部分的值減1;
- 如果不是最后一個(gè)到達(dá)，則調(diào)用internalAwaitAdvance() 方法自旋或排隊(duì)等待;
- 如果是最后一個(gè)到達(dá)的，則調(diào)用onAdvance() 方法，然后修改state的值為下一階段對(duì)應(yīng)的值，并喚醒其它等待的線程;
- 返回下一階段俄值。
internalAwaitAdvance方法
internalAwaitAdvance方法。實(shí)際上Phaser中阻塞都是通過(guò)這個(gè)語(yǔ)句實(shí)現(xiàn)的。這個(gè)語(yǔ)句必須通過(guò)根相移器調(diào)用。換句話說(shuō)所有的阻塞都是在根相移器阻塞的。
輸入?yún)?shù)中phase是需要阻塞的階段。node是用來(lái)跟蹤可能中斷的阻塞節(jié)點(diǎn)。
- // 等待onAdvance()方法執(zhí)行完畢
- // 原理是先自旋一定次數(shù)，如果進(jìn)入下一個(gè)階段，這個(gè)方法直接返回了，
- // 如果自旋一定次數(shù)還沒(méi)有進(jìn)入下一個(gè)階段，則當(dāng)前線程入隊(duì)列，等待onAdvance()執(zhí)行完成喚醒
- private int internalAwaitAdvance(int phase, QNode node) {
- // assert root == this;
- // 保證隊(duì)列為空
- releaseWaiters(phase-1); // ensure old queue clean
- boolean queued = false; // true when node is enqueued
- int lastUnarrived = 0; // to increase spins upon change
- // 自旋的次數(shù)
- int spins = SPINS_PER_ARRIVAL;
- long s;
- int p;
- // 檢查當(dāng)前階段是否變化，如果變化了說(shuō)明進(jìn)入下一個(gè)階段了，這時(shí)候就沒(méi)有必要自旋了
- while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
- // 如果node為空，注冊(cè)的時(shí)候傳入的為空
- if (node == null) { // spinning in noninterruptible mode
- // 未完成的參與者數(shù)量
- int unarrived = (int)s & UNARRIVED_MASK;
- // unarrived 有變化，增加自旋次數(shù)
- if (unarrived != lastUnarrived &&
- (lastUnarrived = unarrived) < NCPU)
- spins += SPINS_PER_ARRIVAL;
- boolean interrupted = Thread.interrupted();
- // 自旋次數(shù)萬(wàn)了，則新建一個(gè)節(jié)點(diǎn)
- if (interrupted || --spins < 0) { // need node to record intr
- node = new QNode(this, phase, false, false, 0L);
- node.wasInterrupted = interrupted;
文章題目：并發(fā)編程之Phaser原理與應(yīng)用
網(wǎng)頁(yè)URL：http://m.fisionsoft.com.cn/article/dpdeidh.html
