Java 並發編程：自定義同步器構建指南——從 AQS 到 Condition 實戰

為什麼需要自定義同步器

當 JDK 內建同步工具（synchronized、ReentrantLock、Semaphore、CountDownLatch 等）無法滿足特定同步語義時，需基於 AbstractQueuedSynchronizer (AQS) 或 Condition 構建自定義同步器。

狀態依賴的通用構造模式

acquire lock on object state;        // 獲取鎖
while (precondition does not hold) {  // 條件不滿足
    release lock;                     // 釋放鎖
    wait until precondition might hold; // 等待條件
    optionally fail if interrupted or timeout; // 中斷/超時處理
    reacquire lock;                   // 重新獲取鎖
}
perform action;                       // 執行動作
release lock;                         // 釋放鎖

有界緩衝區基類實現

public class BaseBoundedBuffer<V> {
    private final V[] buf;
    private int tail, head, count;
    
    @SuppressWarnings("unchecked")
    public BaseBoundedBuffer(int capacity) {
        this.buf = (V[]) new Object[capacity];
    }
    
    // 核心操作（已同步）
    public synchronized void doPut(V v) {
        buf[tail] = v;
        if (++tail == buf.length) tail = 0;
        count++;
    }
    
    public synchronized V doTake() {
        V v = buf[head];
        if (++head == buf.length) head = 0;
        count--;
        return v;
    }
    
    public final synchronized boolean isFull() { return count == buf.length; }
    public final synchronized boolean isEmpty() { return count == 0; }
}

三種阻塞實現方式對比

方式一：拋出異常（不推薦）

public synchronized void put1(V v) throws Exception {
    if (isFull()) throw new Exception("Buffer full");
    doPut(v);
}

缺點：異常用於控制流、調用者需處理失敗、未解決等待問題。

方式二：輪詢 + 休眠（不推薦）

public void put2(V v) throws InterruptedException {
    while (true) {
        synchronized (this) {
            if (!isFull()) { doPut(v); return; }
        }
        Thread.sleep(SLEEP_TIME); // 難以調優：太短佔 CPU，太長響應慢
    }
}

方式三：Condition 隊列（推薦）

public synchronized void put3(V v) throws InterruptedException {
    while (isFull()) wait(); // 釋放鎖並進入等待隊列
    doPut(v);
    notifyAll(); // 喚醒所有等待線程
}

優點：高效阻塞、自動釋放鎖、JVM 級調度、響應及時。

Condition 隊列使用規範（關鍵）

1. 條件謂詞：操作前提條件（如「緩衝區不滿」），由狀態變量組成
2. 三元關係：鎖、wait/notify、條件謂詞 —— **必須同一對象**
3. 標準模式：

   
   void stateDependentMethod() throws InterruptedException {
       synchronized(lock) {
           while (!conditionPredicate) lock.wait(); // 必須用 while！
           // 執行受保護動作
       }
       // 執行完畢後通知
       synchronized(lock) { lock.notifyAll(); }
   }
   

4. notifyAll 優於 notify：避免喚醒錯誤線程導致信號丟失（經典「信號丟失」問題）

顯式 Condition 對象（java.util.concurrent.locks）

比內置監視器更靈活：每個 Lock 可有多個 Condition、可中斷等待、超時等待、公平/非公平隊列。

有界緩衝區完整實現（雙 Condition）

public class ConditionBoundedBuffer<V> {
    private final V[] buf;
    private int tail, head, count;
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    
    @SuppressWarnings("unchecked")
    public ConditionBoundedBuffer(int capacity) {
        this.buf = (V[]) new Object[capacity];
    }
    
    public void put(V v) throws InterruptedException {
        lock.lock();
        try {
            while (count == buf.length) notFull.await(); // 等待不滿
            buf[tail] = v;
            if (++tail == buf.length) tail = 0;
            count++;
            notEmpty.signal(); // 喚醒消費者
        } finally {
            lock.unlock();
        }
    }
    
    public V take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) notEmpty.await(); // 等待不空
            V v = buf[head];
            buf[head] = null; // GC 友好
            if (++head == buf.length) head = 0;
            count--;
            notFull.signal(); // 喚醒生產者
            return v;
        } finally {
            lock.unlock();
        }
    }
}

AQS 框架核心概念

• 狀態：int 類型，語義由子類定義（如 ReentrantLock 為重入計數，Semaphore 為許可數）
• 同步隊列：CLH 變體雙向隊列，管理爭用線程
• 模板方法：tryAcquire/tryRelease/tryAcquireShared/tryReleaseShared —— 子類實現具體語義
• ConditionObject：每個 Condition 維護獨立等待隊列

自定義同步器開發清單

總結

優先使用 JDK 內建工具；需自定義時，優先用 Lock + Condition 組合，複雜同步語義再考慮繼承 AQS。核心原則：**條件等待必須在 while 循環中、鎖保護狀態變量、notifyAll 防信號丟失**。