countdownlatch 是一個(gè)同步工具類褂策，它允許一個(gè)或多個(gè)線 程一直等待忧便，直到其他線程的操作執(zhí)行完畢再執(zhí)行徒役。從命 名可以解讀到 countdown 是倒數(shù)的意思寞蚌，類似于我們倒計(jì)時(shí)的概念。
countdownlatch 提供了兩個(gè)方法麸折，一個(gè)是 countDown锡凝， 一個(gè)是 await， countdownlatch 初始化的時(shí)候需要傳入一 個(gè)整數(shù)垢啼，在這個(gè)整數(shù)倒數(shù)到 0 之前窜锯，調(diào)用了 await 方法的程序都必須要等待，然后通過 countDown 來倒數(shù)芭析。
案例
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch=new CountDownLatch(3);
System.out.println("main start");
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"->begin");
countDownLatch.countDown(); //初始值-1 =3-1=2;
System.out.println(Thread.currentThread().getName()+"->end");
},"t1").start();
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"->begin");
countDownLatch.countDown(); //2-1=1;
System.out.println(Thread.currentThread().getName()+"->end");
},"t2").start();
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"->begin");
countDownLatch.countDown(); //1-1=1;
System.out.println(Thread.currentThread().getName()+"->end");
},"t3").start();
countDownLatch.await(); //阻塞Main線程
System.out.println("main end");
}
}
從代碼的實(shí)現(xiàn)來看衬浑，有點(diǎn)類似 join 的功能，但是比 join 更 加靈活放刨。CountDownLatch 構(gòu)造函數(shù)會接收一個(gè) int 類型 的參數(shù)作為計(jì)數(shù)器的初始值，當(dāng)調(diào)用 CountDownLatch 的 countDown 方法時(shí)尸饺，這個(gè)計(jì)數(shù)器就會減一进统。
通過 await 方法去阻塞去阻塞主流程.
源碼分析
對于 CountDownLatch，我們僅僅需要關(guān)心兩個(gè)方法浪听，一 個(gè)是 countDown() 方法螟碎，另一個(gè)是 await() 方法。 countDown() 方法每次調(diào)用都會將 state 減 1迹栓，直到 state 的值為 0;而 await 是一個(gè)阻塞方法掉分，當(dāng) state 減 為 0 的時(shí)候，await 方法才會返回克伊。await 可以被多個(gè)線 程調(diào)用酥郭，大家在這個(gè)時(shí)候腦子里要有個(gè)圖:所有調(diào)用了 await 方法的線程阻塞在 AQS 的阻塞隊(duì)列中，等待條件 滿足(state == 0)愿吹，將線程從隊(duì)列中一個(gè)個(gè)喚醒過來不从。
初始化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);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
protected final void setState(int newState) {
state = newState;
}
await
tryAcquireShared嘗試獲取共享鎖，如果state為0犁跪，不阻塞當(dāng)前線程椿息，否則阻塞，從方法名可以看出坷衍，此處判斷是否獲取共享鎖寝优。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
doAcquireSharedInterruptibly和互斥鎖的代碼基本相同
不同點(diǎn)：
- addWaiter 設(shè)置為 shared 模式。
- tryAcquire 和 tryAcquireShared 的返回值不同枫耳，因此會
多出一個(gè)判斷過程 - 在 判 斷 前 驅(qū) 節(jié) 點(diǎn) 是 頭 節(jié) 點(diǎn) 后 乏矾， 調(diào) 用 了
setHeadAndPropagate 方法，而不是簡單的更新一下頭節(jié)點(diǎn)。
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//添加共享鎖node
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
//嘗試獲取共享鎖
int r = tryAcquireShared(arg);
if (r >= 0) {
//更新頭節(jié)點(diǎn)并釋放其他共享node
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
countDown
由于線程被 await 方法阻塞了妻熊，所以只有等到 countdown 方法使得 state=0 的時(shí)候才會被喚醒夸浅，我們 來看看 countdown 做了什么？
- 只有當(dāng) state 減為 0 的時(shí)候扔役，tryReleaseShared 才返
回 true, 否則只是簡單的 state = state - 1 - 如果 state=0, 則調(diào)用 doReleaseShared
喚醒處于 await 狀態(tài)下的線程
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
//嘗試釋放共享鎖
if (tryReleaseShared(arg)) {
//釋放共享鎖
doReleaseShared();
return true;
}
return false;
}
用自旋的方法實(shí)現(xiàn) state 減 1
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
如果狀態(tài)為0帆喇，判斷頭結(jié)點(diǎn)是不是 SIGNAL 狀態(tài)，如果是亿胸，則修改為 0坯钦，并且喚醒頭結(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
//PROPAGATE: 標(biāo)識為 PROPAGATE 狀態(tài)的節(jié)點(diǎn)，是共享鎖模式下的節(jié)點(diǎn)狀態(tài)侈玄，處于這個(gè)狀態(tài) 下的節(jié)點(diǎn)婉刀，會對線程的喚醒進(jìn)行傳播
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 如果到這里的時(shí)候，前面喚醒的線 程已經(jīng)占領(lǐng)了 head序仙，那么再循環(huán)
// 通過檢查頭節(jié)點(diǎn)是否改變了突颊，如果改變了就繼續(xù)循環(huán)
if (h == head) // loop if head changed
break;
}
}
h == head:說明頭節(jié)點(diǎn)還沒有被剛剛用 unparkSuccessor 喚醒的線程(這里可以理解為 ThreadB)占有，此時(shí) break 退出循環(huán)潘悼。
h != head:頭節(jié)點(diǎn)被剛剛喚醒的線程(這里可以理解為 ThreadB)占有律秃，那么這里重新進(jìn)入下一輪循環(huán)，喚醒下 一個(gè)節(jié)點(diǎn)(這里是 ThreadB )治唤。
doAcquireSharedInterruptibly
一旦 第一個(gè)等待線程 被喚醒棒动，代碼又會繼續(xù)回到 doAcquireSharedInterruptibly 中來執(zhí)行。如果當(dāng)前 state 滿足=0 的條件宾添，則會執(zhí)行 setHeadAndPropagate 方法船惨。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
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())
doReleaseShared();
}
}
這個(gè)方法的主要作用是把被喚醒的節(jié)點(diǎn)，設(shè)置成 head 節(jié) 點(diǎn)缕陕。 然后繼續(xù)喚醒隊(duì)列中的其他線程粱锐。
代碼分析完畢，可以看出ReentrantLock學(xué)習(xí)后扛邑，再學(xué)習(xí)其他的并發(fā)工具的效果是事半功倍的卜范。
