Java 并發(fā)特性之 CountDownLatch 詳解！
CountDownLatch 是 Java 中的一個(gè)用于管理并發(fā)控制的同步輔助類，允許一個(gè)或多個(gè)線程等待其他線程完成操作，它的工作機(jī)制類似于“倒計(jì)時(shí)閂鎖”，線程會(huì)阻塞等待，直到閂鎖的計(jì)數(shù)器減少到 0，然后才能繼續(xù)執(zhí)行。這篇文章，我們將深度剖析其工作原理。
一、什么是 CountDownLatch？
CountDownLatch是java.util.concurrent 包的一部分，用于同步一個(gè)或多個(gè)線程以等待特定條件的滿足。它在創(chuàng)建時(shí)初始化一個(gè)給定的計(jì)數(shù)，表示必須發(fā)生的事件數(shù)量，才能使線程繼續(xù)執(zhí)行。這個(gè)計(jì)數(shù)通過調(diào)用 countDown() 方法來遞減，等待該條件的線程調(diào)用 await() 方法來阻塞，直到計(jì)數(shù)達(dá)到零。
CountDownLatch的關(guān)鍵組件包含:
- 計(jì)數(shù)：CountDownLatch 的核心概念是計(jì)數(shù)。它從創(chuàng)建鎖存器時(shí)指定的初始值開始，只能遞減，不能重置。
- await() ：線程使用此方法等待計(jì)數(shù)達(dá)到零。如果當(dāng)前計(jì)數(shù)大于零，這些線程將被置于等待狀態(tài)。
- countDown() ：調(diào)用此方法以遞減計(jì)數(shù)。當(dāng)計(jì)數(shù)達(dá)到零時(shí)，所有等待的線程將被釋放。
- 線程安全：CountDownLatch 是線程安全的，它使用內(nèi)部的 AQS（AbstractQueuedSynchronizer）來管理狀態(tài)，確保計(jì)數(shù)的可見性和原子性。
二、工作原理
CountDownLatch 本質(zhì)上是一種簡化的信號(hào)量（Semaphore）。它的核心思想是設(shè)定一個(gè)計(jì)數(shù)器，當(dāng)計(jì)數(shù)器值為 0 時(shí)，其他被阻塞的線程才會(huì)開始運(yùn)行，線程的釋放建立在調(diào)用 countDown 方法去減少計(jì)數(shù)器次數(shù)的基礎(chǔ)上。
CountDownLatch 的典型功能包括：
- 使多個(gè)線程等待一系列事件發(fā)生。
- 讓一個(gè)線程等待完成多個(gè)步協(xié)作操作的線程。
- 在某個(gè)條件達(dá)到之前阻塞線程。
它包含了兩個(gè)核心方法：
- countDown(): 當(dāng)前線程執(zhí)行完任務(wù)后，調(diào)用該方法時(shí)，計(jì)數(shù)器 -1；當(dāng)計(jì)數(shù)器為 1，調(diào)用該方法可以使計(jì)數(shù)器變?yōu)?0。
- await(): 當(dāng)前線程調(diào)用后，會(huì)阻塞，進(jìn)入等待狀態(tài)，直到計(jì)數(shù)器為 0。
通過這兩種操作，我們就可以構(gòu)建出各種靈活的并發(fā)控制邏輯。
1.簡單實(shí)現(xiàn)
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
// 創(chuàng)建一個(gè)計(jì)數(shù)器，設(shè)置初始的計(jì)數(shù)值為 3
CountDownLatch latch = new CountDownLatch(3);
// 創(chuàng)建三個(gè)工作線程
new Thread(() -> {
try {
Thread.sleep(1000); // 模擬任務(wù)耗時(shí)
System.out.println("Thread 1 finished");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown(); // 每個(gè)線程任務(wù)完成后，使計(jì)數(shù)器減 1
}
}).start();
new Thread(() -> {
try {
Thread.sleep(2000); // 模擬任務(wù)耗時(shí)
System.out.println("Thread 2 finished");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown(); // 每個(gè)線程任務(wù)完成后，使計(jì)數(shù)器減 1
}
}).start();
new Thread(() -> {
try {
Thread.sleep(3000); // 模擬任務(wù)耗時(shí)
System.out.println("Thread 3 finished");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown(); // 每個(gè)線程任務(wù)完成后，使計(jì)數(shù)器減 1
}
}).start();
// 主線程會(huì)在此阻塞，直到計(jì)數(shù)器減為 0
latch.await();
System.out.println("All tasks finished. Main thread proceeding.");
}
}輸出結(jié)果如下：
Thread 1 finished
Thread 2 finished
Thread 3 finished
All tasks finished. Main thread proceeding.在上述代碼中，主線程調(diào)用 latch.await(); 進(jìn)入阻塞狀態(tài)，等待三個(gè)工作線程完成任務(wù)后，計(jì)數(shù)器將變?yōu)?0，然后解除阻塞并進(jìn)入后續(xù)邏輯。
2.底層工作原理
從底層工作原理來看，CountDownLatch 內(nèi)部維護(hù)了一個(gè) Sync 類，這實(shí)際上是一個(gè)基于 AQS（AbstractQueuedSynchronizer, 抽象隊(duì)列同步器）的同步工具。
- State 的初始值為計(jì)數(shù)器值，也就是通過構(gòu)造函數(shù)傳遞的參數(shù)（n）。
- await 方法的實(shí)現(xiàn)就是通過驗(yàn)證 state 的值是否為 0，若不為 0，則會(huì)阻塞當(dāng)前線程并加入AQS等待隊(duì)列中，否則繼續(xù)向下執(zhí)行。
- countDown 方法會(huì)將 state 的值減 1，當(dāng)state==0時(shí)，會(huì)喚醒所有在 AQS 阻塞隊(duì)列中的線程。
內(nèi)部實(shí)現(xiàn)機(jī)制對(duì)線程的阻塞、喚醒、隊(duì)列管理等是通過 AQS 實(shí)現(xiàn)的，AQS 的設(shè)計(jì)模式使得它能高效、安全地管理同步狀態(tài)。
3.核心代碼片段
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) {
// 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;
}
}
}三、使用場(chǎng)景分析
CountDownLatch 的應(yīng)用場(chǎng)景比較廣泛，尤其是在處理并發(fā)問題時(shí)，這里列舉了幾個(gè)：
1.批量任務(wù)協(xié)調(diào)
有時(shí)候，不同子線程可能會(huì)同時(shí)執(zhí)行各自的任務(wù)，然而主線程會(huì)等待所有子線程的執(zhí)行完畢后，才繼續(xù)執(zhí)行后續(xù)操作。
比如 Web 應(yīng)用中多個(gè) API 的響應(yīng)聚合：假設(shè)有多個(gè)遠(yuǎn)程服務(wù)需要調(diào)用，主線程希望在所有調(diào)用都返回結(jié)果后，再執(zhí)行后續(xù)處理，可以使用 CountDownLatch 來等待響應(yīng)的到來。示例代碼如下：
// 類似一些應(yīng)用需要同時(shí)從多個(gè)微服務(wù)中拉數(shù)據(jù)，再一起處理
CountDownLatch latch = new CountDownLatch(3);
ExecutorService executor = Executors.newFixedThreadPool(3);
for (int i = 0; i < 3; i++) {
executor.submit(() -> {
try {
// 模擬拉取和處理數(shù)據(jù)
} catch (Exception e) {
// 異常處理
} finally {
latch.countDown(); // 每個(gè)任務(wù)結(jié)束后調(diào)用
}
});
}
latch.await(); // 等待所有子任務(wù)執(zhí)行結(jié)束
System.out.println("匯總所有數(shù)據(jù).");2.并行計(jì)算
假如有這樣一個(gè)情景：計(jì)算任務(wù)很耗時(shí)，但是可以分成多個(gè)部分并行處理，然后將結(jié)果進(jìn)行合并。
實(shí)現(xiàn)方式：我們先將任務(wù)分解成 n 個(gè)子任務(wù)，全部執(zhí)行完畢后，將子任務(wù)的結(jié)果進(jìn)行匯總分析。示例代碼如下：
CountDownLatch latch = new CountDownLatch(n);
List<Integer> results = new CopyOnWriteArrayList<>();
for (int i = 0; i < n; i++) {
new Thread(() -> {
try {
int result = // 處理部分任務(wù)
results.add(result);
} finally {
latch.countDown(); // 完成后計(jì)數(shù)減 1
}
}).start();
}
latch.await(); // 等到結(jié)果全部處理完
int finalResult = results.stream().mapToInt(Integer::intValue).sum();3.服務(wù)啟動(dòng)檢查
CountDownLatch 還可以為應(yīng)用服務(wù)做“健康檢查”。例如，系統(tǒng)在完全啟動(dòng)之前，需要依賴多個(gè)外部服務(wù)，那么我們可以通過異步方式檢測(cè)各個(gè)服務(wù)的健康狀態(tài)，只有當(dāng)所有服務(wù)都正常啟動(dòng)時(shí)，才允許繼續(xù)執(zhí)行下一步。
簡單的示例代碼如下：
public class ServiceStartChecker {
private final CountDownLatch latch;
public ServiceStartChecker(int serviceCount) {
latch = new CountDownLatch(serviceCount);
}
public void checkServices() throws InterruptedException {
// 啟動(dòng)多個(gè)異步線程去檢查服務(wù)是否就緒
for (int i = 0; i < 3; i++) {
new Thread(() -> {
try {
// 模擬服務(wù)檢查
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() + " is ready");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown(); // 完成后調(diào)用，減少計(jì)數(shù)器
}
}).start();
}
latch.await(); // 等待所有服務(wù)就緒
System.out.println("All services are up. System is ready.");
}
public static void main(String[] args) throws InterruptedException {
ServiceStartChecker checker = new ServiceStartChecker(3);
checker.checkServices();
}
}四、與其他并發(fā)工具對(duì)比
CountDownLatch 只是 Java 并發(fā)工具包中的一個(gè)工具，其功能與一些其他工具如 CyclicBarrier、Semaphore 等具有一定的共性和不同點(diǎn)。
1.CountDownLatch vs CyclicBarrier
- 用途：CountDownLatch適用于一組線程完成某一工作后進(jìn)入“繼續(xù)工作”的狀態(tài)，且無法進(jìn)行reset重新使用。而 CyclicBarrier 更適合復(fù)用，在一組線程都到達(dá)某一屏障時(shí)統(tǒng)一放行，之后可以通過 reset 重復(fù)使用。
- 結(jié)束條件：CyclicBarrier 需要每個(gè)等待的線程都到達(dá)某個(gè)同步點(diǎn)才能繼續(xù)；而 CountDownLatch 則更靈活，它并不關(guān)心是哪個(gè)線程調(diào)用了 countDown，只關(guān)注 countDown 是否次數(shù)到了。
2.CountDownLatch vs Semaphore
任務(wù)控制力度的差異：Semaphore 更傾向于對(duì)信號(hào)量的數(shù)量進(jìn)行限流。簡單來說，Semaphore 可以限制某個(gè)操作的并發(fā)次數(shù)，比如最多只允許 5 個(gè)線程同時(shí)執(zhí)行某個(gè)任務(wù)。而 CountDownLatch 只是簡單的減少計(jì)數(shù)，不去限流，只是關(guān)注完成情況。
五、實(shí)際項(xiàng)目中的使用
在多線程爬蟲、分布式系統(tǒng)、并行數(shù)據(jù)處理等具體項(xiàng)目中，CountDownLatch 都能找到合適的應(yīng)用場(chǎng)景。
1.分布式系統(tǒng)的啟動(dòng)控制
假設(shè)我們?cè)谝粋€(gè)分布式服務(wù)系統(tǒng)中，每個(gè)微服務(wù)間可能有復(fù)雜的依賴關(guān)系，借助 CountDownLatch，我可以構(gòu)建出一個(gè)依賴的啟動(dòng)順序。
2.性能測(cè)試
在進(jìn)行性能測(cè)試時(shí)，可能需要多個(gè)線程同時(shí)工作，例如使用 CountDownLatch 控制開始時(shí)間，以模擬高并發(fā)訪問場(chǎng)景。
CountDownLatch ready = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(N);
for (int i = 0; i < N; i++) {
new Thread(() -> {
try {
ready.await(); // 等待所有線程就緒
// 執(zhí)行模擬請(qǐng)求
} finally {
done.countDown();
}
}).start();
}
// 開始測(cè)試
ready.countDown();
done.await(); // 等到所有線程結(jié)束通過這樣的實(shí)踐，我們可以輕松模擬高并發(fā)性能測(cè)試和壓力測(cè)試場(chǎng)景。
總結(jié)
本文我們深度剖析了CountDownLatch，CountDownLatch雖然是一個(gè)簡單的并發(fā)工具，對(duì)其整體總結(jié)如下：
- 核心工作原理：CountDownLatch基于AQS機(jī)制，用于管理一個(gè)線程集合的執(zhí)行流程控制。
- 適用場(chǎng)景: 主要用于在處理并行操作時(shí)控制線程的執(zhí)行順序。
- 與其他工具的對(duì)比：與CyclicBarrier和信號(hào)量Semaphore分別有不同的貢獻(xiàn)場(chǎng)景。
- 多應(yīng)用場(chǎng)景使用：包括但不限于服務(wù)啟動(dòng)依賴、并行計(jì)算結(jié)果收集、并發(fā)控制等。
