併發編程之 wait notify 方法剖析

前言

2018 元旦快樂。

摘要：

1. notify wait 如何使用？
2. 爲何必須在同步塊中？
3. 使用 notify wait 實現一個簡單的生產者消費者模型
4. 底層實現原理

1. notify wait 如何使用？

今天咱們要學習或者說分析的是 Object 類中的 wait notify 這兩個方法，其實說是兩個方法，這兩個方法包括他們的重載方法一共有5個，而Object 類中一共才 12 個方法，可見這2個方法的重要性。咱們先看看 JDK 中的代碼：

public final native void notify();

public final native void notifyAll();
 
public final void wait() throws InterruptedException {
    wait(0);
}

public final native void wait(long timeout) throws InterruptedException;

public final void wait(long timeout, int nanos) throws InterruptedException {
    if (timeout < 0) {
        throw new IllegalArgumentException("timeout value is negative");
    }

    if (nanos < 0 || nanos > 999999) {
        throw new IllegalArgumentException(
                            "nanosecond timeout value out of range");
    }

    if (nanos > 0) {
        timeout++;
    }

    wait(timeout);
}
複製代碼

就是這五個方法。其中有3個方法是 native 的，也就是由虛擬機本地的c代碼執行的。有2個 wait 重載方法最終仍是調用了 wait（long） 方法。

首先仍是 know how。來一個最簡單的例子，看看如何使用這兩個方法。

package cn.think.in.java.two;

import java.util.concurrent.TimeUnit;

public class WaitNotify {

  final static Object lock = new Object();

  public static void main(String[] args) {

    new Thread(new Runnable() {
      @Override
      public void run() {
        System.out.println("線程 A 等待拿鎖");
        synchronized (lock) {
          try {
            System.out.println("線程 A 拿到鎖了");
            TimeUnit.SECONDS.sleep(1);
            System.out.println("線程 A 開始等待並放棄鎖");
            lock.wait();
            System.out.println("被通知能夠繼續執行 則 繼續運行至結束");
          } catch (InterruptedException e) {
          }
        }
      }
    }, "線程 A").start();

    new Thread(new Runnable() {
      @Override
      public void run() {
        System.out.println("線程 B 等待鎖");
        synchronized (lock) {
          System.out.println("線程 B 拿到鎖了");
          try {
            TimeUnit.SECONDS.sleep(5);
          } catch (InterruptedException e) {
          }
          lock.notify();
          System.out.println("線程 B 隨機通知 Lock 對象的某個線程");
        }
      }
    }, "線程 B").start();
  }


}

複製代碼

運行結果：

線程 A 等待拿鎖 線程 B 等待鎖 線程 A 拿到鎖了 線程 A 開始等待並放棄鎖 線程 B 拿到鎖了 線程 B 隨機通知 Lock 對象的某個線程 被通知能夠繼續執行 則 繼續運行至結束

在上面的代碼中，線程 A 和 B 都會搶這個 lock 對象的鎖，A 的運氣比較好（也可能使 B 拿到鎖），他先拿到了鎖，而後調用了 wait 方法，放棄了鎖，並掛起了本身，這個時候等待鎖的 B 就拿到了鎖，而後通知了A，可是請注意，通知完畢以後，B 線程並無執行完同步代碼塊中的代碼，所以，A 仍是拿不到鎖的，所以沒法運行，等到B線程執行完畢，出了同步塊，這個時候 A 線程才被激活得以繼續執行。

使用 wait 方法和 notify 方法可使 2 個無關的線程進行通訊。也就是面試題中常提到的線程之間如何通訊。

若是沒有 wait 方法和 noitfy 方法，咱們如何讓兩個線程通訊呢？簡單的辦法就是讓某個線程循環去檢查某個標記變量，好比：

while （value != flag） {
  Thread.sleep(1000);
}
doSomeing();
複製代碼

上面的這段代碼在條件不知足使就睡眠一段時間，這樣作到目的是防止過快的」無效嘗試「，這種方式看似可以實現所需的功能，可是卻存在以下問題：

1. 難以確保及時性。由於等待的1000時間會致使時間差。
2. 難以下降開銷，若是確保了及時性，休眠時間縮短，將大大消耗CPU。

可是有了Java 自帶的 wait 方法 和 notify 方法，一切迎刃而解。官方說法是等待/通知機制。一個線程在等待，另外一個線程能夠通知這個線程，實現了線程之間的通訊。

2. 爲何必須在同步塊中？

