阻塞隊(duì)列簡介

阻塞隊(duì)列(BlockingQueue)首先是一個支持先進(jìn)先出的隊(duì)列，與普通的隊(duì)列完全相同；
其次是一個支持阻塞操作的隊(duì)列，即：

• 當(dāng)隊(duì)列滿時，會阻塞執(zhí)行插入操作的線程，直到隊(duì)列不滿。
• 當(dāng)隊(duì)列為空時，會阻塞執(zhí)行獲取操作的線程，直到隊(duì)列不為空。
  

阻塞隊(duì)列用在多線程的場景下，因此阻塞隊(duì)列使用了鎖機(jī)制來保證同步，這里使用的可重入鎖；
而對于阻塞與喚醒機(jī)制則有與鎖綁定的Condition實(shí)現(xiàn)

應(yīng)用場景：生產(chǎn)者消費(fèi)者模式

java中的阻塞隊(duì)列

java中的阻塞隊(duì)列根據(jù)容量可以分為有界隊(duì)列和無界隊(duì)列：

• 有界隊(duì)列：隊(duì)列中只能存儲有限個元素，超出后存放元素線程會被阻塞或者失敗。
• 無界隊(duì)列：隊(duì)列中可以存儲無限個元素。

java8中提供了7種阻塞隊(duì)列阻塞隊(duì)列供開發(fā)者使用，如下表：

另外還有一個在ScheduledThreadPoolExecutor中實(shí)現(xiàn)的DelayedWorkQueue阻塞隊(duì)列，
但這個阻塞隊(duì)列開發(fā)者不能使用。它們之間的UML類圖如下圖：

BlockingQueue接口是阻塞隊(duì)列對外的訪問接口，所有的阻塞隊(duì)列都實(shí)現(xiàn)了BlockQueue中的方法

BlockQueue中方法

作為一個隊(duì)列的核心方法就是入隊(duì)和出隊(duì)。由于存在阻塞策略，BlockQueue將出隊(duì)入隊(duì)的情況分為了四組，每組提供不同的方法：

• 拋出異常：當(dāng)隊(duì)列滿時，如果再往隊(duì)列中插入元素，則拋出IllegalStateException異常；當(dāng)隊(duì)列為空時，從隊(duì)列中獲取元素則拋出NoSuchElementException異常。
• 返回特定值(布爾值)：當(dāng)隊(duì)列滿時，如果再往隊(duì)列中插入元素，則返回false；當(dāng)隊(duì)列為空時，從隊(duì)列中獲取元素則返回null。
• 一直阻塞：當(dāng)隊(duì)列滿時，如果再往隊(duì)列中插入元素，阻塞當(dāng)前線程直到隊(duì)列中至少一個被移除或者響應(yīng)中斷退出；當(dāng)隊(duì)列為空時，則阻塞當(dāng)前線程直到至少一個元素元素入隊(duì)或者響應(yīng)中斷退出。
• 超時退出：當(dāng)隊(duì)列滿時，如果再往隊(duì)列中插入元素，阻塞當(dāng)前線程直到隊(duì)列中至少一個被移除或者達(dá)到指定的等待時間退出或者響應(yīng)中斷退出；當(dāng)隊(duì)列為空時，則阻塞當(dāng)前線程直到至少一個元素元素入隊(duì)或者達(dá)到指定的等待時間退出或者響應(yīng)中斷退出。

對于每種情況BlockingQueue提供的方法如下表：

上述方法一般用于生產(chǎn)者-消費(fèi)者模型中，是其中的生產(chǎn)和消費(fèi)操作隊(duì)列的核心方法。
除了這些方法，BlockingQueue還提供了一些其他的方法如下表：

阻塞隊(duì)列的實(shí)現(xiàn)原理

阻塞隊(duì)列的實(shí)現(xiàn)依靠通知模式實(shí)現(xiàn)：當(dāng)生產(chǎn)者向滿了的隊(duì)列中添加元素時，會阻塞住生產(chǎn)者，
直到消費(fèi)者消費(fèi)了一個隊(duì)列中的元素后會通知消費(fèi)者隊(duì)列可用，此時再由生產(chǎn)者向隊(duì)列中添加元素。反之亦然。

阻塞隊(duì)列的阻塞喚醒依靠Condition——條件隊(duì)列來實(shí)現(xiàn)。

以ArrayBlockingQueue為例說明：

