條件佇列是什麼？可能很多人和我一樣答不出來，不過今天終於搞清楚了！

什麼是條件佇列

條件佇列：當某個執行緒呼叫了wait方法，或者通過Condition物件呼叫了await相關方法，執行緒就會進入阻塞狀態，並加入到對應條件佇列中。

在等待喚醒機制相關文章中我們提到了條件佇列，即當物件獲取到同步鎖之後，如果呼叫了wait方法，當前執行緒會進入到條件佇列中，並釋放鎖。

synchronized(物件){ // 獲取鎖失敗，執行緒會加入到同步佇列中 
    while(條件不滿足){
        物件.wait();// 呼叫wait方法當前執行緒加入到條件佇列中
    }
}

基於synchcronized的內建條件佇列存在一些缺陷。每個內建鎖都只能有一個相關聯的條件佇列，因而存在多個執行緒可能在同一個條件佇列上等待不同的條件謂詞，並且在最常見的加鎖模式下公開條件佇列物件。

Java中的鎖的實現可以分為兩種，一種是基於synchronized的隱式鎖，它是基於JVM層面實現的；而另一種則是基於AQS框架在程式碼層面實現的鎖，如ReentrantLock等，在進行併發控制過程中，很多情況下他們都可以相互替代。

其中同步佇列和條件佇列是AQS中兩個比較核心的概念，它們是程式碼層面實現鎖的關鍵。關於同步佇列的內容，我們已經在圖解AQS的設計與實現，手摸手帶你實現一把互斥鎖！中進行了詳細的介紹。

與Object配合synchronized相比，基於AQS的Lock&Condition實現的等待喚醒模式更加靈活，支援多個條件佇列，支援等待狀態中不響應中斷以及超時等待功能； 其次就是基於AQS實現的條件佇列是"肉眼可見"的，我們可以通過原始碼進行debug，而synchronized則是完全隱式的。

同步佇列和條件佇列

與條件佇列密不可分的類則是ConditionObject， 是AQS中實現了Condition介面的內部類，通常配合基於AQS實現的鎖一同使用。當執行緒獲取到鎖之後，可以呼叫await方法進入條件佇列並釋放鎖，或者呼叫singinal方法喚醒對應條件佇列中等待時間最久的執行緒並加入到等待佇列中。

在AQS中，執行緒會被封裝成Node物件加入佇列中，而條件佇列中則複用了同步佇列中的Node物件。

Condition相關方法和描述

Condition介面一共定義了以下幾個方法：

await(): 當前執行緒進入等待狀態，直到被通知(siginal)或中斷【和wait方法語義相同】。

awaitUninterruptibly(): 當前執行緒進入等待狀態，直到被通知，對中斷不敏感。

awaitNanos(long timeout): 當前執行緒進入等待狀態直到被通知(siginal)，中斷或超時。

awaitUnitil(Date deadTime): 當前執行緒進入等待狀態直到被通知(siginal)，中斷或到達某個時間。

signal(): 喚醒一個等待在Condition上的執行緒，該執行緒從等待方法返回前必須獲得與Condition關聯的鎖【和notify方法語義相同】

signalAll(): 喚醒所有等待在Condition上的執行緒，能夠從等待方法返回的執行緒必須獲得與Condition關聯的鎖【和notifyAll方法語義相同】。

條件佇列入隊操作

當執行緒獲取到鎖之後，Condition物件呼叫await相關的方法，執行緒會進入到對應的條件佇列中。

/**
  * 如果當前執行緒被終端，丟擲 InterruptedException 異常
  */
public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 添加當前執行緒到【條件佇列】
    Node node = addConditionWaiter();
    // 釋放已經獲取的鎖資源，並返回釋放前的同步狀態
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 如果當前節點不在【同步佇列】中， 執行緒進入阻塞狀態，等待被喚醒
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

條件隊出隊操作

Condition物件呼叫signal或者signalAll方法時，

/**
 * 將【條件佇列】中第一個有效的元素移除並且新增到【同步佇列】中
 * 所謂有效指的是非null，並且狀態嗎
 * @param first 條件佇列中第一個非空的元素
 */
private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    // 將條件佇列中等待最久的那個有效元素新增到同步佇列中
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

/**
 * 將條件佇列中的節點轉換到同步佇列中
 */
final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     * 如果節點的等待狀態不能被修改，說明當前執行緒已經被取消等待【多個執行緒執行siginal時會出現的情況】
     */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    /*
     * 加入到【同步佇列】中，並且嘗試將前驅節點設定為可喚醒狀態
     */
    Node p = enq(node); // 將node新增到同步佇列中，並返回它的前驅節點
    int ws = p.waitStatus;
    // 如果前驅節點不需要喚醒，或者設定狀態為‘喚醒’失敗，則喚醒執行緒時期重新爭奪同步狀態
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

