JUC 同步工具類全景圖：閉鎖、柵欄、信號量、阻塞隊列——從原理到生產級用法
java.util.concurrent 包提供了四大同步工具類，封裝了線程協調所需的狀態與操作，讓開發者無需手寫 wait/notify 即可實現複雜協作。本文通過可運行 Demo 逐一拆解 CountDownLatch、CyclicBarrier、Semaphore、BlockingQueue 的核心機制、典型場景與避坑指南。
一、閉鎖 —— CountDownLatch：一個線程等待多個事件完成
核心模型
- 內部計數器
count，初始化為 N
countDown()：計數器 -1（不可重置）await()：阻塞直到計數器歸零
典型場景：主線程等待多個初始化任務完成
public class CountDownLatchDemo {
private static final Logger log = LoggerFactory.getLogger(CountDownLatchDemo.class);
public static void main(String[] args) throws InterruptedException {
int taskCount = 3;
CountDownLatch latch = new CountDownLatch(taskCount);
ExecutorService executor = Executors.newFixedThreadPool(taskCount);
for (int i = 1; i <= taskCount; i++) {
final int id = i;
executor.submit(() -> {
try {
log.info("Task-{} start", id);
Thread.sleep(1000L * id); // 模擬耗時初始化
log.info("Task-{} finish", id);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
latch.countDown(); // 關鍵：任務結束倒數
}
});
}
log.info("Main thread waiting...");
latch.await(); // 阻塞直到所有任務 countDown
log.info("All tasks done, main continue...");
executor.shutdown();
}
}
輸出
Main thread waiting...
Task-1 start
Task-2 start
Task-3 start
Task-1 finish
Task-2 finish
Task-3 finish
All tasks done, main continue...
進階用法：FutureTask 亦可作閉鎖
FutureTask future = new FutureTask<>(() -> { /* 耗時計算 */ return "result"; });
new Thread(future).start();
String result = future.get(); // 相當於 await()，阻塞直到任務完成
二、柵欄 —— CyclicBarrier：多線程相互等待，齊點同步
核心差異
| 特性 | CountDownLatch | CyclicBarrier |
|---|
| 計數器可重置 | ❌ 一次性 | ✅ reset() 可重用 |
| 等待主體 | 通常 1 個線程等 N 個 | N 個線程**相互等待** |
| 攔截動作 | 無 | 可選 Runnable barrierAction，最後到達線程執行 |
| 異常處理 | 無特殊 | 任一線程異常 → 所有等待線程拋 BrokenBarrierException |
典型場景：多線程分階段計算，每階段匯總再進入下一階段
public class CyclicBarrierDemo {
private static final Logger log = LoggerFactory.getLogger(CyclicBarrierDemo.class);
public static void main(String[] args) {
int parties = 3;
CyclicBarrier barrier = new CyclicBarrier(parties, () ->
log.info("=== Barrier action: all parties arrived, doing aggregation ===")
);
ExecutorService executor = Executors.newFixedThreadPool(parties);
for (int i = 1; i <= parties; i++) {
final int id = i;
executor.submit(() -> {
try {
for (int phase = 1; phase <= 2; phase++) {
log.info("Worker-{} phase-{} working...", id, phase);
Thread.sleep(500L * id);
log.info("Worker-{} phase-{} await barrier", id, phase);
barrier.await(); // 阻塞直到所有 worker 到達
log.info("Worker-{} phase-{} continue", id, phase);
}
} catch (InterruptedException | BrokenBarrierException e) {
Thread.currentThread().interrupt();
}
});
}
executor.shutdown();
}
}
三、信號量 —— Semaphore：控制併發訪問資源池
核心模型
- 許可證計數器
permits
acquire()：獲取許可（阻塞直到可用）release()：歸還許可- 公平/非公平模式（構造函數
fair 參數）
典型場景：數據庫連接池、限流器、對象池
public class SemaphoreDemo {
private static final Logger log = LoggerFactory.getLogger(SemaphoreDemo.class);
public static void main(String[] args) {
int permits = 2; // 只允許 2 個併發
Semaphore semaphore = new Semaphore(permits, true); // 公平模式
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 1; i <= 5; i++) {
final int id = i;
executor.submit(() -> {
try {
log.info("Worker-{} try acquire", id);
semaphore.acquire(); // 阻塞等待許可
log.info("Worker-{} acquired, working...", id);
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
log.info("Worker-{} release", id);
semaphore.release(); // 必須在 finally 釋放
}
});
}
executor.shutdown();
}
}
輸出
Worker-1 try acquire
Worker-1 acquired, working...
Worker-2 try acquire
Worker-2 acquired, working...
Worker-3 try acquire
Worker-4 try acquire
Worker-5 try acquire
Worker-1 release
Worker-3 acquired, working...
Worker-2 release
Worker-4 acquired, working...
Worker-3 release
Worker-5 acquired, working...
Worker-4 release
Worker-5 release
四、阻塞隊列 —— BlockingQueue：生產者-消費者模式標準件
核心特性
| 隊列 | 特點 | 適用場景 |
|---|
ArrayBlockingQueue | 有界、數組、公平鎖可選 | 固定緩衝區、高吞吐 |
LinkedBlockingQueue | 可選有界、鏈表、兩把鎖 | 通用生產消費、吞吐優先 |
PriorityBlockingQueue | 無界、優先級、非阻塞入隊 | 任務調度、延遲隊列 |
SynchronousQueue | 零容量、直接移交 | 線程池 CachedThreadPool、移交場景 |
DelayQueue | 無界、延遲取出 | 定時任務、緩存過期清理 |
關鍵方法對照
| 操作 | 拋異常 | 返回布爾 | 阻塞 | 超時阻塞 |
|---|
| 入隊 | add(e) | offer(e) | put(e) | offer(e, t, u) |
| 出隊 | remove() | poll() | take() | poll(t, u) |
| 查看 | element() | peek() | — | — |
經典生產者-消費者 Demo
public class ProducerConsumerDemo {
private static final BlockingQueue queue = new LinkedBlockingQueue<>(10);
private static final Logger log = LoggerFactory.getLogger(ProducerConsumerDemo.class);
static class Producer implements Runnable {
@Override public void run() {
try {
for (int i = 0; i < 20; i++) {
String msg = "Msg-" + i;
queue.put(msg); // 隊列滿阻塞
log.info("Produced: {}", msg);
Thread.sleep(100);
}
} catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
}
static class Consumer implements Runnable {
@Override public void run() {
try {
while (true) {
String msg = queue.take(); // 隊列空阻塞
log.info("Consumed: {}", msg);
Thread.sleep(150);
}
} catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
}
public static void main(String[] args) {
new Thread(new Producer(), "Producer").start();
for (int i = 1; i <= 3; i++) {
new Thread(new Consumer(), "Consumer-" + i).start();
}
}
}
五、避坑指南與最佳實踐
- CountDownLatch 不可重用：需重置請用
CyclicBarrier 或重新 new
- CyclicBarrier 異常傳染：任一線程異常導致所有
await() 拋 BrokenBarrierException，建議在任務內捕獲異常並決定是否 barrier.reset()
- Semaphore 許可證洩漏：
acquire() 後**必須在 finally 塊** release()
- BlockingQueue 無界風險：
LinkedBlockingQueue() 默認 Integer.MAX_VALUE，生產快消費慢會 OOM，**生產環境務必指定容量**
- 公平性選擇：高併發下非公平吞吐更高，公平防止餓死，根據業務 SLA 選擇
- 優先用高層抽象：
CompletableFuture、parallelStream、ExecutorService 往往比手動協調更簡潔
六、總結對比
| 工具類 | 核心隱喻 | 典型場景 | 可重用 | 攔截動作 |
|---|
| CountDownLatch | 倒數閂 | 主線程等多任務啟動/關閉 | ❌ | 無 |
| CyclicBarrier | 柵欄/集合點 | 多線程分階段同步、MapReduce | ✅ | ✅ 最後到達執行 |
| Semaphore | 許可證/信號燈 | 資源池、限流、併發控制 | ✅ | 無 |
| BlockingQueue | 緩衝區/管道 | 生產消費、任務調度、線程間傳遞 | ✅ | 無 |
掌握四大工具類的**適用邊界**與**組合用法**（如：Semaphore 保護有界隊列 + CountDownLatch 等待關閉），是編寫高質量併發代碼的關鍵。
