Java多線程Condition接口原理介紹

更新時間：2020年09月30日 14:36:20 作者：拉里·佩奇

這篇文章主要介紹了Java多線程Condition接口原理介紹，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

Condition接口提供了類似Object的監(jiān)視器方法，與Lock配合可以實現(xiàn)等待/通知模式，但是這兩者在使用方式以及功能特性上還是有差別的

Condition接口詳解

Condition定義了等待/通知兩種類型的方法，當前線程調(diào)用這些方法時，需要提前獲取到Condition對象關聯(lián)的鎖。Condition對象是由Lock對象（調(diào)用Lock對象的newCondition()方法）創(chuàng)建出來的，換句話說，Condition是依賴Lock對象的。

Lock lock = new ReentrantLock();
 Condition condition = lock.newCondition();
 public void conditionWait() throws InterruptedException {
  lock.lock();
  try {
   condition.await();
  } finally {
   lock.unlock();
  }
 }
 public void conditionSignal() throws InterruptedException {
  lock.lock();
  try {
   condition.signal();
  } finally {
   lock.unlock();
  }
 }

一般都會將Condition對象作為成員變量。當調(diào)用await()方法后，當前線程會釋放鎖并在此等待，而其他線程調(diào)用Condition對象的signal()方法，通知當前線程后，當前線程才從await()方法返回，并且在返回前已經(jīng)獲取了鎖。

 /**
  * 當前線程進入等待狀態(tài)直到被通知(signal)或中斷，當前線程進入后臺運行狀態(tài)且從await()方法返回
  * 其他線程調(diào)用該Condition的signal或者signalAll方法，而當前線程被選中喚醒
  * 1、其他線程（interrupt）中斷當前線程
  * 2、如果當前等待線程從await方法返回，那么表明當前線程已經(jīng)獲取了Condition對象的鎖
  */
 void await() throws InterruptedException;

 /**
  * 當前線程進入等待狀態(tài)直到被通知，對中斷不響應
  */
 void awaitUninterruptibly();

 /**
  * <pre> {@code
  * boolean aMethod(long timeout, TimeUnit unit) {
  * long nanos = unit.toNanos(timeout);
  * lock.lock();
  * try {
  *  while (!conditionBeingWaitedFor()) {
  *  if (nanos <= 0L)
  *   return false;
  *  nanos = theCondition.awaitNanos(nanos);
  *  }
  *  // ...
  * } finally {
  *  lock.unlock();
  * }
  * }}</pre>
  * 當前線程進入等待狀態(tài)直到被通知、中斷或超時。返回值表示剩余時間，如果在nanosTimeout納秒之前被喚醒，那么返回值就是nanosTimeout-實際耗時
  * 返回值<=0說明超時
  * 
  */
 long awaitNanos(long nanosTimeout) throws InterruptedException;

 /**
  * 當前線程進入等待狀態(tài)直到被通知、中斷或超時，如果沒有到指定時間被通知返回true，否則返回false
  */
 boolean await(long time, TimeUnit unit) throws InterruptedException;

 /**
  * 喚醒一個等待在Condition上的線程，該線程從等待方法返回之前必須獲得與Condition相關聯(lián)的鎖
  */
 void signal();

獲取一個Condition必須通過Lock的newCondition()方法。下面通過一個有界隊列的示例來深入了解Condition的使用方式。

有界隊列是一種特殊的隊列，當隊列為空時，隊列的獲取操作將會阻塞獲取線程，直到隊列中有新增元素，當隊列已滿時，隊列的插入操作將會阻塞插入線程，直到隊列出現(xiàn)“空位”

public class BoundedQueue<T> {
 private Object[] items;

 // 添加的下標，刪除的下標和數(shù)組當前數(shù)量
 private int addIndex,removeIndex,count;
 private Lock lock = new ReentrantLock();
 private Condition notEmpty = lock.newCondition();
 private Condition notFull = lock.newCondition();
 public BoundedQueue(int size){
  items = new Object[size];
 }

 /**
  * 添加一個元素，如果數(shù)組滿，則添加線程進入等待狀態(tài)，直到有"空位"
  * @author fuyuwei
  * 2017年5月21日 下午6:14:55
  * @param t
  * @throws InterruptedException
  */
 public void add(T t) throws InterruptedException{
  lock.lock();
  try{
   while(count == items.length){
    notFull.await();
   }
   items[addIndex] = t;
   if(++addIndex == items.length)
    addIndex = 0;
   ++count;
   notEmpty.signal();
  }finally{
   lock.unlock();
  }
 }

 /**
  * 由頭部刪除一個元素，如果數(shù)組空，則刪除線程進入等待狀態(tài)，直到有新添加元素
  * @author fuyuwei
  * 2017年5月21日 下午6:20:54
  * @return
  * @throws InterruptedException
  */
 @SuppressWarnings("unchecked")
 public T remove() throws InterruptedException{
  lock.lock();
  try{
   while(count == 0)
    notEmpty.await();
   Object x = items[removeIndex];
   if(++removeIndex == items.length)
    removeIndex = 0;
   --count;
   notFull.signal();
   return (T)x;
  }finally{
   lock.unlock();
  }
 }
}

