您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

Java 並發包 java.util.concurrent 核心組件全景指南：從 Executor 到同步工具的實戰詳解

編輯：Java 綜合教學

前言：為什麼需要 concurrent 套件？

JDK 1.5 前，開發者需自行處理 wait/notify、synchronized、鎖排序、線程安全集合等底層細節，極易產生死鎖、活鎖、效能瓶頸。Doug Lea 主導的 java.util.concurrent (JUC) 提供高效、經實戰驗證的並發基建，讓開發者專注業務邏輯。

核心組件架構圖

java.util.concurrent
├── Executor 框架          → 任務提交與執行解耦
│   ├── Executor           → 單一方法 execute(Runnable)
│   ├── ExecutorService    → 生命週期管理、Future 提交
│   │   ├── ThreadPoolExecutor  → 核心線程池實現
│   │   ├── ScheduledThreadPoolExecutor → 定時/周期任務
│   │   └── ForkJoinPool       → 分治任務、工作竊取
│   └── CompletableFuture  → 非同步組合、回調地獄終結者
├── 同步器                  → 協調線程協作
│   ├── Semaphore          → 計數信號量、資源池隔離
│   ├── CountDownLatch     → 一次性倒數閘、啟動/完成等待
│   ├── CyclicBarrier      → 可重用屏障、分階段同步
│   ├── Phaser             → 多階段、動態註冊同步器 (JDK 7+)
│   └── Exchanger          → 雙方資料交換、管道模式
├── 鎖與條件                → 顯式鎖、比 synchronized 更強
│   ├── ReentrantLock      → 可重入、公平/非公平、可中斷、超時
│   ├── ReentrantReadWriteLock → 讀寫分離、降級
│   ├── StampedLock        → 樂觀讀、無寫入時無鎖 (JDK 8+)
│   └── Condition          → 多條件變量、精準通知
├── 線程安全集合            → 高並發容器
│   ├── ConcurrentHashMap  → 分段鎖/CAS、弱一致迭代器
│   ├── ConcurrentLinkedQueue → 無鎖隊列、非阻塞
│   ├── BlockingQueue 介面 → 阻塞隊列族群
│   │   ├── ArrayBlockingQueue  → 有界、陣列
│   │   ├── LinkedBlockingQueue → 可選有界、鏈表
│   │   ├── PriorityBlockingQueue → 優先級
│   │   ├── DelayQueue          → 延遲取出
│   │   └── SynchronousQueue    → 0 容量、直接移交
│   └── CopyOnWriteArrayList/Set → 讀多寫少、快照迭代
└── 工具類
    ├── ThreadLocalRandom  → 線程本地隨機數、無競爭
    ├── LongAdder/DoubleAdder → 高並發計數器、優於 AtomicLong
    └── CompletableFuture  → 非同步編排核心

重點組件深度解析與實戰代碼

1. Executor 框架：線程池最佳實踐

三大工廠方法對比

方法	特性	適用場景	風險
newFixedThreadPool(n)	核心=最大=n、無界隊列	長期穩定負載、任務耗時可預期	隊列堆積導致 OOM
newCachedThreadPool()	0 核心、無界線程、60s 回收	大量短任務、突發流量	線程數失控、OOM
newSingleThreadExecutor()	單線程、串行保證	順序執行、事件循環	單點阻塞、無法並行

生產環境建議：自定義 ThreadPoolExecutor

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    8,                      // corePoolSize
    16,                     // maximumPoolSize
    60L, TimeUnit.SECONDS,  // keepAliveTime
    new LinkedBlockingQueue<>(1024), // 有界隊列，防 OOM
    new ThreadFactoryBuilder().setNameFormat("biz-worker-%d").build(),
    new ThreadPoolExecutor.AbortPolicy() // 或 CallerRunsPolicy 緩衝
);

有界隊列 + 拒絕策略 是防禦流量洪峰的關鍵。
線程命名便於日誌追蹤與 Arthas 診斷。
監控指標：activeCount、queue.size()、completedTaskCount。

