深入理解Java 線程通信

更新時間：2020年06月22日 09:50:18 作者：認真對待世界的小白

這篇文章主要介紹了Java 線程通信的的相關資料，文中講解非常細致，代碼幫助大家更好的理解和學習，感興趣的朋友可以了解下

當線程在系統(tǒng)內運行時，線程的調度具有一定的透明性，程序通常無法準確控制線程的輪換執(zhí)行，但 Java 也提供了一些機制來保證線程協(xié)調運行。

傳統(tǒng)的線程通信

假設現(xiàn)在系統(tǒng)中有兩個線程，這兩個線程分別代表存款者和取錢者——現(xiàn)在假設系統(tǒng)有一種特殊的要求，系統(tǒng)要求存款者和取錢者不斷地重復存款、取錢的動作，而且要求每當存款者將錢存入指定賬戶后，取錢者就立即取出該筆錢。不允許存款者連續(xù)兩次存錢，也不允許取錢者連續(xù)兩次取錢。

為了實現(xiàn)這種功能，可以借助于 Object 類提供的 wait()、 notify() 和 notifyAll() 三個方法，這三個方法并不屬于 Thread 類，而是屬于 Object 類。但這三個方法必須由同步監(jiān)視器對象來調用，這可分成以下兩種情況。

對于使用 synchronized 修飾的同步方法，因為該類的默認實例（this）就是同步監(jiān)視器，所以可以在同步方法中直接調用這三個方法。
對于使用 synchronized 修飾的同步代碼塊，同步監(jiān)視器是 synchronized 后括號里的對象，所以必須使用該對象調用這三個方法。

關于這三個方法的解釋如下。

wait()：導致當前線程等待，直到其他線程調用該同步監(jiān)視器的 notify() 方法或 notifyAll() 方法來喚醒該線程。該 wait() 方法有三種形式——無時間參數(shù)的 wait (—直等待，直到其他線程通知）、帶毫秒?yún)?shù)的 wait() 和帶毫秒、毫微秒?yún)?shù)的 wait() （這兩種方法都是等待指定時間后自動蘇醒）。調用 wait() 方法的當前線程會釋放對該同步監(jiān)視器的鎖定。
notify()：喚醒在此同步監(jiān)視器上等待的單個線程。如果所有線程都在此同步監(jiān)視器上等待，則會選擇喚醒其中一個線程。選擇是任意性的。只有當前線程放棄對該同步監(jiān)視器的鎖定后（使用 wait() 方法），才可以執(zhí)行被喚醒的線程。
notifyAll()：喚醒在此同步監(jiān)視器上等待的所有線程。只有當前線程放棄對該同步監(jiān)視器的鎖定后，才可以執(zhí)行被喚醒的線程。

程序中可以通過一個旗標來標識賬戶中是否已有存款，當旗標為 false 時，表明賬戶中沒有存款，存款者線程可以向下執(zhí)行，當存款者把錢存入賬戶后，將旗標設為 true ，并調用 notify() 或 notifyAll() 方法來喚醒其他線程；當存款者線程進入線程體后，如果旗標為 true 就調用 wait() 方法讓該線程等待。

當旗標為 true 時，表明賬戶中已經(jīng)存入了存款，則取錢者線程可以向下執(zhí)行，當取錢者把錢從賬戶中取出后，將旗標設為 false ，并調用 notify() 或 notifyAll() 方法來喚醒其他線程；當取錢者線程進入線程體后，如果旗標為 false 就調用 wait() 方法讓該線程等待。

本程序為 Account 類提供 draw() 和 deposit() 兩個方法，分別對應該賬戶的取錢、存款等操作，因為這兩個方法可能需要并發(fā)修改 Account 類的 balance 成員變量的值，所以這兩個方法都使用 synchronized 修飾成同步方法。除此之外，這兩個方法還使用了 wait() 和 notifyAll() 來控制線程的協(xié)作。

public class Account{
  private String accountNo;
  private double balance;
  //標識賬戶中是否已有存款的旗標
  private boolean flag = false;

  public Account(){}

  public Account(String accountNo , double balance){
    this.accountNo = accountNo;
    this.balance = balance;
  }