ArrayBlockingQueue的定義：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
 implements BlockingQueue<E>, java.io.Serializable {
 
 /** The queued items */
 //以數(shù)組的結(jié)構(gòu)存儲隊(duì)列的元素，采用的是循環(huán)數(shù)組
 final Object[] items;

 /** items index for next take, poll, peek or remove */
 //隊(duì)列的隊(duì)頭索引
 int takeIndex;

 /** items index for next put, offer, or add */
 //隊(duì)列的隊(duì)尾索引
 int putIndex;

 /** Number of elements in the queue */
 //隊(duì)列中元素的個數(shù)
 int count;

 /** Main lock guarding all access */
 //對于ArrayBlockingQueue所有的操作都需要加鎖，
 final ReentrantLock lock;

 /** Condition for waiting takes */
 //條件隊(duì)列，當(dāng)隊(duì)列為空時阻塞消費(fèi)者并在生產(chǎn)者生產(chǎn)后喚醒消費(fèi)者
 private final Condition notEmpty;

 /** Condition for waiting puts */
 //條件隊(duì)列，當(dāng)隊(duì)列滿時阻塞生產(chǎn)者，并在消費(fèi)者消費(fèi)隊(duì)列后喚醒生產(chǎn)者
 private final Condition notFull;
}

根據(jù)類的定義字段可以看到，有兩個Condition條件隊(duì)列，猜測以下過程

• 當(dāng)隊(duì)列為空，消費(fèi)者試圖消費(fèi)時應(yīng)該調(diào)用notEmpty.await()方法阻塞，并在生產(chǎn)者生產(chǎn)后調(diào)用notEmpty.single()方法
• 當(dāng)隊(duì)列已滿，生產(chǎn)者試圖放入元素應(yīng)調(diào)用notFull.await()方法阻塞，并在消費(fèi)者消費(fèi)隊(duì)列后調(diào)用notFull.single()方法向隊(duì)

向隊(duì)列中添加元素put()方法的添加過程。

 /**
 * 向隊(duì)列中添加元素
 * 當(dāng)隊(duì)列已滿時需要阻塞當(dāng)前線程
 * 放入元素后喚醒因隊(duì)列為空阻塞的消費(fèi)者
 */
 public void put(E e) throws InterruptedException {
 checkNotNull(e);
 final ReentrantLock lock = this.lock;
 lock.lockInterruptibly();
 try {
  //當(dāng)隊(duì)列已滿時需要notFull.await()阻塞當(dāng)前線程
  //offer(e,time,unit)方法就是阻塞的時候加了超時設(shè)定
  while (count == items.length)
  notFull.await();
  //放入元素的過程
  enqueue(e);
 } finally {
  lock.unlock();
 }
 }
 
 /**enqueue實(shí)際添加元素的方法*/
 private void enqueue(E x) {
 // assert lock.getHoldCount() == 1;
 // assert items[putIndex] == null;
 final Object[] items = this.items;
 items[putIndex] = x;
 if (++putIndex == items.length)
  putIndex = 0;
 count++;
 //如果條件隊(duì)列中存在等待的線程
 //喚醒
 notEmpty.signal();
 }

從隊(duì)列中獲取元素take()方法的獲取過程。

 /**
 * 從隊(duì)列中獲取元素
 * 當(dāng)隊(duì)列已空時阻塞當(dāng)前線程
 * 從隊(duì)列中消費(fèi)元素后喚醒等待的生產(chǎn)線程
 */
 public E take() throws InterruptedException {
 final ReentrantLock lock = this.lock;
 lock.lockInterruptibly();
 try {
  //隊(duì)列為空需要阻塞當(dāng)前線程
  while (count == 0)
  notEmpty.await();
  //獲取元素的過程
  return dequeue();
 } finally {
  lock.unlock();
 }
 }
 
 /**dequeue實(shí)際消費(fèi)元素的方法*/
 private E dequeue() {
 // assert lock.getHoldCount() == 1;
 // assert items[takeIndex] != null;
 final Object[] items = this.items;
 @SuppressWarnings("unchecked")
 E x = (E) items[takeIndex];
 items[takeIndex] = null;
 if (++takeIndex == items.length)
  takeIndex = 0;
 count--;
 if (itrs != null)
  itrs.elementDequeued();
 //消費(fèi)元素后從喚醒阻塞的生產(chǎn)者線程
 notFull.signal();
 return x;
 }

總結(jié)

阻塞隊(duì)列提供了不同于普通隊(duì)列的增加、刪除元素的方法，核心在與隊(duì)列滿時阻塞生產(chǎn)者和隊(duì)列空時阻塞消費(fèi)者。
這一阻塞過程依靠與鎖綁定的Condition對象實(shí)現(xiàn)。Condition接口的實(shí)現(xiàn)在AQS中實(shí)現(xiàn)，具體的實(shí)現(xiàn)類是
ConditionObject

以上就是詳解java中的阻塞隊(duì)列的詳細(xì)內(nèi)容，更多關(guān)于java 阻塞隊(duì)列的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論