注意，這兩個方法的使用必須是在 synchroized 同步塊中，而且在當前對象的同步塊中，若是在 A 對象的方法中調用 B 對象的 wait 或者 notify 方法，虛擬機會拋出 IllegalMonitorStateException，非法的監視器異常，由於你這個線程持有的監視器和你調用的監視器的不是一個對象。

那麼爲何這兩個方法必定要在同步塊中呢？

這裏要說一個專業名詞：競態條件。什麼是競太條件呢？

當兩個線程競爭同一資源時，若是對資源的訪問順序敏感，就稱存在競態條件。

競態條件會致使程序在併發狀況下出現一些bugs。多線程對一些資源的競爭的時候就會產生競態條件，若是首先要執行的程序競爭失敗排到後面執行了，那麼整個程序就會出現一些不肯定的bugs。這種bugs很難發現並且會重複出現，這是由於線程間會隨機競爭。

假設有2個線程，分別是生產者和消費者，他們有各自的任務。

1.1生產者檢查條件（如緩存滿了）-> 1.2生產者必須等待 2.1消費者消費了一個單位的緩存 -> 2.2從新設置了條件（如緩存沒滿） -> 2.3調用notifyAll()喚醒生產者

咱們但願的順序是： 1.1->1.2->2.1->2.2->2.3 可是因爲CPU執行是隨機的，可能會致使 2.3 先執行，1.2 後執行，這樣就會致使生產者永遠也醒不過來了！

因此咱們必須對流程進行管理，也就是同步，經過在同步塊中並結合 wait 和 notify 方法，咱們能夠手動對線程的執行順序進行調整。

3. 使用 notify wait 實現一個簡單的生產者消費者模型

雖然不少書中都不建議咱們直接使用 notify 和 wait 方法進行併發編程，但仍然須要咱們重點掌握。樓主寫了一個簡單的生產者消費者例子：

簡單的緩存類：

public class Queue {

  final int num;
  final List<String> list;
  boolean isFull = false;
  boolean isEmpty = true;


  public Queue(int num) {
    this.num = num;
    this.list = new ArrayList<>();
  }