  public void setAccountNo(String accountNo){
    this.accountNo = accountNo;
  }
  public String getAccountNo(){
     return this.accountNo;
  }

  public double getBalance(){
     return this.balance;
  }
  public synchronized void draw(double drawAmount){
    try{
      //如果flag為假，表明賬戶中還沒有人存錢進去，則取錢方法阻塞
      if (!flag){
        wait();
      }else{
        //執(zhí)行取錢
        System.out.println(Thread.currentThread().getName() + " 取錢:" + drawAmount);
        balance -= drawAmount;
        System.out.println("賬戶余額為：" + balance);
        //將標識賬戶是否已有存款的旗標設為false。
        flag = false;
        //喚醒其他線程
        notifyAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
  }
  public synchronized void deposit(double depositAmount){
    try{
      //如果flag為真，表明賬戶中已有人存錢進去，則存錢方法阻塞
      if (flag){  // ①
        wait();
      }else{
        //執(zhí)行存款
        System.out.println(Thread.currentThread().getName() + " 存款:" + depositAmount);
        balance += depositAmount;
        System.out.println("賬戶余額為：" + balance);
        //將表示賬戶是否已有存款的旗標設為true
        flag = true;
        //喚醒其他線程
        notifyAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
  }

  public int hashCode(){
    return accountNo.hashCode();
  }
  public boolean equals(Object obj){
    if (obj != null && obj.getClass() == Account.class){
      Account target = (Account)obj;
      return target.getAccountNo().equals(accountNo);
    }
    return false;
  }
}

上面程序中的粗體字代碼使用 wait() 和 notifyAll() 進行了控制，對存款者線程而言，當程序進入 deposit() 方法后，如果 flag 為 true ，則表明賬戶中已有存款，程序調用 wait() 方法阻塞；否則程序向下執(zhí)行存款操作，當存款操作執(zhí)行完成后，系統(tǒng)將 flag 設為 true，然后調用 notifyAll() 來喚醒其他被阻塞的線程——如果系統(tǒng)中有存款者線程，存款者線程也會被喚醒，但該存款者線程執(zhí)行到①號代碼處時再次進入阻塞狀態(tài)，只有執(zhí)行 draw() 方法的取錢者線程才可以向下執(zhí)行。同理，取錢者線程的運行流程也是如此。

程序中的存款者線程循環(huán)100次重復存款，而取錢者線程則循環(huán)100次重復取錢，存款者線程和取錢者線程分別調用 Account 對象的 deposit()、 draw() 方法來實現(xiàn)。

public class DrawThread extends Thread{
  //模擬用戶賬戶
  private Account account;
  //當前取錢線程所希望取的錢數(shù)
  private double drawAmount;

  public DrawThread(String name, Account account, double drawAmount){
    super(name);
    this.account = account;
    this.drawAmount = drawAmount;
  }

  //重復100次執(zhí)行取錢操作
  public void run(){
    for (int i = 0 ; i < 100 ; i++ ){
      account.draw(drawAmount);
    }
  }
}

public class DepositThread extends Thread{
  //模擬用戶賬戶
  private Account account;
  //當前取錢線程所希望存款的錢數(shù)
  private double depositAmount;

  public DepositThread(String name, Account account, double depositAmount){
    super(name);
    this.account = account;
    this.depositAmount = depositAmount;
  }

  //重復100次執(zhí)行存款操作
  public void run(){
    for (int i = 0 ; i < 100 ; i++ ){
      account.deposit(depositAmount);
    }    
  }
}

主程序可以啟動任意多個存款線程和取錢線程，可以看到所有的取錢線程必須等存款線程存錢后才可以向下執(zhí)行，而存款線程也必須等取錢線程取錢后才可以向下執(zhí)行。主程序代碼如下。

public class TestDraw{
  public static void main(String[] args){
    //創(chuàng)建一個賬戶
    Account acct = new Account("1234567" , 0);
    new DrawThread("取錢者" , acct , 800).start();
    new DepositThread("存款者甲" , acct , 800).start();
    new DepositThread("存款者乙" , acct , 800).start();
    new DepositThread("存款者丙" , acct , 800).start();
  }
}

運行該程序，可以看到存款者線程、取錢者線程交替執(zhí)行的情形，每當存款者向賬戶中存入800元之后，取錢者線程立即從賬戶中取出這筆錢。存款完成后賬戶余額總是800元，取錢結束后賬戶余額總是0元。運行該程序，會看到如下圖所示的結果。

從上圖中可以看出， 3個存款者線程隨機地向賬戶中存款，只有1個取錢者線程執(zhí)行取錢操作。只有當取錢者取錢后，存款者才可以存款；同理，只有等存款者存款后，取錢者線程才可以取錢。

上圖顯示程序最后被阻塞無法繼續(xù)向下執(zhí)行，這因為3個存款者線程共有300次存款操作，但1個取錢者線程只有100次取錢操作，所以程序最后被阻塞。

注意：上圖所示的阻塞并不是死鎖，對于這種情況，取錢者線程已經(jīng)執(zhí)行結束，而存款者線程只是在等待其他線程來取錢而已，并不是等待其他線程釋放同步監(jiān)視器。不要把死鎖和程序阻塞等同起來！

使用 Condition 控制線程通信

如果程序不使用 synchronized 關鍵字來保證同步，而是直接使用 Lock 對象來保證同步，則系統(tǒng)中不存在隱式的同步監(jiān)視器，也就不能使用 wait()、notify()、notifyAll() 方法進行線程通信了。

當使用 Lock 對象來保證同步時，Java 提供了一個 Condition 類來保持協(xié)調，使用 Condition 可以讓那些已經(jīng)得到 Lock 對象卻無法繼續(xù)執(zhí)行的線程釋放 Lock 對象，Condition 對象也可以喚醒其他處于等待的線程。

Condition 將同步監(jiān)視器方法（wait()、notify() 和 notifyAll() ）分解成截然不同的對象，以便通過將這些對象與 Lock 對象組合使用，為每個對象提供多個等待集（wait-set）。在這種情況下，Lock 替代了同步方法或同步代碼塊，Condition 替代了同步監(jiān)視器的功能。

Condition 實例被綁定在一個 Lock 對象上。要獲得特定 Lock 實例的 Condition 實例，調用 Lock 對象的 newCondition() 方法即可。Condition 類提供了如下三個方法。

await()：類似于隱式同步監(jiān)視器上的 wait() 方法，導致當前線程等待，直到其他線程調用該 Condition 的 signal() 方法或 signalAll() 方法來喚醒該線程。該 await() 方法有更多變體，如 long awaitNanos(long nanosTimeout)、 void awaitUninterruptibly() 、 awaitUntil(Date deadline) 等，可以完成更豐富的等待操作。
signal()：喚醒在此 Lock 對象上等待的單個線程。如果所有線程都在該 Lock 對象上等待，則會選擇喚醒其中一個線程。選擇是任意性的。只有當前線程放棄對該 Lock 對象的鎖定后（使用 await() 方法），才可以執(zhí)行被喚醒的線程。
signalAll()：喚醒在此 Lock 對象上等待的所有線程。只有當前線程放棄對該 Lock 對象的鎖定后，才可以執(zhí)行被喚醒的線程。

下面程序中 Account 使用 Lock 對象來控制同步，并使用 Condition 對象來控制線程的協(xié)調運行。

public class Account{
  //顯示定義Lock對象
  private final Lock lock = new ReentrantLock();
  //獲得指定Lock對象對應的條件變量
  private final Condition cond = lock.newCondition(); 

  private String accountNo;
  private double balance;

  //標識賬戶中是否已經(jīng)存款的旗標
  private boolean flag = false;

  public Account(){}

  public Account(String accountNo , double balance){
    this.accountNo = accountNo;
    this.balance = balance;
  }

  public void setAccountNo(String accountNo){
    this.accountNo = accountNo;
  }
  public String getAccountNo(){
     return this.accountNo;
  }

  public double getBalance(){
     return this.balance;
  }
  public void draw(double drawAmount){
    //加鎖
    lock.lock();
    try{
      //如果賬戶中還沒有存入存款，該線程等待
      if (!flag){
        cond.await();
      }else{
        //執(zhí)行取錢操作
        System.out.println(Thread.currentThread().getName() + " 取錢:" + drawAmount);
        balance -= drawAmount;
        System.out.println("賬戶余額為：" + balance);
        //將標識是否成功存入存款的旗標設為false
        flag = false;
        //喚醒該Lock對象對應的其他線程
        cond.signalAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
    //使用finally塊來確保釋放鎖
    finally{
      lock.unlock();
    }
  }
  public void deposit(double depositAmount){
    lock.lock();
    try{
      //如果賬戶中已經(jīng)存入了存款，該線程等待
      if(flag){
        cond.await();        
      }else{
        //執(zhí)行存款操作
        System.out.println(Thread.currentThread().getName() + " 存款:" + depositAmount);
        balance += depositAmount;
        System.out.println("賬戶余額為：" + balance);
        //將標識是否成功存入存款的旗標設為true
        flag = true;
        //喚醒該Lock對象對應的其他線程
        cond.signalAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
    //使用finally塊來確保釋放鎖
    finally{
      lock.unlock();
    }
  }

  public int hashCode(){
    return accountNo.hashCode();
  }
  public boolean equals(Object obj){
    if (obj != null && obj.getClass() == Account.class){
      Account target = (Account)obj;
      return target.getAccountNo().equals(accountNo);
    }
    return false;
  }
}

顯式地使用 Lock 對象來充當同步監(jiān)視器，則需要使用 Condition 對象來暫停、喚醒指定線程。存取錢的代碼和最上面相同。

使用阻塞隊列（BlockingQueue）控制線程通信

Java5 提供了一個 BlockingQueue 接口，雖然 BlockingQueue 也是 Queue 的子接口，但它的主要用途并不是作為容器，而是作為線程同步的工具。 BlockingQueue 具有一個特征：當生產(chǎn)者線程試圖向 BlockingQueue 中放入元素時，如果該隊列已滿，則該線程被阻塞；當消費者線程試圖從 BlockingQueue 中取出元素時，如果該隊列已空，則該線程被阻塞。

程序的兩個線程通過交替向 BlockingQueue 中放入元素、取出元素，即可很好地控制線程的通信。BlockingQueue 提供如下兩個支持阻塞的方法。

put(E e)：嘗試把 E 元素放入 BlockingQueue 中，如果該隊列的元素己滿，則阻塞該線程。
take()：嘗試從 BlockingQueue 的頭部取出元素，如果該隊列的元素已空，則阻塞該線程。

BlockingQueue 繼承了 Queue 接口，當然也可使用 Queue 接口中的方法。這些方法歸納起來可分為如下三組。

在隊列尾部插入元素。包括 add(E e)、offer(E e) 和 put(E e) 方法，當該隊列已滿時，這三個方法分別會拋出異常、返回 false 、阻塞隊列。
在隊列頭部刪除并返回刪除的元素。包括 remove()、 poll() 和 take() 方法。當該隊列已空時，這三個方法分別會拋出異常、返回 false 、阻塞隊列。
在隊列頭部取出但不刪除元素。包括 element() 和 peek() 方法，當隊列已空時，這兩個方法分別拋出異常、返回 false

BlockingQueue 包含的方法之間的對應關系如下表所示：

BlockingQueue 與其實現(xiàn)類之間的類圖如下圖所示。

上圖中以黑色方框框出的都是 Java7 新增的阻塞隊列?？梢钥吹?， BlockingQueue 包含如下5個實現(xiàn)類。

ArrayBlockingQueue：基于數(shù)組實現(xiàn)的 BlockingQueue 隊列。
LinkedBlockingQueue：基于鏈表實現(xiàn)的 BlockingQueue 隊列。
PriorityBlockingQueue：它并不是標準的阻塞隊列。與前面介紹的 PriorityQueue 類似，該隊列調用 remove()、poll()、take() 等方法取出元素時，并不是取出隊列中存在時間最長的元素，而是隊列中最小的元素。 PriorityBlockingQueue 判斷元素的大小即可根據(jù)元素（實現(xiàn) Comparable 接口）的本身大小來自然排序，也可使用 Comparator 進行定制排序。
SynchronousQueue：同步隊列。對該隊列的存、取操作必須交替進行。
DelayQueue：它是一個特殊的 BlockingQueue ，底層基于 PriorityBlockingQueue 實現(xiàn)。不過，DelayQueue 要求集合元素都實現(xiàn) Delay 接口（該接口里只有一個 long getDelay() 方法），DelayQueue 根據(jù)集合元素的 getDalay() 方法的返回值進行排序。

下面以 ArrayBlockingQueue 為例介紹阻塞隊列的功能和用法。下面先用一個最簡單的程序來測試 BlockingQueue 的 put() 方法。

public class BlockingQueueTest {
  public static void main(String[] args) throws Exception {
    BlockingQueue<String> bq = new ArrayBlockingQueue<>(2);
    bq.put("Java"); // 與bq.add("Java")、bq.offer("Java") 相同
    bq.put("Java"); // 與bq.add("Java")、bq.offer("Java") 相同
    bq.put("Java"); // ① 阻塞線程
  }
}

上面程序先定義一個大小為2的 BlockingQueue，程序先向該隊列中放入兩個元素，此時隊列還沒有滿，兩個元紊都可以放入，因此使用 put()、add() 和 offer() 方法效果完全一樣。當程序試圖放入第三個元素時，如果使用 put() 方法嘗試放入元素將會阻寒線程，如上面程序①號代碼所示。如果使用 add() 方法嘗試放入元素將會引發(fā)異常；如果使用 offer() 方法嘗試放入元素則會返回 false，元素不會被放入。

與此類似的是，在 BlockingQueue 已空的情況下，程序使用 take() 方法嘗試取出元素將會阻塞線程：使用 remove() 方法嘗試取出元素將引發(fā)異常：使用 poll() 方法嘗試取出元素將返回 false，元索不會被刪除。

掌握了 BlodcingQuene 阻塞隊列的特性之后，下面程序就可以利用 BlockingQueue 來實現(xiàn)線程通信了。

public class Producer extends Thread {

  private BlockingQueue<String> bq;
  public Producer(BlockingQueue<String> bq) {
    this.bq = bq;
  }
  public void run() {
    String[] strArr = new String[] {
        "Java",
        "Struts",
        "Spring"
    };
    
    for(int i=0;i<99999999;i++) {
      System.out.println(getName()+"生產(chǎn)者準備生產(chǎn)集合元素");
      try {
        Thread.sleep(200);
        // 嘗試放入元素，如果隊列已滿，則線程被阻塞
        bq.put(strArr[i%3]);
      }catch(Exception ex) {
        ex.printStackTrace();
      }
      System.out.println(getName()+"生產(chǎn)完成："+bq);
    }
  }
}

public class Consumer extends Thread {
  private BlockingQueue<String> bq;
  public Consumer(BlockingQueue<String> bq) {
    this.bq = bq;
  }
  public void run() {
    while(true) {
      System.out.println(getName()+"消費者準備消費集合元素！");
      try {
        Thread.sleep(200);
        // 嘗試取出元素，如果隊列已空，則線程被阻塞
        bq.take();
      }catch(Exception ex) {
        ex.printStackTrace();
      }
      System.out.println(getName()+"消費完成："+bq);
    }
  }
}


public class BlockingQueueTest2 {
  public static void main(String[] args) {
    // 創(chuàng)建一個容量為1的BlockingQueue
    BlockingQueue<String> bq = new ArrayBlockingQueue<>(1);
    // 啟動3個生產(chǎn)者線程
    new Producer(bq).start();
    new Producer(bq).start();
    new Producer(bq).start();
    // 啟動一個消費者線程
    new Consumer(bq).start();
  }
}

上面程序啟動了 3個生產(chǎn)者線程向 BlockingQueue 集合放入元素，啟動了 1個消費者線程從 BlockingQueue 集合取出元素。本程序的 BlockingQueue 集合容量為1，因此3個生產(chǎn)者線程無法連續(xù)放入元素，必須等待消費者線程取出一個元素后， 3個生產(chǎn)者線程的其中之一才能放入一個元素。運行該程序，會看到如下圖所示的結果。