JUC - CountDownLach原理分析
作者:一角錢技術
CountDownLatch 和 Semaphore 一樣都是共享模式下資源問題,這些源碼實現AQS的模版方法,然后使用CAS+循環重試實現自己的功能。
CountDownLach閉鎖
背景
- CountDownLatch是在Java1.5被引入,跟它一起被引入的工具類還有CyclicBarrier、Semaphore、ConcurrenthashMap和BlockingQueue。
- 在java.util.cucurrent包下。
概念
- CountDownLatch這個類使一個線程等待其它線程各自執行完畢后再執行。
- 是通過一個計數器來實現的,計數器的初始值是線程的數量。每當一個線程執行完畢后,計數器的值就-1,當計數器的值為0時,表示所有線程都執行完畢,然后在閉鎖上等待的線程就可以恢復工作來。
源碼
- countDownLatch類中只提供了一個構造器
- public CountDownLatch(int count) {
- if (count < 0) throw new IllegalArgumentException("count < 0");
- this.sync = new Sync(count);
- }
- 類中有三個方法是最重要的
- // 調用await()方法的線程會被掛起,它會等待直到count值為0才繼續執行
- public void await() throws InterruptedException {
- sync.acquireSharedInterruptibly(1);
- }//和await()方法類似,只不過等待一定的時間后count值還沒變為0的化就會繼續執行
- public boolean await(long timeout, TimeUnit unit)
- throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
- }//將count值減1
- public void countDown() { sync.releaseShared(1);
- }
示例
普通示例:
- public class CountDownLatchTest {
- public static void main(String[] args) {
- final CountDownLatch latch = new CountDownLatch(2);
- System.out.println("主線程開始執行…… ……");
- //第一個子線程執行
- ExecutorService es1 = Executors.newSingleThreadExecutor();
- es1.execute(new Runnable() {
- @Override
- public void run() {
- try {
- Thread.sleep(3000);
- System.out.println("子線程:"+Thread.currentThread().getName()+"執行");
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- latch.countDown();
- }
- });
- es1.shutdown();
- //第二個子線程執行
- ExecutorService es2 = Executors.newSingleThreadExecutor();
- es2.execute(new Runnable() {
- @Override
- public void run() {
- try {
- Thread.sleep(3000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- System.out.println("子線程:"+Thread.currentThread().getName()+"執行");
- latch.countDown();
- }
- });
- es2.shutdown();
- System.out.println("等待兩個線程執行完畢…… ……");
- try {
- latch.await();
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- System.out.println("兩個子線程都執行完畢,繼續執行主線程");
- }
- }
結果集:
- 主線程開始執行…… ……
- 等待兩個線程執行完畢…… ……子線程:pool-1-thread-1執行子線程:pool-2-thread-1執行兩個子線程都執行完畢,繼續執行主線程
模擬并發示例:
- public class Parallellimit {
- public static void main(String[] args) {
- ExecutorService pool = Executors.newCachedThreadPool(); CountDownLatch cdl = new CountDownLatch(100);
- for (int i = 0; i < 100; i++) {
- CountRunnable runnable = new CountRunnable(cdl);
- pool.execute(runnable); } }} class CountRunnable implements Runnable {
- private CountDownLatch countDownLatch;
- public CountRunnable(CountDownLatch countDownLatch) {
- this.countDownLatch = countDownLatch;
- } @Override
- public void run() {
- try {
- synchronized (countDownLatch) { /*** 每次減少一個容量*/
- countDownLatch.countDown(); System.out.println("thread counts = " + (countDownLatch.getCount()));
- } countDownLatch.await();
- System.out.println("concurrency counts = " + (100 - countDownLatch.getCount()));
- } catch (InterruptedException e) {
- e.printStackTrace(); } }}
源碼分析
- public class CountDownLatch {
- //繼承AQS來實現他的模板方法(tryAcquireShared,tryReleaseShared)
- private static final class Sync extends AbstractQueuedSynchronizer { //計數個數Count
- Sync(int count) {
- setState(count); } int getCount() {
- return getState();
- } //AQS方法getState(),返回同步狀態,這里指計數器值 protected int tryAcquireShared(int acquires) {
- return (getState() == 0) ? 1 : -1;
- } //循環+cas重試 直到計數器為0 跳出,則release(實現aqs共享模式釋放方法)
- 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;
- } } } private final Sync sync; //實例化
- public CountDownLatch(int count) {
- if (count < 0) throw new IllegalArgumentException("count < 0");
- this.sync = new Sync(count); } public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1);
- } //帶有一個超時時間的awit public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
- } public void countDown() { sync.releaseShared(1);
- } public long getCount() { return sync.getCount();
- }}
總結
CountDownLatch 和 Semaphore 一樣都是共享模式下資源問題,這些源碼實現AQS的模版方法,然后使用CAS+循環重試實現自己的功能。在RT多個資源調用,或者執行某種操作依賴其他操作完成下可以發揮這個計數器的作用。
責任編輯:姜華
來源:
今日頭條