java并發(fā)包工具CountDownLatch源碼分析
一：簡述
本篇文章對java并發(fā)包工具CountDownLatch進行介紹，并且通過對CountDownLatch源碼的分析來加深對CountDownLatch的理解。
二：什么是CountDownLatch
CountDownLatch是java并發(fā)包中提供的一個工具類，CountDownLatch的作用很簡單，它可以讓一個或者一組線程在開始執(zhí)行操作之前,必須要等到其他線程執(zhí)行完才執(zhí)行，它是基于AQS的共享鎖來實現(xiàn)的。
三：CountDownLatch的使用
簡單介紹下CountDownLatch的使用
CountDownLatch的主要方法有三個：
1.構造函數(shù)
2.countDown()
3.await()
簡單給大家寫一個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("計數(shù)器減1");
countDownLatch.countDown();
},"線程A").start();
new Thread(()->{
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("計數(shù)器減1");
countDownLatch.countDown();
},"線程B").start();
countDownLatch.await();
System.out.println("主線程被喚醒");
}
}
線程C和主線程調(diào)用await()方法后會進行阻塞，直到線程A和線程B調(diào)用countdown()方法將計數(shù)值減為0之后才會繼續(xù)執(zhí)行。
輸出結果：
四：CountDownLatch原理分析
前面兩個小節(jié)是為了幫助不知道沒使用過CountDownLatch的同學。那么接下來進入正題，對CountDownLatch的原理分析。我們將以CountDownLatch的構造函數(shù)，countDown()，await()三個方法對CountDownLatch的源碼進行解析。
構造函數(shù)
CountDownLatch只有一個有參的構造函數(shù)，我們需要傳遞一個大于0的整數(shù)，構造函數(shù)會初始化一個Sync的實例，而Sync正是繼承了AbstractQueuedSynchronizer（簡稱AQS）。Sync初始化的時候會將我們設置的整數(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;
}
接下來我們看await()方法
await()方法：
流程圖：
源碼分析：
首先線程調(diào)用await()方法后會去判斷當前state是否大于0，如果不是大于0，那么直接就返回繼續(xù)執(zhí)行業(yè)務代碼，如果大于0，那么就會調(diào)用doAcquireSharedInterruptibly()。所以重點是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();
//判斷當前的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()方法，所以接下來我們重點分析doAcquireSharedInterruptibly()方法。
doAcquireSharedInterruptibly()
首先通過addWaiter()方法將當前線程封裝成一個類型為SHARED的Node節(jié)點，然后判斷當前節(jié)點的前一個節(jié)點是否是head節(jié)點，分為兩種情況：
1. 當前節(jié)點的前置節(jié)點是head節(jié)點
那么就會再次調(diào)用tryAcquireShared()判斷一下state的值是等于0，又分為兩種情況
a. state如果等于0
那么就調(diào)用setHeadAndPropagate()方法將當前節(jié)點設置為頭節(jié)點，并且調(diào)用喚醒下一個狀態(tài)不為CANCELLED的節(jié)點。
b. 如果state不等于0
那么就調(diào)用shouldParkAfterFailedAcquire()方法將前一個節(jié)點的狀態(tài)修改為SIGNAL，并且調(diào)用parkAndCheckInterrupt()方法將當前線程阻塞起來。
2. 當前節(jié)點的前置節(jié)點不是head節(jié)點
那么就掉用shouldParkAfterFailedAcquire()方法將前一個節(jié)點的狀態(tài)修改為SIGNAL，并且調(diào)用parkAndCheckInterrupt()方法將當前線程阻塞起來。
(需要注意的是線程被喚醒之后繼續(xù)執(zhí)行這里的代碼)
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//調(diào)用addWaiter()方法將線程封裝成Node并且放入到AQS隊列的尾部
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
//獲取當前節(jié)點的前一個節(jié)點
final Node p = node.predecessor();
//如果前一個節(jié)點是head節(jié)點
if (p == head) {
//再次判斷state的值是否為0
int r = tryAcquireShared(arg);
// tryAcquireShared()返回1代表state為0
if (r >= 0) {
//將當前節(jié)點設置為頭節(jié)點 并且喚醒下一個正常的節(jié)點
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//shouldParkAfterFailedAcquire()方法將當前節(jié)點的前一個節(jié)點的狀態(tài)設置為SIGNAL，
//parkAndCheckInterrupt()方法將當前線程阻塞
//線程被喚醒之后繼續(xù)從這里開始執(zhí)行
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
addWaiter()
addWaiter()作用是將當前線程封裝成Node節(jié)點，并且加入到AQS隊列中。
private Node addWaiter(Node mode) {
//將沒有獲得鎖的線程封裝成一個node
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
//如果AQS尾結點不為null 代表AQS鏈表已經(jīng)初始化 嘗試將構建好的節(jié)點添加到鏈表的尾部
if (pred != null) {
node.prev = pred;
//cas替換AQS的尾結點
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//沒有初始化調(diào)用enq()方法
enq(node);
return node;
}
private Node enq(final Node node) {
//自旋
for (;;) {
Node t = tail;
//尾結點為空 說明AQS鏈表還沒有初始化 那么進行初始化
if (t == null) { // Must initialize
//cas 將AQS的head節(jié)點 初始化 成功初始化head之后，將尾結點也初始化
//注意 這里我們可以看到head節(jié)點是不存儲線程信息的 也就是說head節(jié)點相當于是一個虛擬節(jié)點
if (compareAndSetHead(new Node()))
tail = head;
} else {
//尾結點不為空 那么直接添加到鏈表的尾部即可
//加入鏈表的時候先指定prev 然后cas成功 再指定next
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
setHeadAndPropagate()
setHeadAndPropagate()的作用就是將當前節(jié)點設置為頭結點，并且調(diào)用doReleaseShared()方法喚醒當前節(jié)點的下一個正常節(jié)點。doReleaseShared()方法我們在下面分析countDown()方法的時候在進行仔細的分析。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
//將當前節(jié)點設置為頭結點
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é)點
//其實也就是當前節(jié)點的下一個節(jié)點，因為前面已經(jīng)將當前節(jié)點設置為新的頭結點了
doReleaseShared();
}
}
shouldParkAfterFailedAcquire()
shouldParkAfterFailedAcquire()方法會將傳入的節(jié)點（傳進來的是當前節(jié)點的前置節(jié)點）的狀態(tài)設置為SIGNAL狀態(tài)。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//如果節(jié)點是SIGNAL狀態(tài) 不需要處理 直接返回
return true;
if (ws > 0) {
//如果節(jié)點狀態(tài)>0 說明節(jié)點是取消狀態(tài) 這種狀態(tài)的節(jié)點需要被清除 用do while循環(huán)順便清除一下前面的連續(xù)的、狀態(tài)為取消的節(jié)點
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//正常的情況下 利用cas將前一個節(jié)點的狀態(tài)替換為 SIGNAL狀態(tài) 也就是-1
//注意 這樣隊列中節(jié)點的狀態(tài) 除了最后一個都是-1 包括head節(jié)點
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
parkAndCheckInterrupt()
parkAndCheckInterrupt()方法的作用就是調(diào)用 LockSupport.park()方法將線程阻塞，并且返回線程的中斷標志。
private final boolean parkAndCheckInterrupt() {
//掛起當前線程 并且返回中斷標志 LockSupport.park(thread) 會調(diào)用UNSAFE.park()方法將線程阻塞起來（是一個native方法）
LockSupport.park(this);
return Thread.interrupted();
}
到這里await()方法也就分析完了 接下來我們分析countDown()方法
countDown()
流程圖：
源碼：
countDown()方法首先查看state的值是否是0，分為兩種情況
1. 如果state為0
說明沒有線程需要被喚醒，那么直接返回。
2. 如果state不為0
那么將利用cas將state的值減1，判斷新的state是否為0 ，如果不為0，說明還不能喚醒阻塞的線程，直接返回，如果新的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 說明沒有需要喚醒的線程 直接返回
if (c == 0)
return false;
int nextc = c-1;
//利用cas將state減一 如果新的state為0 說明需要喚醒阻塞的線程，否則不需要喚醒
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
countDown()方法核心是doReleaseShared()方法 所以我們重點分析doReleaseShared()。
doReleaseShared()
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
//如果頭結點的狀態(tài)是SIGNAL
if (ws == Node.SIGNAL) {
//cas修改節(jié)點的狀態(tài)為0 失敗的話繼續(xù)自旋
// 成功的話調(diào)用unparkSuccessor喚醒頭結點的下一個正常節(jié)點
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
//如果節(jié)點狀態(tài)為0 那么cas替換為PROPAGATE 失敗進入下一次自旋
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é)點后第一個不為null且狀態(tài)不為cancelled的節(jié)點。
private void unparkSuccessor(Node node) {
//獲取頭結點的狀態(tài) 將頭結點狀態(tài)設置為0 代表現(xiàn)在正在有線程被喚醒 如果head狀態(tài)為0 就不會進入這個方法了
int ws = node.waitStatus;
if (ws < 0)
//將頭結點狀態(tài)設置為0
compareAndSetWaitStatus(node, ws, 0);
//喚醒頭結點的下一個狀態(tài)不是cancelled的節(jié)點 （因為頭結點是不存儲阻塞線程的）
Node s = node.next;
//當前節(jié)點是null 或者是cancelled狀態(tài)
if (s == null || s.waitStatus > 0) {
s = null;
//從aqs鏈表的尾部開始遍歷 找到離頭結點最近的 不為空的 狀態(tài)不是cancelled的節(jié)點 賦值給s
//這里為什么從尾結點開始遍歷而不是頭結點 是因為添加結點的時候是先初始化結點的prev的， 從尾結點開始遍歷 不會出現(xiàn)prve沒有賦值的情況
//如果從頭結點進行遍歷 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()方法中。如果線程沒有被中斷過，會判斷state的值，這里線程是被調(diào)用countDown方法喚醒的，所以state一定是0，所以會調(diào)用setHeadAndPropagate()方法更新頭結點并繼續(xù)喚醒之后的線程。這樣就會把依次將所有阻塞的阻塞線程都喚醒。（因為countDownLatch的計數(shù)器為0之后需要將所有調(diào)用await()阻塞的線程喚醒）
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
//
for (;;) {
//獲取當前節(jié)點的前一個節(jié)點
final Node p = node.predecessor();
//如果前一個節(jié)點是head節(jié)點
if (p == head) {
//再次判斷state的值是否為0
int r = tryAcquireShared(arg);
// tryAcquireShared()返回1代表state為0
if (r >= 0) {
//將當前節(jié)點設置為頭節(jié)點 并且喚醒下一個正常的節(jié)點
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//線程被喚醒之后繼續(xù)從這里開始執(zhí)行 如果線程沒有被中斷過 會進入都下次for循環(huán)
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
五：最后
本篇文章主要介紹了CountDownLatch的使用并且通過分析其源碼對CountDownLatch的原理進行了分析。
注：其實像addWaiter()，unparkSuccessor(),shouldParkAfterFailedAcquire()等一些AQS公用的方法在我的另外一篇文章里分析過，原文地址：ReentrantLock源碼分析
更多關于java并發(fā)包CountDownLatch的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要介紹了mybatis-plus:xml拼接sql方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-08-08
Apache Commons Math3探索之多項式曲線擬合實現(xiàn)代碼
這篇文章主要介紹了Apache Commons Math3探索之多項式曲線擬合實現(xiàn)代碼，小編覺得挺不錯的，這里分享給大家，供需要的朋友參考。2017-10-10
SpringBoot從0到1整合銀聯(lián)無跳轉支付功能附源碼
這篇文章主要介紹了SpringBoot從0到1整合銀聯(lián)無跳轉功能支付附源碼,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2020-11-11
這篇文章主要為大家詳細介紹了java實現(xiàn)上傳文件到服務器和客戶端，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-01-01
數(shù)組(Array)是有序的元素序列。 若將有限個類型相同的變量的集合命名，那么這個名稱為數(shù)組名。組成數(shù)組的各個變量稱為數(shù)組的分量，也稱為數(shù)組的元素，有時也稱為下標變量2021-09-09
這篇文章主要給大家介紹了關于Mybatis批量插入更新xml方式和注解方式的相關資料，文中通過示例代碼介紹的非常詳細，對大家學習或者使用Mybatis具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧2019-12-12
Idea安裝Eslint插件提示：Plugin NativeScript was not installed的問題
這篇文章主要介紹了Idea安裝Eslint插件提示：Plugin NativeScript was not installed的問題,本文通過圖文并茂的形式給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2020-10-10
今天給大家總結一下線程池的拒絕策略,文中有非常詳細的介紹及代碼示例,對正在學習java的小伙伴們有很好地幫助,需要的朋友可以參考下2021-05-05
這篇文章主要介紹了JAVA內(nèi)存分配與參數(shù)傳遞，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-03-03
如何使用Spring AOP預處理Controller的參數(shù)
這篇文章主要介紹了如何使用Spring AOP預處理Controller的參數(shù)操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-08-08