  public synchronized void put(String value) {
    try {
      if (isFull) {
        System.out.println("putThread 暫停了，讓出了鎖");
        this.wait();
        System.out.println("putThread 被喚醒了，拿到了鎖");
      }

      list.add(value);
      System.out.println("putThread 放入了" + value);
      if (list.size() >= num) {
        isFull = true;
      }
      if (isEmpty) {
        isEmpty = false;
        System.out.println("putThread 通知 getThread");
        this.notify();
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public synchronized String get(int index) {
    try {
      if (isEmpty) {
        System.err.println("getThread 暫停了，並讓出了鎖");
        this.wait();
        System.err.println("getThread 被喚醒了，拿到了鎖");
      }

      String value = list.get(index);
      System.err.println("getThread 獲取到了" + value);
      list.remove(index);

      Random random = new Random();
      int randomInt = random.nextInt(5);
      if (randomInt == 1) {
        System.err.println("隨機數等於1， 清空集合");
        list.clear();
      }

      if (getSize() < num) {
        if (getSize() == 0) {
          isEmpty = true;
        }
        if (isFull) {
          isFull = false;
          System.err.println("getThread 通知 putThread 能夠添加了");
          Thread.sleep(10);
          this.notify();
        }
      }
      return value;


    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return null;
  }


  public int getSize() {
    return list.size();
  }


複製代碼

生產者線程：

class PutThread implements Runnable {

  Queue queue;

  public PutThread(Queue queue) {
    this.queue = queue;
  }

  @Override
  public void run() {
    int i = 0;
    for (; ; ) {
      i++;
      queue.put(i + "號");

    }
  }
}

複製代碼

消費者線程：

class GetThread implements Runnable {

  Queue queue;

  public GetThread(Queue queue) {
    this.queue = queue;
  }

  @Override
  public void run() {
    for (; ; ) {
      for (int i = 0; i < queue.getSize(); i++) {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        String value = queue.get(i);

      }
    }
  }
}

複製代碼

你們有興趣能夠跑跑看，可以加深這兩個方法的理解，實際上，JDK 內部的阻塞隊列也是相似這種實現，可是，不是用的 synchronized ，而是使用的重入鎖。

基本上經典的生產者消費者模式的有着以下規則：

等待方遵循以下規則：

1. 獲取對象的鎖。
2. 若是條件不知足，那麼調用對象的 wait 方法，被通知後仍要檢查條件。
3. 條件知足則執行相應的邏輯。

對應的僞代碼入下：

synchroize( 對象 ){
    while(條件不知足){
      對象.wait();
    }
    對應的處理邏輯......
}
複製代碼

通知方遵循以下規則：

1. 得到對象的鎖。
2. 改變條件。
3. 通知全部等待在對象上的線程。

對應的僞代碼以下：

synchronized（對象）{
  改變條件
  對象.notifyAll();
}
複製代碼

4. 底層實現原理

知道了如何使用，就得知道他的原理究竟是什麼？

首先咱們看，使用這兩個方法的順序通常是什麼？

1. 使用 wait ，notify 和 notifyAll 時須要先對調用對象加鎖。
2. 調用 wait 方法後，線程狀態有 Running 變爲 Waiting，並將當前線程放置到對象的 等待隊列。
3. notify 或者 notifyAll 方法調用後， 等待線程依舊不會從 wait 返回，須要調用 noitfy 的線程釋放鎖以後，等待線程纔有機會從 wait 返回。
4. notify 方法將等待隊列的一個等待線程從等待隊列種移到同步隊列中，而 notifyAll 方法則是將等待隊列種全部的線程所有移到同步隊列，被移動的線程狀態由 Waiting 變爲 Blocked。
5. 從 wait 方法返回的前提是得到了調用對象的鎖。

從上述細節能夠看到，等待/通知機制依託於同步機制，其目的就是確保等待線程從 wait 方法返回後可以感知到通知線程對變量作出的修改。

該圖描述了上面的步驟：

WaitThread 得到了對象的鎖，調用對象的 wait 方法，放棄了鎖，進入的等待隊列，而後 NotifyThread 拿到了對象的鎖，而後調用對象的 notify 方法，將 WatiThread 移動到同步隊列中，最後，NotifyThread 執行完畢，釋放鎖， WaitThread 再次得到鎖並從 wait 方法返回繼續執行。

到這裏，關於應用層面的 wait 和 notify 基本就差很少了，後面的是關於虛擬機層面的拋磚引玉，涉及到 Java 的內置鎖實現，synchronized 關鍵字底層實現，JVM 源碼。算是本文的擴展吧。

注意：咱們看到圖中出現了 Monitor 這個詞，也就是監視器，實際上，在 JDK 的註釋中，也有 The current thread must own this object's monitor 這句話，當前線程必須擁有該對象的監視器。

若是咱們編譯這段含有 synchronized 關鍵字的代碼，就會發現有一段代碼被 monitorenter 指令和 monitorexit 指令括住了，這就是 synchronized 在編譯期間作的事情，那麼，在字節碼被執行的時侯，該指令對應的 c 代碼將會被執行。這裏，咱們必須打住，這裏已經開始涉及到 synchronized 的相關原理了，本篇文章不會討論這個。

wait noitfy 的答案都在 Java HotSpot 虛擬機的 C 代碼中。但 R 大告訴咱們不要輕易閱讀虛擬機源碼，衆多細節可能會掩蓋抽象，致使學習效率不高。若是同窗們有興趣，有大神寫了3篇文章專門從 HotSpot 中解析源碼，地址：

Java的wait()、notify()學習三部曲之一：JVM源碼分析， Java的wait()、notify()學習三部曲之二：修改JVM源碼看參數， Java的wait()、notify()學習三部曲之三：修改JVM源碼控制搶鎖順序， 還有狼哥的 JVM源碼分析之Object.wait/notify實現.

上面四篇文章都從 JVM 的源碼層面解析了 wait ，notify 的實現原理，很是清楚。

拾遺

1. wait(long) 方法，該方法參數是毫秒，也就是說，若是線程等待了指定的毫秒數，就會自動返回該線程。
2. wait（long, int）方法，該方法增長了納秒級別的設置，算法是，前面的毫秒加上後面的納秒，注意，是直接加一毫秒。
3. notify 方法調用後，若是等待的線程不少，JDK 源碼中說將會隨機找一個，可是 JVM 的源碼中其實是找第一個。
4. notifyAll 和 notify 不會當即生效，必須等到調用方執行完同步代碼塊，放棄鎖以後才起做用。

總結

好了，關於 wait noitfy 的使用和基本原理就介紹到這裏，不知道你們發現沒有，併發和虛擬機高度相關。所以，能夠說，學習併發的過程就是學習虛擬機的過程。而閱讀虛擬機裏的 openjdk 代碼讓人頭大，但無論怎麼樣，醜媳婦早晚見公婆，openjdk 代碼是必定要看的，加油！！！！