高手程序員與新手程序員一個簡單的判斷標準,就是有沒有使用過CountDownLatch,在互聯網公司工作超過3年的程序員基本上應該都用過。CountDownLatch中文名稱叫做閉鎖,也叫計數鎖,不過不是用來加鎖的,而是通過計數實現條件等待的功能。CountDownLatch的使用場景有兩個:
CountDownLatch常用的方法就兩個,countDown()方法用來將計數器減一,await()方法會阻塞當前線程,直到計數器值等于0。
先看一下第一種場景,也是最常用的場景:
在工作中什么時候會遇到這種場景呢?比如當前線程需要查詢3個數據庫,并且把查詢結果匯總返回給前端。查詢3個數據庫的邏輯,可以分別使用3個線程加快查詢速度。但是怎么判斷3個線程都執行結束了呢?這時候就可以使用CountDownLatch了。
import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/** * @author 一燈架構 * @apiNote CountDownLatch測試類(場景1) **/public class CountDownLatchTest1 { public static void main(String[] args) throws InterruptedException { // 1. 創建一個線程池,用來執行3個查詢任務 ExecutorService executorService = Executors.newFixedThreadPool(3); // 2. 創建一個計數鎖,數量是3 CountDownLatch countDownLatch = new CountDownLatch(3); // 3. 啟動3個查詢任務 for (int i = 0; i < 3; i++) { executorService.submit(() -> { try { // 4. 睡眠1秒,模擬任務執行過程 Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + " 執行完成"); // 5. 任務執行完成,計數器減一 countDownLatch.countDown(); } catch (InterruptedException e) { } }); } // 6. 等待所有任務執行完成 countDownLatch.await(); System.out.println("所有任務執行完成。"); // 7. 關閉線程池 executorService.shutdown(); }}
輸出結果:
pool-1-thread-2 執行完成 pool-1-thread-1 執行完成 pool-1-thread-3 執行完成 所有任務執行完成。
需要注意的是,這里創建CountDownLatch計數器的時候,指定的數量是3,因為有3個任務。在3個任務沒有執行完成之前,await()方法會一直阻塞,直到3個任務都執行完成。
再看一下第二種場景,有些情況用的也比較多:
什么情況下會遇到這種場景呢?比如系統中多個任務線程存在先后依賴關系,必須等待其他線程啟動完成后,才能一起執行。
/** * @author 一燈架構 * @apiNote CountDownLatch測試類(場景2) **/public class CountDownLatchTest { public static void main(String[] args) throws InterruptedException { // 1. 創建一個線程池,用來執行3個任務 ExecutorService executorService = Executors.newFixedThreadPool(3); // 2. 創建一個計數鎖,數量是1 CountDownLatch countDownLatch = new CountDownLatch(1); // 3. 啟動3個任務 for (int i = 0; i < 3; i++) { executorService.submit(() -> { try { System.out.println(Thread.currentThread().getName() + " 啟動完成"); // 4. 等待其他任務啟動完成 countDownLatch.await(); // 5. 睡眠1秒,模擬任務執行過程 Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + " 執行完成"); } catch (InterruptedException e) { } }); } // 6. 所有任務啟動完成,計數器減一 countDownLatch.countDown(); System.out.println("所有任務啟動完成,開始執行。"); // 7. 關閉線程池 executorService.shutdown(); }}
輸出結果:
pool-1-thread-1 啟動完成 pool-1-thread-2 啟動完成 pool-1-thread-3 啟動完成 所有任務啟動完成,開始執行。 pool-1-thread-1 執行完成 pool-1-thread-3 執行完成 pool-1-thread-2 執行完成
需要注意的是,與場景1不同,這里創建CountDownLatch計數器的時候,指定的數量是1,因為3個任務需要滿足同一個條件,就是都啟動完成,也就是只需要調用一次countDown()方法。 看完了CountDownLatch的使用方式,再看一下CountDownLatch的源碼實現。
public class CountDownLatch { // 只有一個Sync同步變量 private final Sync sync; // Sync繼承自AQS,主要邏輯都在這里面 private static final class Sync extends AbstractQueuedSynchronizer { // 只有這一個構造方法,需要指定計數器數值 Sync(int count) { setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } 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; } } }}
跟ReentrantLock一樣,CountDownLatch也沒有直接繼承AQS,也是采用組合的方式,使用Sync同步變量實現計數的功能,而Sync同步變量才是真正繼承AQS的。
public void countDown() { // 底層調用父類AQS中的releaseShared()方法 sync.releaseShared(1);}
countDown()方法里面調用的是父類AQS中的releaseShared()方法,而releaseShared()方法又在調用子類Sync中tryReleaseShared()方法。
/** * 父類AQS */public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { public final boolean releaseShared(int arg) { // tryReleaseShared()由子類實現 if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } // 定義抽象方法,由子類實現 protected boolean tryReleaseShared(int arg) { throw new UnsupportedOperationException(); }}
/** * 子類Sync */private static final class Sync extends AbstractQueuedSynchronizer { // 實現父類AQS中的tryReleaseShared()方法 @Override 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同步類中tryReleaseShared()方法邏輯也很簡單,就是把同步狀態state值減一。
await()方法底層也是調用父類中acquireSharedInterruptibly()方法,而父類AQS又需要調用子類Sync中的具體實現。
public void await() throws InterruptedException { // 底層調用父類AQS中的releaseShared()方法 sync.acquireSharedInterruptibly(1);}
/** * 父類AQS */public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) { throw new InterruptedException(); } // tryAcquireShared()由子類實現 if (tryAcquireShared(arg) < 0) { doAcquireSharedInterruptibly(arg); } } // 定義抽象方法,由子類實現 protected int tryAcquireShared(int arg) { throw new UnsupportedOperationException(); }}
子類Sync只需要實現tryAcquireShared()方法即可,而tryAcquireShared()方法的作用就是判斷鎖是否已經完全釋放,即同步狀態state=0。
/** * 子類Sync */private static final class Sync extends AbstractQueuedSynchronizer { // 實現父類AQS中的tryAcquireShared()方法 @Override protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; }}
看完了CountDownLatch的所有源碼,是不是覺得CountDownLatch邏輯很簡單。
因為加鎖流程的編排工作已經在父類AQS中實現,子類只需要實現具體的加鎖邏輯即可,也就是實現tryReleaseShared()方法和tryAcquireShared()方法。而加鎖邏輯也很簡單,也就是修改同步狀態state的值即可。想要詳細了解父類AQS的流程,可以翻看前幾篇文章。
下篇文章再一塊學習一下共享鎖Semaphore的源碼實現。
本文鏈接:http://www.www897cc.com/showinfo-26-76485-0.html沒看過ReentrantLock源碼,別說精通Java并發編程
聲明:本網頁內容旨在傳播知識,若有侵權等問題請及時與本網聯系,我們將在第一時間刪除處理。郵件:2376512515@qq.com
上一篇: 工作中最常見的六種OOM問題