首先需要獲得鎖，目的是確保數(shù)組修改的可見性和排他性。當數(shù)組數(shù)量等于數(shù)組長度時，表示數(shù)組已滿，則調(diào)用notFull.await()，當前線程隨之釋放鎖并進入等待狀態(tài)。如果數(shù)組數(shù)量不等于數(shù)組長度，表示數(shù)組未滿，則添加元素到數(shù)組中，同時通知等待在notEmpty上的線程，數(shù)組中已經(jīng)有新元素可以獲取。

在添加和刪除方法中使用while循環(huán)而非if判斷，目的是防止過早或意外的通知，只有條件符合才能夠退出循環(huán)?；叵胫疤岬降牡却?通知的經(jīng)典范式，二者是非常類似的

Condition原理分析

ConditionObject是同步器AbstractQueuedSynchronizer的內(nèi)部類，因為Condition的操作需要獲取相關聯(lián)的鎖，所以作為同步器的內(nèi)部類也較為合理。每個Condition對象都包含著一個隊列，該隊列是Condition對象實現(xiàn)等待/通知功能的關鍵。下面將分析Condition的實現(xiàn)，主要包括：等待隊列、等待和通知

等待隊列

等待隊列是一個FIFO的隊列，在隊列中的每個節(jié)點都包含了一個線程引用，該線程就是在Condition對象上等待的線程，如果一個線程調(diào)用了Condition.await()方法，那么該線程將會釋放鎖、構(gòu)造成節(jié)點加入等待隊列并進入等待狀態(tài)

一個Condition包含一個等待隊列，Condition擁有首節(jié)點（firstWaiter）和尾節(jié)點（lastWaiter）。當前線程調(diào)用Condition.await()方法，將會以當前線程構(gòu)造節(jié)點，并將節(jié)點從尾部加入等待隊列，等待隊列的基本結(jié)構(gòu)如下圖所示

如圖所示，Condition擁有首尾節(jié)點的引用，而新增節(jié)點只需要將原有的尾節(jié)點nextWaiter指向它，并且更新尾節(jié)點即可。上述節(jié)點引用更新的過程并沒有使用CAS保證，原因在于調(diào)用await()方法的線程必定是獲取了鎖的線程，也就是說該過程是由鎖來保證線程安全的。在Object的監(jiān)視器模型上，一個對象擁有一個同步隊列和等待隊列，而并發(fā)包中的Lock（更確切地說是同步器）擁有一個同步隊列和多個等待隊列，其對應關系如下圖所示

等待

調(diào)用Condition的await()方法（或者以await開頭的方法），會使當前線程進入等待隊列并釋放鎖，同時線程狀態(tài)變?yōu)榈却隣顟B(tài)。當從await()方法返回時，當前線程一定獲取了Condition相關聯(lián)的鎖。

如果從隊列（同步隊列和等待隊列）的角度看await()方法，當調(diào)用await()方法時，相當于同步隊列的首節(jié)點（獲取了鎖的節(jié)點）移動到Condition的等待隊列中

public final void await() throws InterruptedException {
  if (Thread.interrupted())
   throw new InterruptedException();
  // 當前線程加入等待隊列
  Node node = addConditionWaiter();
  // 釋放同步狀態(tài)，也就是釋放鎖
  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)
   unlinkCancelledWaiters();
  if (interruptMode != 0)
   reportInterruptAfterWait(interruptMode);
 }

調(diào)用該方法的線程成功獲取了鎖的線程，也就是同步隊列中的首節(jié)點，該方法會將當前線程構(gòu)造成節(jié)點并加入等待隊列中，然后釋放同步狀態(tài)，喚醒同步隊列中的后繼節(jié)點，然后當前線程會進入等待狀態(tài)。當?shù)却犃兄械墓?jié)點被喚醒，則喚醒節(jié)點的線程開始嘗試獲取同步狀態(tài)。如果不是通過其他線程調(diào)用Condition.signal()方法喚醒，而是對等待線程進行中斷，則會拋出InterruptedException

通知

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

public final void signal() {
  if (!isHeldExclusively())
   throw new IllegalMonitorStateException();
  Node first = firstWaiter;
  if (first != null)
   doSignal(first);
 }

調(diào)用該方法的前置條件是當前線程必須獲取了鎖，可以看到signal()方法進行了isHeldExclusively()檢查，也就是當前線程必須是獲取了鎖的線程。接著獲取等待隊列的首節(jié)點，將其移動到同步隊列并使用LockSupport喚醒節(jié)點中的線程

節(jié)點從等待隊列移動到同步隊列的過程如下圖所示