一、基本概念
CountDownLatch ：倒計時器，是一個同步工具類，它允許一個或多個線程一直等待，直到其他線程的操作執行完後再執行。
二、CountDownLatch 的三種典型用法
- 某一線程在開始運作前等待 n 個線程執行完畢。
- 實作多個線程開始執行任務的最大并行性。注意是并行性，不是并發，強調的是多個線程在某一時刻同時開始執行。類似于賽跑，将多個線程放到起點，等待發令槍響，然後同時開跑。
- 死鎖的檢測
（1） 某一線程在開始運作前等待 n 個線程執行完畢。
将 CountDownLatch 的計數器初始化為 n ：new CountDownLatch(n)，每當一個任務線程執行完畢，就将計數器減 1 countdownlatch.countDown()，當計數器的值變為 0 時，在CountDownLatch上 await() 的線程就會被喚醒。一個典型應用場景就是啟動一個服務時，主線程需要等待多個元件加載完畢，之後再繼續執行。
情況①：設定周遊次數、核心線程數 countdownLatch都為size時，直到等待所有核心線程都執行完成後，再調用主線程的代碼
public class CountDownLatchTest {
private static final int size = 8;
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(size);
CountDownLatch countDownLatch = new CountDownLatch(size);
IntStream.range(0,size).forEach(
r-> executorService.execute( () -> {
System.out.println(Thread.currentThread().getName() + " start countDownLatch Test" + r);
try {
Thread.sleep(1000);//模拟耗時操作
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " end countDownLatch Test"+ r);
countDownLatch.countDown();
})
);
executorService.shutdown();
countDownLatch.await();
System.out.println("Finish countDownLatch Test");
}
}
運作結果：
情況②：設定核心線程數與周遊次數和countdownLatch不等時，也是同樣的效果。
注：執行完的線程也會進入阻塞狀态
public class CountDownLatchTest {
private static final int size = 8;
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = new ThreadPoolExecutor(5,10, 0,TimeUnit.SECONDS,
new LinkedBlockingDeque<Runnable>());
CountDownLatch countDownLatch = new CountDownLatch(size);
IntStream.range(0,size).forEach(
r-> executorService.execute( () -> {
System.out.println(Thread.currentThread().getName() + " start countDownLatch Test" + r);
try {
Thread.sleep(1000); //模拟耗時操作
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " end countDownLatch Test"+ r);
countDownLatch.countDown();
})
);
executorService.shutdown();
countDownLatch.await();
System.out.println("Finish countDownLatch Test");
}
結果：
（二）實作多個線程開始執行任務的最大并行性。注意是并行性，不是并發，強調的是多個線程在某一時刻同時開始執行。類似于賽跑，将多個線程放到起點，等待發令槍響，然後同時開跑。
做法是初始化一個共享的 CountDownLatch 對象，将其計數器初始化為 1 ：new CountDownLatch(1)，多個線程在開始執行任務前首先 coundownlatch.await()，當主線程調用 countDown() 時，計數器變為 0，多個線程同時被喚醒。
public class CountDownLatchTest1 {
private static final int size = 8;
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = new ThreadPoolExecutor(5,10, 0,TimeUnit.SECONDS,
new LinkedBlockingDeque<Runnable>());
CountDownLatch countDownLatch = new CountDownLatch(1);
IntStream.range(0,size).forEach(
r-> executorService.execute( () -> {
System.out.println(Thread.currentThread().getName() + " start countDownLatch Test" + r);
try {
countDownLatch.await();
Thread.sleep(1000);//模拟耗時操作
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " end countDownLatch Test"+ r);
})
);
executorService.shutdown();
countDownLatch.countDown();
System.out.println("Finish countDownLatch Test");
}
}
運作結果：
（三）死鎖的檢測：
一個非常友善的使用場景是，你可以使用 n 個線程通路共享資源，在每次測試階段的線程數目是不同的，并嘗試産生死鎖。
public class CountDownLatchTest2 {
public static void main(String[] args) throws InterruptedException {
System.out.println("Hello World!");
ExecutorService executorService = Executors.newFixedThreadPool(2);
byte[] i = new byte[0];
byte[] j = new byte[0];
final CountDownLatch countDownLatch = new CountDownLatch(2);
executorService.execute(new DeadThread1(i, j,countDownLatch));
executorService.execute(new DeadThread2(i, j,countDownLatch));
countDownLatch.await();
System.out.println("done !!!");
}
public static class DeadThread1 implements Runnable {
private byte[] i;
private byte[] j;
private CountDownLatch countDownLatch;
public DeadThread1(byte[] i, byte[] j, CountDownLatch countDownLatch) {
this.i = i;
this.j = j;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
synchronized (i) {
try {
System.out.println(Thread.currentThread().getName() + "step1 is running!!");
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (j) {
System.out.println(Thread.currentThread().getName() + " step2 is running!!");
countDownLatch.countDown();
System.out.println(Thread.currentThread().getName() + " step3 is end!!");
}
}
}
}
public static class DeadThread2 implements Runnable {
private byte[] i;
private byte[] j;
private CountDownLatch countDownLatch;
public DeadThread2(byte[] i, byte[] j, CountDownLatch countDownLatch) {
this.i = i;
this.j = j;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
synchronized (j) {
try {
System.out.println(Thread.currentThread().getName() + " step1 is running!!");
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (i) {
System.out.println(Thread.currentThread().getName() + " step2 is running!!");
countDownLatch.countDown();
System.out.println(Thread.currentThread().getName() + " step3 is end!!");
}
}
}
}
}
分析：由于發生死鎖，是以子線程并沒有執行countdown(）方法，是以主線程調用await()發生阻塞，沒有列印出“done !!!”
運作結果：
三、CountDownLatch實作原理
源碼：
CountDownLatch類内，與ReentrantLock類似，有靜态内部類AbstractQueuedSynchronizer的Sync
，并重寫了父類的方法
- tryAcquireShared()：共享方式，嘗試擷取資源方法，能擷取到傳回true
- tryReleaseShared()：共享方式，嘗試釋放資源方法，能擷取到傳回true
public class CountDownLatch {
/**
* Synchronization control For CountDownLatch.
* Uses AQS state to represent 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;
}
}
}
（1）CountDownLatch()構造方法，初始化Sync
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
(2) countDown() -同步器釋放一個資源，電腦減1
/**
* 同步器釋放一個資源，電腦減1
**/
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
//自旋
for (;;) {
//獲得目前計數器的數值
int c = getState();
if (c == 0)
//如果計數器為0,直接傳回
return false;
int nextc = c-1;
//調用cas，設定修改state = c-1
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
（3）await()-判斷目前計數器數值，如果為0,
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//如果目前計數器state為0，await（）執行完成
//如果目前計數器state部位0，調用doAcquireSharedInterruptibly，目前線程進入阻塞狀态
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
四、CountDownLatch的不足
CountDownLatch 是一次性的，計數器的值隻能在構造方法中初始化一次，之後沒有任何機制再次對其設定值，當 CountDownLatch 使用完畢後，它不能再次被使用。