2. Semaphore：資源池隔離與限流

// 資料庫連接池模擬：最多 10 條連線
Semaphore dbPermits = new Semaphore(10, true); // 公平模式

public Connection borrow() throws InterruptedException {
    dbPermits.acquire(); // 阻塞直到拿到許可
    return dataSource.getConnection();
}

public void release(Connection conn) {
    try { DataSourceUtils.releaseConnection(conn, dataSource); }
    finally { dbPermits.release(); } // 歸還許可
}

acquireUninterruptibly() 忽略中斷，配合 try-finally 確保釋放。
可實現 滑動窗口限流：定時 release(n) 補充許可。

3. ReentrantLock：顯式鎖完全掌握

private final ReentrantLock lock = new ReentrantLock(true); // 公平鎖

public void transfer(Account from, Account to, long amount) {
    boolean locked = false;
    try {
        locked = lock.tryLock(500, TimeUnit.MILLISECONDS); // 超時獲取
        if (!locked) throw new RuntimeException("鎖獲取超時");
        // 業務邏輯
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        if (locked) lock.unlock();
    }
}

核心優勢

可中斷：lockInterruptibly() 響應中斷，避免死鎖無法恢復。
超時獲取：tryLock(timeout) 防止無限等待。
公平/非公平：公平鎖吞吐低但無飢餓；非公平（預設）吞吐高。
多 Condition：lock.newCondition() 實現多隊列精準通知（如生產者/消費者分開等待）。

4. BlockingQueue：生產者-消費者標準模板

// 有界隊列，生產者滿時阻塞，消費者空時阻塞
BlockingQueue<Task> queue = new LinkedBlockingQueue<>(100);

// 生產者
public void produce(Task t) throws InterruptedException {
    queue.put(t); // 滿則阻塞
}

// 消費者
public void consume() throws InterruptedException {
    while (running) {
        Task t = queue.take(); // 空則阻塞
        process(t);
    }
}

隊列選型指南

ArrayBlockingQueue：固定容量、低延遲、GC 友好。
LinkedBlockingQueue：預設無界（慎用！）、吞吐略高。
SynchronousQueue：零容量、直接移交、newCachedThreadPool 內部使用。
DelayQueue：定時任務、重試機制。

5. CountDownLatch vs CyclicBarrier vs Phaser

特性	CountDownLatch	CyclicBarrier	Phaser
可重用	否	是	是
動態註冊	否	否	是
階段同步	單階段	多階段 (barrier action)	多階段、彈性
典型場景	啟動等待、分散聚合	迭代計算、多階段流水線	動態任務樹、ForkJoin 輔助

CountDownLatch 範例：並行啟動服務

CountDownLatch latch = new CountDownLatch(3);
executor.submit(() -> { initDB(); latch.countDown(); });
executor.submit(() -> { initCache(); latch.countDown(); });
executor.submit(() -> { initMQ(); latch.countDown(); });
latch.await(); // 主線程等待全部就緒
startWebServer();

CyclicBarrier 範例：多階段迭代

CyclicBarrier barrier = new CyclicBarrier(3, () -> mergeResults()); // 最後一個觸發合併
for (int phase = 0; phase < 5; phase++) {
    executor.submit(() -> { computePhase(phase); barrier.await(); });
}

6. CompletableFuture：非同步編排瑞士刀

// 並行聚合
CompletableFuture<User> u = CompletableFuture.supplyAsync(() -> userService.get(uid), executor);
CompletableFuture<List<Order>> o = CompletableFuture.supplyAsync(() -> orderService.list(uid), executor);
CompletableFuture<Profile> p = u.thenCombine(o, (user, orders) -> buildProfile(user, orders));

// 超時兜底
Profile profile = p.orTimeout(2, TimeUnit.SECONDS)
                   .exceptionally(ex -> defaultProfile())
                   .join();