java并發(fā)包工具CountDownLatch源碼分析
一：簡(jiǎn)述
本篇文章對(duì)java并發(fā)包工具CountDownLatch進(jìn)行介紹，并且通過(guò)對(duì)CountDownLatch源碼的分析來(lái)加深對(duì)CountDownLatch的理解。
二：什么是CountDownLatch
CountDownLatch是java并發(fā)包中提供的一個(gè)工具類(lèi)，CountDownLatch的作用很簡(jiǎn)單，它可以讓一個(gè)或者一組線程在開(kāi)始執(zhí)行操作之前,必須要等到其他線程執(zhí)行完才執(zhí)行，它是基于AQS的共享鎖來(lái)實(shí)現(xiàn)的。
三：CountDownLatch的使用
簡(jiǎn)單介紹下CountDownLatch的使用
CountDownLatch的主要方法有三個(gè)：
1.構(gòu)造函數(shù)
2.countDown()
3.await()
簡(jiǎn)單給大家寫(xiě)一個(gè)demo：
public class TestThread {
static CountDownLatch countDownLatch = new CountDownLatch(2);
public static void main(String[] args) throws InterruptedException {
new Thread(()->{
System.out.println("線程等待");
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("線程C被喚醒");
},"線程C").start();
new Thread(()->{
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("計(jì)數(shù)器減1");
countDownLatch.countDown();
},"線程A").start();
new Thread(()->{
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("計(jì)數(shù)器減1");
countDownLatch.countDown();
},"線程B").start();
countDownLatch.await();
System.out.println("主線程被喚醒");
}
}
線程C和主線程調(diào)用await()方法后會(huì)進(jìn)行阻塞，直到線程A和線程B調(diào)用countdown()方法將計(jì)數(shù)值減為0之后才會(huì)繼續(xù)執(zhí)行。
輸出結(jié)果：
四：CountDownLatch原理分析
前面兩個(gè)小節(jié)是為了幫助不知道沒(méi)使用過(guò)CountDownLatch的同學(xué)。那么接下來(lái)進(jìn)入正題，對(duì)CountDownLatch的原理分析。我們將以CountDownLatch的構(gòu)造函數(shù)，countDown()，await()三個(gè)方法對(duì)CountDownLatch的源碼進(jìn)行解析。
構(gòu)造函數(shù)
CountDownLatch只有一個(gè)有參的構(gòu)造函數(shù)，我們需要傳遞一個(gè)大于0的整數(shù)，構(gòu)造函數(shù)會(huì)初始化一個(gè)Sync的實(shí)例，而Sync正是繼承了AbstractQueuedSynchronizer（簡(jiǎn)稱(chēng)AQS）。Sync初始化的時(shí)候會(huì)將我們?cè)O(shè)置的整數(shù)傳遞給AQS的成員變量state。
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
}
protected final void setState(int newState) {
state = newState;
}
接下來(lái)我們看await()方法
await()方法：
流程圖：
源碼分析：
首先線程調(diào)用await()方法后會(huì)去判斷當(dāng)前state是否大于0，如果不是大于0，那么直接就返回繼續(xù)執(zhí)行業(yè)務(wù)代碼，如果大于0，那么就會(huì)調(diào)用doAcquireSharedInterruptibly()。所以重點(diǎn)是doAcquireSharedInterruptibly()方法。
public void await() throws InterruptedException {
//調(diào)用Sync的acquireSharedInterruptibly()方法
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//判斷當(dāng)前的state的值是否等于0 如果等于0返回1 否則返回-1
if (tryAcquireShared(arg) < 0)
// 如果state等于0 那么什么都不做 直接返回，如果大于0 就執(zhí)行doAcquireSharedInterruptibly()
doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
await()方法的核心在于doAcquireSharedInterruptibly()方法，所以接下來(lái)我們重點(diǎn)分析doAcquireSharedInterruptibly()方法。
doAcquireSharedInterruptibly()
首先通過(guò)addWaiter()方法將當(dāng)前線程封裝成一個(gè)類(lèi)型為SHARED的Node節(jié)點(diǎn)，然后判斷當(dāng)前節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)是否是head節(jié)點(diǎn)，分為兩種情況：
1. 當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)是head節(jié)點(diǎn)
那么就會(huì)再次調(diào)用tryAcquireShared()判斷一下state的值是等于0，又分為兩種情況
a. state如果等于0
那么就調(diào)用setHeadAndPropagate()方法將當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn)，并且調(diào)用喚醒下一個(gè)狀態(tài)不為CANCELLED的節(jié)點(diǎn)。
b. 如果state不等于0
那么就調(diào)用shouldParkAfterFailedAcquire()方法將前一個(gè)節(jié)點(diǎn)的狀態(tài)修改為SIGNAL，并且調(diào)用parkAndCheckInterrupt()方法將當(dāng)前線程阻塞起來(lái)。
2. 當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)不是head節(jié)點(diǎn)
那么就掉用shouldParkAfterFailedAcquire()方法將前一個(gè)節(jié)點(diǎn)的狀態(tài)修改為SIGNAL，并且調(diào)用parkAndCheckInterrupt()方法將當(dāng)前線程阻塞起來(lái)。
(需要注意的是線程被喚醒之后繼續(xù)執(zhí)行這里的代碼)
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//調(diào)用addWaiter()方法將線程封裝成Node并且放入到AQS隊(duì)列的尾部
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
//獲取當(dāng)前節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)
final Node p = node.predecessor();
//如果前一個(gè)節(jié)點(diǎn)是head節(jié)點(diǎn)
if (p == head) {
//再次判斷state的值是否為0
int r = tryAcquireShared(arg);
// tryAcquireShared()返回1代表state為0
if (r >= 0) {
//將當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn) 并且喚醒下一個(gè)正常的節(jié)點(diǎn)
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//shouldParkAfterFailedAcquire()方法將當(dāng)前節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)的狀態(tài)設(shè)置為SIGNAL，
//parkAndCheckInterrupt()方法將當(dāng)前線程阻塞
//線程被喚醒之后繼續(xù)從這里開(kāi)始執(zhí)行
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
addWaiter()
addWaiter()作用是將當(dāng)前線程封裝成Node節(jié)點(diǎn)，并且加入到AQS隊(duì)列中。
private Node addWaiter(Node mode) {
//將沒(méi)有獲得鎖的線程封裝成一個(gè)node
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
//如果AQS尾結(jié)點(diǎn)不為null 代表AQS鏈表已經(jīng)初始化 嘗試將構(gòu)建好的節(jié)點(diǎn)添加到鏈表的尾部
if (pred != null) {
node.prev = pred;
//cas替換AQS的尾結(jié)點(diǎn)
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//沒(méi)有初始化調(diào)用enq()方法
enq(node);
return node;
}
private Node enq(final Node node) {
//自旋
for (;;) {
Node t = tail;
//尾結(jié)點(diǎn)為空 說(shuō)明AQS鏈表還沒(méi)有初始化 那么進(jìn)行初始化
if (t == null) { // Must initialize
//cas 將AQS的head節(jié)點(diǎn) 初始化 成功初始化head之后，將尾結(jié)點(diǎn)也初始化
//注意 這里我們可以看到head節(jié)點(diǎn)是不存儲(chǔ)線程信息的 也就是說(shuō)head節(jié)點(diǎn)相當(dāng)于是一個(gè)虛擬節(jié)點(diǎn)
if (compareAndSetHead(new Node()))
tail = head;
} else {
//尾結(jié)點(diǎn)不為空 那么直接添加到鏈表的尾部即可
//加入鏈表的時(shí)候先指定prev 然后cas成功 再指定next
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
setHeadAndPropagate()
setHeadAndPropagate()的作用就是將當(dāng)前節(jié)點(diǎn)設(shè)置為頭結(jié)點(diǎn)，并且調(diào)用doReleaseShared()方法喚醒當(dāng)前節(jié)點(diǎn)的下一個(gè)正常節(jié)點(diǎn)。doReleaseShared()方法我們?cè)谙旅娣治鯿ountDown()方法的時(shí)候在進(jìn)行仔細(xì)的分析。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
//將當(dāng)前節(jié)點(diǎn)設(shè)置為頭結(jié)點(diǎn)
setHead(node);
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
//喚醒頭結(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)
//其實(shí)也就是當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)，因?yàn)榍懊嬉呀?jīng)將當(dāng)前節(jié)點(diǎn)設(shè)置為新的頭結(jié)點(diǎn)了
doReleaseShared();
}
}
shouldParkAfterFailedAcquire()
shouldParkAfterFailedAcquire()方法會(huì)將傳入的節(jié)點(diǎn)（傳進(jìn)來(lái)的是當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)）的狀態(tài)設(shè)置為SIGNAL狀態(tài)。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//如果節(jié)點(diǎn)是SIGNAL狀態(tài) 不需要處理 直接返回
return true;
if (ws > 0) {
//如果節(jié)點(diǎn)狀態(tài)>0 說(shuō)明節(jié)點(diǎn)是取消狀態(tài) 這種狀態(tài)的節(jié)點(diǎn)需要被清除 用do while循環(huán)順便清除一下前面的連續(xù)的、狀態(tài)為取消的節(jié)點(diǎn)
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//正常的情況下 利用cas將前一個(gè)節(jié)點(diǎn)的狀態(tài)替換為 SIGNAL狀態(tài) 也就是-1
//注意 這樣隊(duì)列中節(jié)點(diǎn)的狀態(tài) 除了最后一個(gè)都是-1 包括head節(jié)點(diǎn)
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
parkAndCheckInterrupt()
parkAndCheckInterrupt()方法的作用就是調(diào)用 LockSupport.park()方法將線程阻塞，并且返回線程的中斷標(biāo)志。
private final boolean parkAndCheckInterrupt() {
//掛起當(dāng)前線程 并且返回中斷標(biāo)志 LockSupport.park(thread) 會(huì)調(diào)用UNSAFE.park()方法將線程阻塞起來(lái)（是一個(gè)native方法）
LockSupport.park(this);
return Thread.interrupted();
}
到這里await()方法也就分析完了 接下來(lái)我們分析countDown()方法
countDown()
流程圖：
源碼：
countDown()方法首先查看state的值是否是0，分為兩種情況
1. 如果state為0
說(shuō)明沒(méi)有線程需要被喚醒，那么直接返回。
2. 如果state不為0
那么將利用cas將state的值減1，判斷新的state是否為0 ，如果不為0，說(shuō)明還不能喚醒阻塞的線程，直接返回，如果新的state為0，那么調(diào)用doReleaseShared()方法喚醒阻塞的線程。
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
//自旋+cas保證線程安全
for (;;) {
//獲取state的值
int c = getState();
//如果state為0 說(shuō)明沒(méi)有需要喚醒的線程 直接返回
if (c == 0)
return false;
int nextc = c-1;
//利用cas將state減一 如果新的state為0 說(shuō)明需要喚醒阻塞的線程，否則不需要喚醒
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
countDown()方法核心是doReleaseShared()方法 所以我們重點(diǎn)分析doReleaseShared()。
doReleaseShared()
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
//如果頭結(jié)點(diǎn)的狀態(tài)是SIGNAL
if (ws == Node.SIGNAL) {
//cas修改節(jié)點(diǎn)的狀態(tài)為0 失敗的話繼續(xù)自旋
// 成功的話調(diào)用unparkSuccessor喚醒頭結(jié)點(diǎn)的下一個(gè)正常節(jié)點(diǎn)
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
//如果節(jié)點(diǎn)狀態(tài)為0 那么cas替換為PROPAGATE 失敗進(jìn)入下一次自旋
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
unparkSuccessor()
unparkSuccessor()方法的作用是喚醒頭節(jié)點(diǎn)后第一個(gè)不為null且狀態(tài)不為cancelled的節(jié)點(diǎn)。
private void unparkSuccessor(Node node) {
//獲取頭結(jié)點(diǎn)的狀態(tài) 將頭結(jié)點(diǎn)狀態(tài)設(shè)置為0 代表現(xiàn)在正在有線程被喚醒 如果head狀態(tài)為0 就不會(huì)進(jìn)入這個(gè)方法了
int ws = node.waitStatus;
if (ws < 0)
//將頭結(jié)點(diǎn)狀態(tài)設(shè)置為0
compareAndSetWaitStatus(node, ws, 0);
//喚醒頭結(jié)點(diǎn)的下一個(gè)狀態(tài)不是cancelled的節(jié)點(diǎn) （因?yàn)轭^結(jié)點(diǎn)是不存儲(chǔ)阻塞線程的）
Node s = node.next;
//當(dāng)前節(jié)點(diǎn)是null 或者是cancelled狀態(tài)
if (s == null || s.waitStatus > 0) {
s = null;
//從aqs鏈表的尾部開(kāi)始遍歷 找到離頭結(jié)點(diǎn)最近的 不為空的 狀態(tài)不是cancelled的節(jié)點(diǎn) 賦值給s
//這里為什么從尾結(jié)點(diǎn)開(kāi)始遍歷而不是頭結(jié)點(diǎn) 是因?yàn)樘砑咏Y(jié)點(diǎn)的時(shí)候是先初始化結(jié)點(diǎn)的prev的， 從尾結(jié)點(diǎn)開(kāi)始遍歷 不會(huì)出現(xiàn)prve沒(méi)有賦值的情況
//如果從頭結(jié)點(diǎn)進(jìn)行遍歷 next為null 并不能保證鏈表遍歷完了
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
//調(diào)用LockSupport.unpark()喚醒指定的線程
LockSupport.unpark(s.thread);
}
線程被喚醒之后，我們需要回到線程阻塞的地方繼續(xù)分析線程被喚醒之后的操作。
前文我們分析await()方法之后已經(jīng)知道了線程阻塞在doAcquireSharedInterruptibly()方法中。如果線程沒(méi)有被中斷過(guò)，會(huì)判斷state的值，這里線程是被調(diào)用countDown方法喚醒的，所以state一定是0，所以會(huì)調(diào)用setHeadAndPropagate()方法更新頭結(jié)點(diǎn)并繼續(xù)喚醒之后的線程。這樣就會(huì)把依次將所有阻塞的阻塞線程都喚醒。（因?yàn)閏ountDownLatch的計(jì)數(shù)器為0之后需要將所有調(diào)用await()阻塞的線程喚醒）
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
//
for (;;) {
//獲取當(dāng)前節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)
final Node p = node.predecessor();
//如果前一個(gè)節(jié)點(diǎn)是head節(jié)點(diǎn)
if (p == head) {
//再次判斷state的值是否為0
int r = tryAcquireShared(arg);
// tryAcquireShared()返回1代表state為0
if (r >= 0) {
//將當(dāng)前節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn) 并且喚醒下一個(gè)正常的節(jié)點(diǎn)
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//線程被喚醒之后繼續(xù)從這里開(kāi)始執(zhí)行 如果線程沒(méi)有被中斷過(guò) 會(huì)進(jìn)入都下次for循環(huán)
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
五：最后
本篇文章主要介紹了CountDownLatch的使用并且通過(guò)分析其源碼對(duì)CountDownLatch的原理進(jìn)行了分析。
注：其實(shí)像addWaiter()，unparkSuccessor(),shouldParkAfterFailedAcquire()等一些AQS公用的方法在我的另外一篇文章里分析過(guò)，原文地址：ReentrantLock源碼分析
更多關(guān)于java并發(fā)包CountDownLatch的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
Java中重載與重寫(xiě)的對(duì)比與區(qū)別
這篇文章主要介紹了Java中重載與重寫(xiě)的對(duì)比與區(qū)別的相關(guān)資料,需要的朋友可以參考下2017-03-03
這篇文章主要介紹了Java異步非阻塞編程的幾種方式總結(jié)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-06-06
java動(dòng)態(tài)構(gòu)建數(shù)據(jù)庫(kù)復(fù)雜查詢教程
這篇文章主要介紹了java動(dòng)態(tài)構(gòu)建數(shù)據(jù)庫(kù)復(fù)雜查詢的實(shí)現(xiàn)示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2021-11-11
spring框架下@value注解屬性static無(wú)法獲取值問(wèn)題
這篇文章主要介紹了spring框架下@value注解屬性static無(wú)法獲取值問(wèn)題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-11-11
如何解決Spring事務(wù)注解@Transactional在類(lèi)內(nèi)部方法調(diào)用不生效
這篇文章主要介紹了如何解決Spring事務(wù)注解@Transactional在類(lèi)內(nèi)部方法調(diào)用不生效問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-08-08
這篇文章主要介紹了java的串口通信問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-05-05
Spring MVC之WebApplicationContext_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理
這篇文章主要介紹了Spring MVC之WebApplicationContext的相關(guān)資料,需要的朋友可以參考下2017-08-08
java環(huán)境配好后jar文件打開(kāi)命令框閃退(無(wú)打開(kāi)方式,無(wú)反應(yīng))解決辦法
在Java開(kāi)發(fā)中我們經(jīng)常會(huì)遇到運(yùn)行Jar包時(shí)閃退的問(wèn)題,下面這篇文章主要給大家介紹了關(guān)于java環(huán)境配好后jar文件打開(kāi)命令框閃退(無(wú)打開(kāi)方式,無(wú)反應(yīng))的解決辦法,文中通過(guò)圖文介紹的非常詳細(xì),需要的朋友可以參考下2024-04-04
Spring Boot 整合單機(jī)websocket的步驟 附github源碼
websocket 是一個(gè)通信協(xié)議，通過(guò)單個(gè) TCP 連接提供全雙工通信，這篇文章主要介紹了Spring Boot 整合單機(jī)websocket的步驟(附github源碼),需要的朋友可以參考下2021-10-10
java分頁(yè)之假分頁(yè)實(shí)現(xiàn)簡(jiǎn)單的分頁(yè)器
這篇文章主要介紹了java分頁(yè)之假分頁(yè)實(shí)現(xiàn)簡(jiǎn)單的分頁(yè)器的相關(guān)資料,需要的朋友可以參考下2016-04-04