一、Condition的概念

介紹

回憶 synchronized 關(guān)鍵字，它配合 Object 的 wait()、notify() 系列方法可以實(shí)現(xiàn)等待/通知模式。

對(duì)于 Lock，通過(guò) Condition 也可以實(shí)現(xiàn)等待/通知模式。

Condition 是一個(gè)接口。
Condition 接口的實(shí)現(xiàn)類是 Lock（AQS）中的 ConditionObject。
Lock 接口中有個(gè) newCondition() 方法，通過(guò)這個(gè)方法可以獲得 Condition 對(duì)象（其實(shí)就是 ConditionObject）。
因此，通過(guò) Lock 對(duì)象可以獲得 Condition 對(duì)象。

Lock lock  = new ReentrantLock();
Condition c1 = lock.newCondition();
Condition c2 = lock.newCondition();

二、Condition的實(shí)現(xiàn)分析

實(shí)現(xiàn)

ConditionObject 類是 AQS 的內(nèi)部類，實(shí)現(xiàn)了 Condition 接口。

public class ConditionObject implements Condition, java.io.Serializable {
        private transient Node firstWaiter;
        private transient Node lastWaiter;
        ...

可以看到，等待隊(duì)列和同步隊(duì)列一樣，使用的都是同步器 AQS 中的節(jié)點(diǎn)類 Node。
同樣擁有首節(jié)點(diǎn)和尾節(jié)點(diǎn)，
每個(gè) Condition 對(duì)象都包含著一個(gè) FIFO 隊(duì)列。
結(jié)構(gòu)圖：

等待

調(diào)用 Condition 的 await() 方法會(huì)使線程進(jìn)入等待隊(duì)列，并釋放鎖，線程狀態(tài)變?yōu)榈却隣顟B(tài)。

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    //釋放同步狀態(tài)（鎖）
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    //判斷節(jié)點(diǎn)是否放入同步對(duì)列
    while (!isOnSyncQueue(node)) {
        //阻塞
        LockSupport.park(this);
        //如果已經(jīng)中斷了，則退出
        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);
}

分析上述方法的大概過(guò)程：

1. 將當(dāng)前線程創(chuàng)建為節(jié)點(diǎn)，加入等待隊(duì)列；
2. 釋放鎖，喚醒同步隊(duì)列中的后繼節(jié)點(diǎn)；
3. while循環(huán)判斷節(jié)點(diǎn)是否放入同步隊(duì)列：

• 沒(méi)有放入，則阻塞，繼續(xù) while 循環(huán)（如果已經(jīng)中斷了，則退出）
• 放入，則退出 while 循環(huán)，執(zhí)行后面的判斷

1. 退出 while 說(shuō)明節(jié)點(diǎn)已經(jīng)在同步隊(duì)列中，調(diào)用 acquireQueued() 方法加入同步狀態(tài)競(jìng)爭(zhēng)。
2. 競(jìng)爭(zhēng)到鎖后從 await() 方法返回，即退出該方法。

addConditionWaiter() 方法：

private Node addConditionWaiter() {
    Node t = lastWaiter;
    if (t != null && t.waitStatus != Node.CONDITION) {
        //清除條件隊(duì)列中所有狀態(tài)不為Condition的節(jié)點(diǎn)
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    //將該線程創(chuàng)建節(jié)點(diǎn)，放入等待隊(duì)列
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

過(guò)程分析：同步隊(duì)列的首節(jié)點(diǎn)移動(dòng)到等待隊(duì)列。加入尾節(jié)點(diǎn)之前會(huì)清除所有狀態(tài)不為 Condition 的節(jié)點(diǎn)。

通知

調(diào)用 Condition 的 signal() 方法，可以喚醒等待隊(duì)列的首節(jié)點(diǎn)（等待時(shí)間最長(zhǎng)），喚醒之前會(huì)將該節(jié)點(diǎn)移動(dòng)到同步隊(duì)列中。

public final void signal() {
    //判斷是否獲取了鎖
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

過(guò)程：

1. 先判斷當(dāng)前線程是否獲取了鎖；
2. 然后對(duì)首節(jié)點(diǎn)調(diào)用 doSignal() 方法。

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
       (first = firstWaiter) != null);
}

過(guò)程：

1. 修改首節(jié)點(diǎn)；
2. 調(diào)用 transferForSignal() 方法將節(jié)點(diǎn)移動(dòng)到同步隊(duì)列。

final boolean transferForSignal(Node node) {
    //將節(jié)點(diǎn)狀態(tài)變?yōu)?   
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
    //將該節(jié)點(diǎn)加入同步隊(duì)列
    Node p = enq(node);
    int ws = p.waitStatus;
    //如果結(jié)點(diǎn)p的狀態(tài)為cancel 或者修改waitStatus失敗，則直接喚醒
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

調(diào)用同步器的 enq 方法，將節(jié)點(diǎn)移動(dòng)到同步隊(duì)列，
滿足條件后使用 LockSupport 喚醒該線程。

當(dāng) Condition 調(diào)用 signalAll() 方法：

public final void signalAll() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignalAll(first);
}
private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}

可以看到 doSignalAll() 方法使用了 do-while 循環(huán)來(lái)喚醒每一個(gè)等待隊(duì)列中的節(jié)點(diǎn)，直到 first 為 null 時(shí)，停止循環(huán)。

一句話總結(jié) signalAll() 的作用：將等待隊(duì)列中的全部節(jié)點(diǎn)移動(dòng)到同步隊(duì)列中，并喚醒每個(gè)節(jié)點(diǎn)的線程。

總結(jié)

整個(gè)過(guò)程可以分為三步：

第一步：一個(gè)線程獲取鎖后，通過(guò)調(diào)用 Condition 的 await() 方法，會(huì)將當(dāng)前線程先加入到等待隊(duì)列中，并釋放鎖。然后就在 await() 中的一個(gè) while 循環(huán)中判斷節(jié)點(diǎn)是否已經(jīng)在同步隊(duì)列，是則嘗試獲取鎖，否則一直阻塞。

第二步：當(dāng)線程調(diào)用 signal() 方法后，程序首先檢查當(dāng)前線程是否獲取了鎖，然后通過(guò) doSignal(Node first) 方法將節(jié)點(diǎn)移動(dòng)到同步隊(duì)列，并喚醒節(jié)點(diǎn)中的線程。

第三步：被喚醒的線程，將從 await() 中的 while 循環(huán)中退出來(lái)，然后調(diào)用 acquireQueued() 方法競(jìng)爭(zhēng)同步狀態(tài)。競(jìng)爭(zhēng)成功則退出 await() 方法，繼續(xù)執(zhí)行。

標(biāo)簽：

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請(qǐng)聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請(qǐng)與原作者聯(lián)系。