您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 並發編程：Condition 條件變量深度解析與生產者消費者模式實戰

編輯：Java 相關知識

Java 並發核心組件：Condition 條件變量完整指南

java.util.concurrent.locks.Condition 是 JDK 1.5 引入的進階線程協調機制，提供比 Object.wait/notify 更精確、更靈活的線程等待/通知能力。本文系統對比兩種機制，詳細講解 Condition API、基本用法、多條件變量精確控制的經典生產者-消費者模式實現。

Condition vs Object 監視器方法對比

功能	Object 監視器	Condition
等待	wait()	await()
超時等待	wait(timeout)	await(time, unit) / awaitNanos / awaitUntil
不可中斷等待	無	awaitUninterruptibly()
單線程喚醒	notify()	signal()
全線程喚醒	notifyAll()	signalAll()
綁定鎖	synchronized (內置鎖)	Lock 實現（ReentrantLock 等）
多條件支持	無（單一等待隊列）	支持多 Condition 實例

Condition 核心 API

void await()                    // 等待直到被喚醒/中斷
boolean await(long time, TimeUnit unit)  // 超時等待，返回是否超時
long awaitNanos(long nanosTimeout)       // 納秒級超時，返回剩餘時間
void awaitUninterruptibly()              // 忽略中斷等待
boolean awaitUntil(Date deadline)        // 截止時間等待
void signal()                     // 喚醒一個等待線程
void signalAll()                  // 喚醒所有等待線程

基礎用法：替換 Object.wait/notify

Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();

// 等待線程
lock.lock();
try {
    condition.await();  // 釋放鎖並等待
} finally {
    lock.unlock();
}

// 喚醒線程
lock.lock();
try {
    condition.signal();  // 喚醒一個等待線程
} finally {
    lock.unlock();
}

關鍵差異：Condition 必須在 持有鎖狀態下 調用 await/signal，否則拋出 IllegalMonitorStateException。

進階場景：有界緩衝區——多條件變量精確控制

經典問題：單一 notifyAll() 無法區分喚醒「生產者」還是「消費者」，導致虛假喚醒、效率低下。

BoundedBuffer 實現

class BoundedBuffer {
    final Lock lock = new ReentrantLock();
    final Condition notFull = lock.newCondition();   // 控制生產者
    final Condition notEmpty = lock.newCondition();  // 控制消費者
    
    final Object[] items = new Object[5];
    int putptr, takeptr, count;

    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length) notFull.await();  // 滿則等待
            items[putptr] = x;
            if (++putptr == items.length) putptr = 0;
            ++count;
            notEmpty.signal();  // 喚醒消費者
            System.out.println(Thread.currentThread().getName() + " put " + x);
        } finally { lock.unlock(); }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) notEmpty.await();  // 空則等待
            Object x = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            --count;
            notFull.signal();  // 喚醒生產者
            System.out.println(Thread.currentThread().getName() + " take " + x);
            return x;
        } finally { lock.unlock(); }
    }
}

測試驗證

BoundedBuffer bb = new BoundedBuffer();
for (int i = 0; i < 10; i++) {
    new Thread(() -> { bb.put(i); }, "P" + i).start();
    new Thread(() -> { bb.take(); }, "C" + i).start();
}