實現阻塞佇列

1. 自定義互斥鎖升級，增加獲取Condition物件介面。

/**
 * 自定義互斥鎖
 *
 * @author cruder
 * @time 2019/11/29 9:43
 */
public class MutexLock {

    private static final Sync STATE_HOLDER = new Sync();

    /**
     * 通過Sync內部類來持有同步狀態， 當狀態為1表示鎖被持有，0表示鎖處於空閒狀態
     */
    private static class Sync extends AbstractQueuedSynchronizer {

        /**
         * 是否被獨佔， 有兩種表示方式
         *  1. 可以根據狀態，state=1表示鎖被佔用，0表示空閒
         *  2. 可以根據當前獨佔鎖的執行緒來判斷，即getExclusiveOwnerThread()!=null 表示被獨佔
         */
        @Override
        protected boolean isHeldExclusively() {
            return getExclusiveOwnerThread() != null;
        }

        /**
         * 嘗試獲取鎖，將狀態從0修改為1，操作成功則將當前執行緒設定為當前獨佔鎖的執行緒
         */
        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        /**
         * 釋放鎖，將狀態修改為0
         */
        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0) {
                throw new UnsupportedOperationException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        //【新增程式碼】
        final ConditionObject newCondition() {
            return new ConditionObject();
        }
    }

    /**
     * 下面的實現Lock介面需要重寫的方法，基本是就是呼叫內部內Sync的方法
     */
    public void lock() {
        STATE_HOLDER.acquire(1);
    }

    public void unlock() {
        STATE_HOLDER.release(1);
    }

    // 【新增程式碼】 獲取條件佇列
    public Condition newCondition(){
        return STATE_HOLDER.newCondition();
    }

}

1. 基於自定義互斥鎖，實現阻塞佇列。阻塞佇列具有兩個特點：

• 新增元素到佇列中, 如果佇列已滿會使得當前執行緒阻塞【加入到條件佇列-佇列不滿】，直到佇列不滿為止
• 移除佇列中的元素，當佇列為空時會使當前執行緒阻塞【加入到條件佇列-佇列不空】，直到佇列不為空為止

/**
 *  有界阻塞阻塞佇列
 *
 * @author Jann Lee
 * @date 2019-12-11 22:20
 **/
public class BoundedBlockingQueue<T> {
    /**
     * list作為底層儲存結構
     */
    private List<T> dataList;
    /**
     * 佇列的大小
     */
    private int size;

    /**
     * 鎖，和條件變數
     */
    private MutexLock lock;
    /**
     * 佇列非空 條件變數
     */
    private Condition notEmpty;
    /**
     * 佇列未滿 條件變數
     */
    private Condition notFull;

    public BoundedBlockingQueue(int size) {
        dataList = new ArrayList<>();
        lock = new MutexLock();
        notEmpty = lock.newCondition();
        notFull = lock.newCondition();
        this.size = size;
    }


    /**
     * 佇列中新增元素 [只有佇列未滿時才可以新增，否則需要等待佇列變成未滿狀態]
     */
    public void add(T data) throws InterruptedException {
        lock.lock();
        try {
            // 如果佇列已經滿了， 需要在等待“佇列未滿”條件滿足
            while (dataList.size() == size) {
                notFull.await();
            }
            dataList.add(data);
            Thread.sleep(2000);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 移除佇列的第一個元素[只有佇列非空才可以移除，否則需要等待變成佇列非空狀態]
     */
    public T remove() throws InterruptedException {
        lock.lock();
        try {
            // 如果為空， 需要在等待“佇列非空”條件滿足
            while (dataList.isEmpty()) {
                notEmpty.await();
            }
            T result = dataList.remove(0);
            notFull.signal();
            return result;
        } finally {
            lock.unlock();
        }
    }

}

總結

1. 條件佇列和同步佇列在Java中有兩種實現，synchronized關鍵字以及基於AQS
2. 每個（基於synchronized的）內建鎖都只能有一個相關聯的條件佇列，會存在多個執行緒可能在同一個條件佇列上等待不同的條件謂詞；而（基於AQS實現的）顯式鎖支援多個條件佇列
3. 與wait，notify，notifyAll 對應的方法時Conditoin介面中的await，signal，signalAll，他們具有相同的語義

最後敬上一個關注了也沒有錯的公眾號~

相關推薦