併發編程的實現原理-Condition-筆記

Condition（java.util.concurrent.locks.Condition）

• 任意一個Java對象，都擁有一組監視器方法（定義在java.lang.Object上），
  • 主要包括wait()、notify()以及notifyAll()方法，
  • 這些方法與synchronized同步關鍵字配合，能夠實現等待/通知模式
• JUC包提供了Condition來對鎖進行精準控制，
  • Condition是一個多線程協調通訊的工具類，
  • 可讓某些線程一塊兒等待某個條件（condition），
  • 只有知足條件時，線程纔會被喚醒。

condition使用案例

• ConditionWait

  • public class ConditionDemoWait implements Runnable{
        private Lock lock;
        private Condition condition;
        public ConditionDemoWait(Lock lock, Condition condition){
            this.lock=lock;
            this.condition=condition;
        }
        @Override
        public void run() {
            System.out.println("begin -ConditionDemoWait");
            try {
                lock.lock();  ////得到鎖
                condition.await();/////// 這裏會添加到condition 隊列，後續unlock釋放鎖，
                                  //////signal 會去condition隊列喚醒他，把他放回AQS隊列
                System.out.println("end - ConditionDemoWait");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                lock.unlock();///釋放鎖
            }
        }
    }

     

• ConditionSignal

  • public class ConditionDemoSignal implements Runnable {
        private Lock lock;
        private Condition condition;
    
        public ConditionDemoSignal(Lock lock, Condition condition) {
            this.lock = lock;
            this.condition = condition;
        }
    
        @Override
        public void run() {
            System.out.println("begin -ConditionDemoSignal");
            try {
                lock.lock();   //////// 獲取鎖
                condition.signal();//////// 喚醒ConditionDemoWait 線程，加到AQS隊列，使得await 後面代碼獲得執行
                System.out.println("end - ConditionDemoSignal");
            } finally {
                lock.unlock();/////////釋放鎖
            }
        }
    }

     

• 經過這個案例簡單實現了wait和notify的功能，
  • 當調用await方法後，當前線程會釋放鎖並等待，
  • 而其餘線程調用condition對象的signal或者signalall方法通知並被阻塞的線程，
  • 而後本身執行unlock釋放鎖，
  • 被喚醒的線程得到以前的鎖繼續執行，最後釋放鎖。
• 因此，condition中兩個最重要的方法，一個是await，一個是signal方法
  • await:把當前線程阻塞掛起
  • signal:喚醒阻塞的線程

await方法

• 調用Condition的await()方法（或者以await開頭的方法），
  • 會使當前線程進入等待隊列
  • 並釋放鎖，
  • 同時線程狀態變爲等待狀態。
• 當從await()方法返回時，
  • 當前線程必定獲取了Condition相關聯的鎖

public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        Node node = addConditionWaiter(); //建立一個新的節點，節點狀態爲condition，採用的數據結構仍然是鏈表
        int savedState = fullyRelease(node); //釋放當前的鎖，獲得鎖的狀態，並喚醒AQS隊列中的一個線程
        int interruptMode = 0;
//若是當前節點沒有在同步隊列上，即尚未被signal，則將當前線程阻塞
//isOnSyncQueue 判斷當前 node 狀態,
// 若是是 CONDITION 狀態,或者不在隊列上了,就繼續阻塞,
// 還在隊列上且不是 CONDITION 狀態了,就結束循環和阻塞
        while (!isOnSyncQueue(node)) {//第一次判斷的是false，由於前面已經釋放鎖了
            LockSupport.park(this); // 第一次老是 park 本身,開始阻塞等待
// 線程判斷本身在等待過程當中是否被中斷了,
//若是沒有中斷,則再次循環,會在 isOnSyncQueue 中判斷本身是否在隊列上.
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
// 當這個線程醒來,會嘗試拿鎖, 當 acquireQueued 返回 false 就是拿到鎖了.
// interruptMode != THROW_IE -> 表示這個線程沒有成功將 node 入隊,但 signal 執行了 enq 方法讓其入隊了.
// 將這個變量設置成 REINTERRUPT.
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
// 若是 node 的下一個等待者不是 null, 則進行清理,清理 Condition 隊列上的節點.
        // 若是是 null ,就沒有什麼好清理的了.
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
// 若是線程被中斷了,須要拋出異常.或者什麼都不作
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

signal

• 調用Condition的signal()方法，
  • 將會喚醒在等待隊列中等待時間最長的節點（首節點），
  • 在喚醒節點以前，會將節點移到同步隊列中

public final void signal() {
        if (!isHeldExclusively()) //先判斷當前線程是否得到了鎖
            throw new IllegalMonitorStateException();
        Node first = firstWaiter; // 拿到 Condition 隊列上第一個節點
        if (first != null)
            doSignal(first);
    }
/** ######################################### */
    private void doSignal(Node first) {
        do {
            if ( (firstWaiter = first.nextWaiter) == null)// 若是第一個節點的下一個節點是 null,
                                                          //那麼, 最後一個節點也是 null.
                    lastWaiter = null; // 將 next 節點設置成 null
            first.nextWaiter = null;
        } while (!transferForSignal(first) &&
                (first = firstWaiter) != null);
    }

• 該方法先是 CAS 修改了節點狀態，
  • 若是成功，就將這個節點放到 AQS 隊列中，
  • 而後喚醒這個節點上的線程。
  • 此時，那個節點就會在 await 方法中甦醒

final boolean transferForSignal(Node node) {
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        Node p = enq(node);
        int ws = p.waitStatus;
        // 若是上一個節點的狀態被取消了, 
        // 或者嘗試設置上一個節點的狀態爲 SIGNAL 失敗了(SIGNAL 表示: 他的
        next 節點須要中止阻塞),
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread); // 喚醒輸入節點上的線程.
        return true;
    }

我的理解：

• 一共是倆隊列，AQS隊列、Condition隊列