此篇博客所有源碼均來自JDK 1.8
在上篇博客中介紹了Java四大并發(fā)工具之一的CyclicBarrier，今天要介紹的CountDownLatch與CyclicBarrier有點兒相似。
CyclicBarrier所描述的是“允許一組線程互相等待，直到到達某個公共屏障點，才會進行后續(xù)任務"，而CountDownLatch所描述的是”在完成一組正在其他線程中執(zhí)行的操作之前，它允許一個或多個線程一直等待“。在API中是這樣描述的：
用給定的計數(shù) 初始化 CountDownLatch。由于調用了 countDown() 方法，所以在當前計數(shù)到達零之前，await 方法會一直受阻塞。之后，會釋放所有等待的線程，await 的所有后續(xù)調用都將立即返回。這種現(xiàn)象只出現(xiàn)一次——計數(shù)無法被重置。如果需要重置計數(shù)，請考慮使用 CyclicBarrier。
CountDownLatch是通過一個計數(shù)器來實現(xiàn)的，當我們在new 一個CountDownLatch對象的時候需要帶入該計數(shù)器值，該值就表示了線程的數(shù)量。每當一個線程完成自己的任務后，計數(shù)器的值就會減1。當計數(shù)器的值變?yōu)?時，就表示所有的線程均已經完成了任務，然后就可以恢復等待的線程繼續(xù)執(zhí)行了。
雖然，CountDownlatch與CyclicBarrier有那么點相似，但是他們還是存在一些區(qū)別的：
- CountDownLatch的作用是允許1或N個線程等待其他線程完成執(zhí)行；而CyclicBarrier則是允許N個線程相互等待
- CountDownLatch的計數(shù)器無法被重置；CyclicBarrier的計數(shù)器可以被重置后使用，因此它被稱為是循環(huán)的barrier
實現(xiàn)分析
CountDownLatch結構如下
通過上面的結構圖我們可以看到，CountDownLatch內部依賴Sync實現(xiàn)，而Sync繼承AQS。CountDownLatch僅提供了一個構造方法：
CountDownLatch(int count) ： 構造一個用給定計數(shù)初始化的 CountDownLatch
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
sync為CountDownLatch的一個內部類，其定義如下：
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
//獲取同步狀態(tài)
int getCount() {
return getState();
}
//獲取同步狀態(tài)
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
//釋放同步狀態(tài)
protected boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
通過這個內部類Sync我們可以清楚地看到CountDownLatch是采用共享鎖來實現(xiàn)的。
await()
CountDownLatch提供await()方法來使當前線程在鎖存器倒計數(shù)至零之前一直等待，除非線程被中斷，定義如下：
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
await其內部使用AQS的acquireSharedInterruptibly(int arg)：
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
在內部類Sync中重寫了tryAcquireShared(int arg)方法：
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
getState()獲取同步狀態(tài)，其值等于計數(shù)器的值，從這里我們可以看到如果計數(shù)器值不等于0，則會調用doAcquireSharedInterruptibly(int arg)，該方法為一個自旋方法會嘗試一直去獲取同步狀態(tài)：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
/**
* 對于CountDownLatch而言，如果計數(shù)器值不等于0，那么r 會一直小于0
*/
int r = tryAcquireShared(arg);
if (r >= 0) {
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()
CountDownLatch提供countDown() 方法遞減鎖存器的計數(shù)，如果計數(shù)到達零，則釋放所有等待的線程。
public void countDown() {
sync.releaseShared(1);
}
內部調用AQS的releaseShared(int arg)方法來釋放共享鎖同步狀態(tài)：
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
tryReleaseShared(int arg)方法被CountDownLatch的內部類Sync重寫：
protected boolean tryReleaseShared(int releases) {
for (;;) {
//獲取鎖狀態(tài)
int c = getState();
//c == 0 直接返回，釋放鎖成功
if (c == 0)
return false;
//計算新“鎖計數(shù)器”
int nextc = c-1;
//更新鎖狀態(tài)（計數(shù)器）
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
總結
CountDownLatch內部通過共享鎖實現(xiàn)。在創(chuàng)建CountDownLatch實例時，需要傳遞一個int型的參數(shù)：count，該參數(shù)為計數(shù)器的初始值，也可以理解為該共享鎖可以獲取的總次數(shù)。當某個線程調用await()方法，程序首先判斷count的值是否為0，如果不會0的話則會一直等待直到為0為止。當其他線程調用countDown()方法時，則執(zhí)行釋放共享鎖狀態(tài)，使count值 - 1。當在創(chuàng)建CountDownLatch時初始化的count參數(shù)，必須要有count線程調用countDown方法才會使計數(shù)器count等于0，鎖才會釋放，前面等待的線程才會繼續(xù)運行。注意CountDownLatch不能回滾重置。
關于共享鎖的請參考：【死磕Java并發(fā)】-----J.U.C之AQS：同步狀態(tài)的獲取與釋放
應用示例
示例仍然使用開會案例。老板進入會議室等待5個人全部到達會議室才會開會。所以這里有兩個線程老板等待開會線程、員工到達會議室：
public class CountDownLatchTest {
private static CountDownLatch countDownLatch = new CountDownLatch(5);
/**
* Boss線程，等待員工到達開會
*/
static class BossThread extends Thread{
@Override
public void run() {
System.out.println("Boss在會議室等待，總共有" + countDownLatch.getCount() + "個人開會...");
try {
//Boss等待
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("所有人都已經到齊了，開會吧...");
}
}
//員工到達會議室
static class EmpleoyeeThread extends Thread{
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "，到達會議室....");
//員工到達會議室 count - 1
countDownLatch.countDown();
}
}
public static void main(String[] args){
//Boss線程啟動
new BossThread().start();
for(int i = 0 ; i < countDownLatch.getCount() ; i++){
new EmpleoyeeThread().start();
}
}
}
運行結果：
此篇博客所有源碼均來自JDK 1.8
在上篇博客中介紹了Java四大并發(fā)工具之一的CyclicBarrier，今天要介紹的CountDownLatch與CyclicBarrier有點兒相似。
CyclicBarrier所描述的是“允許一組線程互相等待，直到到達某個公共屏障點，才會進行后續(xù)任務"，而CountDownLatch所描述的是”在完成一組正在其他線程中執(zhí)行的操作之前，它允許一個或多個線程一直等待“。在API中是這樣描述的：
用給定的計數(shù) 初始化 CountDownLatch。由于調用了 countDown() 方法，所以在當前計數(shù)到達零之前，await 方法會一直受阻塞。之后，會釋放所有等待的線程，await 的所有后續(xù)調用都將立即返回。這種現(xiàn)象只出現(xiàn)一次——計數(shù)無法被重置。如果需要重置計數(shù)，請考慮使用 CyclicBarrier。
CountDownLatch是通過一個計數(shù)器來實現(xiàn)的，當我們在new 一個CountDownLatch對象的時候需要帶入該計數(shù)器值，該值就表示了線程的數(shù)量。每當一個線程完成自己的任務后，計數(shù)器的值就會減1。當計數(shù)器的值變?yōu)?時，就表示所有的線程均已經完成了任務，然后就可以恢復等待的線程繼續(xù)執(zhí)行了。
雖然，CountDownlatch與CyclicBarrier有那么點相似，但是他們還是存在一些區(qū)別的：
- CountDownLatch的作用是允許1或N個線程等待其他線程完成執(zhí)行；而CyclicBarrier則是允許N個線程相互等待
- CountDownLatch的計數(shù)器無法被重置；CyclicBarrier的計數(shù)器可以被重置后使用，因此它被稱為是循環(huán)的barrier
實現(xiàn)分析
CountDownLatch結構如下
通過上面的結構圖我們可以看到，CountDownLatch內部依賴Sync實現(xiàn)，而Sync繼承AQS。CountDownLatch僅提供了一個構造方法：
CountDownLatch(int count) ： 構造一個用給定計數(shù)初始化的 CountDownLatch
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
sync為CountDownLatch的一個內部類，其定義如下：
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
//獲取同步狀態(tài)
int getCount() {
return getState();
}
//獲取同步狀態(tài)
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
//釋放同步狀態(tài)
protected boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
通過這個內部類Sync我們可以清楚地看到CountDownLatch是采用共享鎖來實現(xiàn)的。
await()
CountDownLatch提供await()方法來使當前線程在鎖存器倒計數(shù)至零之前一直等待，除非線程被中斷，定義如下：
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
await其內部使用AQS的acquireSharedInterruptibly(int arg)：
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
在內部類Sync中重寫了tryAcquireShared(int arg)方法：
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
getState()獲取同步狀態(tài)，其值等于計數(shù)器的值，從這里我們可以看到如果計數(shù)器值不等于0，則會調用doAcquireSharedInterruptibly(int arg)，該方法為一個自旋方法會嘗試一直去獲取同步狀態(tài)：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
/**
* 對于CountDownLatch而言，如果計數(shù)器值不等于0，那么r 會一直小于0
*/
int r = tryAcquireShared(arg);
if (r >= 0) {
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()
CountDownLatch提供countDown() 方法遞減鎖存器的計數(shù)，如果計數(shù)到達零，則釋放所有等待的線程。
public void countDown() {
sync.releaseShared(1);
}
內部調用AQS的releaseShared(int arg)方法來釋放共享鎖同步狀態(tài)：
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
tryReleaseShared(int arg)方法被CountDownLatch的內部類Sync重寫：
protected boolean tryReleaseShared(int releases) {
for (;;) {
//獲取鎖狀態(tài)
int c = getState();
//c == 0 直接返回，釋放鎖成功
if (c == 0)
return false;
//計算新“鎖計數(shù)器”
int nextc = c-1;
//更新鎖狀態(tài)（計數(shù)器）
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
總結
CountDownLatch內部通過共享鎖實現(xiàn)。在創(chuàng)建CountDownLatch實例時，需要傳遞一個int型的參數(shù)：count，該參數(shù)為計數(shù)器的初始值，也可以理解為該共享鎖可以獲取的總次數(shù)。當某個線程調用await()方法，程序首先判斷count的值是否為0，如果不會0的話則會一直等待直到為0為止。當其他線程調用countDown()方法時，則執(zhí)行釋放共享鎖狀態(tài)，使count值 - 1。當在創(chuàng)建CountDownLatch時初始化的count參數(shù)，必須要有count線程調用countDown方法才會使計數(shù)器count等于0，鎖才會釋放，前面等待的線程才會繼續(xù)運行。注意CountDownLatch不能回滾重置。
關于共享鎖的請參考：【死磕Java并發(fā)】-----J.U.C之AQS：同步狀態(tài)的獲取與釋放
應用示例
示例仍然使用開會案例。老板進入會議室等待5個人全部到達會議室才會開會。所以這里有兩個線程老板等待開會線程、員工到達會議室：
public class CountDownLatchTest {
private static CountDownLatch countDownLatch = new CountDownLatch(5);
/**
* Boss線程，等待員工到達開會
*/
static class BossThread extends Thread{
@Override
public void run() {
System.out.println("Boss在會議室等待，總共有" + countDownLatch.getCount() + "個人開會...");
try {
//Boss等待
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("所有人都已經到齊了，開會吧...");
}
}
//員工到達會議室
static class EmpleoyeeThread extends Thread{
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "，到達會議室....");
//員工到達會議室 count - 1
countDownLatch.countDown();
}
}
public static void main(String[] args){
//Boss線程啟動
new BossThread().start();
for(int i = 0 ; i < countDownLatch.getCount() ; i++){
new EmpleoyeeThread().start();
}
}
}
運行結果：
此篇博客所有源碼均來自JDK 1.8
在上篇博客中介紹了Java四大并發(fā)工具之一的CyclicBarrier，今天要介紹的CountDownLatch與CyclicBarrier有點兒相似。
CyclicBarrier所描述的是“允許一組線程互相等待，直到到達某個公共屏障點，才會進行后續(xù)任務"，而CountDownLatch所描述的是”在完成一組正在其他線程中執(zhí)行的操作之前，它允許一個或多個線程一直等待“。在API中是這樣描述的：
用給定的計數(shù) 初始化 CountDownLatch。由于調用了 countDown() 方法，所以在當前計數(shù)到達零之前，await 方法會一直受阻塞。之后，會釋放所有等待的線程，await 的所有后續(xù)調用都將立即返回。這種現(xiàn)象只出現(xiàn)一次——計數(shù)無法被重置。如果需要重置計數(shù)，請考慮使用 CyclicBarrier。
[圖片上傳中。。。（1）]
CountDownLatch是通過一個計數(shù)器來實現(xiàn)的，當我們在new 一個CountDownLatch對象的時候需要帶入該計數(shù)器值，該值就表示了線程的數(shù)量。每當一個線程完成自己的任務后，計數(shù)器的值就會減1。當計數(shù)器的值變?yōu)?時，就表示所有的線程均已經完成了任務，然后就可以恢復等待的線程繼續(xù)執(zhí)行了。
雖然，CountDownlatch與CyclicBarrier有那么點相似，但是他們還是存在一些區(qū)別的：
- CountDownLatch的作用是允許1或N個線程等待其他線程完成執(zhí)行；而CyclicBarrier則是允許N個線程相互等待
- CountDownLatch的計數(shù)器無法被重置；CyclicBarrier的計數(shù)器可以被重置后使用，因此它被稱為是循環(huán)的barrier
實現(xiàn)分析
CountDownLatch結構如下
[圖片上傳中。。。（2）]
通過上面的結構圖我們可以看到，CountDownLatch內部依賴Sync實現(xiàn)，而Sync繼承AQS。CountDownLatch僅提供了一個構造方法：
CountDownLatch(int count) ： 構造一個用給定計數(shù)初始化的 CountDownLatch
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
sync為CountDownLatch的一個內部類，其定義如下：
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
//獲取同步狀態(tài)
int getCount() {
return getState();
}
//獲取同步狀態(tài)
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
//釋放同步狀態(tài)
protected boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
通過這個內部類Sync我們可以清楚地看到CountDownLatch是采用共享鎖來實現(xiàn)的。
await()
CountDownLatch提供await()方法來使當前線程在鎖存器倒計數(shù)至零之前一直等待，除非線程被中斷，定義如下：
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
await其內部使用AQS的acquireSharedInterruptibly(int arg)：
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
在內部類Sync中重寫了tryAcquireShared(int arg)方法：
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
getState()獲取同步狀態(tài)，其值等于計數(shù)器的值，從這里我們可以看到如果計數(shù)器值不等于0，則會調用doAcquireSharedInterruptibly(int arg)，該方法為一個自旋方法會嘗試一直去獲取同步狀態(tài)：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
/**
* 對于CountDownLatch而言，如果計數(shù)器值不等于0，那么r 會一直小于0
*/
int r = tryAcquireShared(arg);
if (r >= 0) {
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()
CountDownLatch提供countDown() 方法遞減鎖存器的計數(shù)，如果計數(shù)到達零，則釋放所有等待的線程。
public void countDown() {
sync.releaseShared(1);
}
內部調用AQS的releaseShared(int arg)方法來釋放共享鎖同步狀態(tài)：
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
tryReleaseShared(int arg)方法被CountDownLatch的內部類Sync重寫：
protected boolean tryReleaseShared(int releases) {
for (;;) {
//獲取鎖狀態(tài)
int c = getState();
//c == 0 直接返回，釋放鎖成功
if (c == 0)
return false;
//計算新“鎖計數(shù)器”
int nextc = c-1;
//更新鎖狀態(tài)（計數(shù)器）
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
總結
CountDownLatch內部通過共享鎖實現(xiàn)。在創(chuàng)建CountDownLatch實例時，需要傳遞一個int型的參數(shù)：count，該參數(shù)為計數(shù)器的初始值，也可以理解為該共享鎖可以獲取的總次數(shù)。當某個線程調用await()方法，程序首先判斷count的值是否為0，如果不會0的話則會一直等待直到為0為止。當其他線程調用countDown()方法時，則執(zhí)行釋放共享鎖狀態(tài)，使count值 - 1。當在創(chuàng)建CountDownLatch時初始化的count參數(shù)，必須要有count線程調用countDown方法才會使計數(shù)器count等于0，鎖才會釋放，前面等待的線程才會繼續(xù)運行。注意CountDownLatch不能回滾重置。
關于共享鎖的請參考：【死磕Java并發(fā)】-----J.U.C之AQS：同步狀態(tài)的獲取與釋放
應用示例
示例仍然使用開會案例。老板進入會議室等待5個人全部到達會議室才會開會。所以這里有兩個線程老板等待開會線程、員工到達會議室：
public class CountDownLatchTest {
private static CountDownLatch countDownLatch = new CountDownLatch(5);
/**
* Boss線程，等待員工到達開會
*/
static class BossThread extends Thread{
@Override
public void run() {
System.out.println("Boss在會議室等待，總共有" + countDownLatch.getCount() + "個人開會...");
try {
//Boss等待
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("所有人都已經到齊了，開會吧...");
}
}
//員工到達會議室
static class EmpleoyeeThread extends Thread{
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "，到達會議室....");
//員工到達會議室 count - 1
countDownLatch.countDown();
}
}
public static void main(String[] args){
//Boss線程啟動
new BossThread().start();
for(int i = 0 ; i < countDownLatch.getCount() ; i++){
new EmpleoyeeThread().start();
}
}
}
運行結果：
