什么是阻塞隊列？
阻塞隊列是一個在隊列基礎上又支持了兩個附加操作的隊列。
2個附加操作：
支持阻塞的插入方法：隊列滿時，隊列會阻塞插入元素的線程，直到隊列不滿。
支持阻塞的移除方法：隊列空時，獲取元素的線程會等待隊列變為非空。
阻塞隊列的應用場景
阻塞隊列常用於生產者和消費者的場景，生產者是向隊列里添加元素的線程，消費者是從隊列里取元素的線程。簡而言之，阻塞隊列是生產者用來存放元素、消費者獲取元素的容器。
幾個方法
在阻塞隊列不可用的時候，上述2個附加操作提供了四種處理方法
| 方法\處理方式 | 拋出異常 | 返回特殊值 | 一直阻塞 | 超時退出 |
|---|---|---|---|---|
| 插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除方法 | remove() | poll() | take() | poll(time,unit) |
| 檢查方法 | element() | peek() | 不可用 | 不可用 |
JAVA里的阻塞隊列
JDK 7 提供了7個阻塞隊列，如下
1、ArrayBlockingQueue 數組結構組成的有界阻塞隊列。
此隊列按照先進先出（FIFO）的原則對元素進行排序，但是默認情況下不保證線程公平的訪問隊列，即如果隊列滿了，那么被阻塞在外面的線程對隊列訪問的順序是不能保證線程公平（即先阻塞，先插入）的。
2、LinkedBlockingQueue一個由鏈表結構組成的有界阻塞隊列
此隊列按照先出先進的原則對元素進行排序
3、PriorityBlockingQueue 支持優先級的無界阻塞隊列
4、DelayQueue 支持延時獲取元素的無界阻塞隊列，即可以指定多久才能從隊列中獲取當前元素
5、SynchronousQueue不存儲元素的阻塞隊列，每一個put必須等待一個take操作，否則不能繼續添加元素。並且他支持公平訪問隊列。
6、LinkedTransferQueue由鏈表結構組成的無界阻塞TransferQueue隊列。相對於其他阻塞隊列，多了tryTransfer和transfer方法
transfer方法
如果當前有消費者正在等待接收元素（take或者待時間限制的poll方法），transfer可以把生產者傳入的元素立刻傳給消費者。如果沒有消費者等待接收元素，則將元素放在隊列的tail節點，並等到該元素被消費者消費了才返回。
tryTransfer方法
用來試探生產者傳入的元素能否直接傳給消費者。，如果沒有消費者在等待，則返回false。和上述方法的區別是該方法無論消費者是否接收，方法立即返回。而transfer方法是必須等到消費者消費了才返回。
7、LinkedBlockingDeque鏈表結構的雙向阻塞隊列，優勢在於多線程入隊時，減少一半的競爭。
如何使用阻塞隊列來實現生產者-消費者模型？
通知模式實現：所謂通知模式，就是當生產者往滿的隊列里添加元素時會阻塞住生產者，當消費者消費了一個隊列中的元素后，會通知生產者當前隊列可用。
使用BlockingQueue解決生產者消費者問題
為什么BlockingQueue適合解決生產者消費者問題
任何有效的生產者-消費者問題解決方案都是通過控制生產者put()方法（生產資源）和消費者take()方法（消費資源）的調用來實現的，一旦你實現了對方法的阻塞控制，那么你將解決該問題。
Java通過BlockingQueue提供了開箱即用的支持來控制這些方法的調用（一個線程創建資源，另一個消費資源）。java.util.concurrent包下的BlockingQueue接口是一個線程安全的可用於存取對象的隊列。
BlockingQueue是一種數據結構，支持一個線程往里存資源，另一個線程從里取資源。這正是解決生產者消費者問題所需要的，那么讓我們開始解決該問題吧。
生產者
以下代碼用於生產者線程
package io.ymq.example.thread;
import java.util.concurrent.BlockingQueue;
/**
* 描述:生產者
*
* @author yanpenglei
* @create 2018-03-14 15:52
**/
class Producer implements Runnable {
protected BlockingQueue<Object> queue;
Producer(BlockingQueue<Object> theQueue) {
this.queue = theQueue;
}
public void run() {
try {
while (true) {
Object justProduced = getResource();
queue.put(justProduced);
System.out.println("生產者資源隊列大小= " + queue.size());
}
} catch (InterruptedException ex) {
System.out.println("生產者 中斷");
}
}
Object getResource() {
try {
Thread.sleep(100);
} catch (InterruptedException ex) {
System.out.println("生產者 讀 中斷");
}
return new Object();
}
}
消費者
以下代碼用於消費者線程
package io.ymq.example.thread;
import java.util.concurrent.BlockingQueue;
/**
* 描述: 消費者
*
* @author yanpenglei
* @create 2018-03-14 15:54
**/
class Consumer implements Runnable {
protected BlockingQueue<Object> queue;
Consumer(BlockingQueue<Object> theQueue) {
this.queue = theQueue;
}
public void run() {
try {
while (true) {
Object obj = queue.take();
System.out.println("消費者 資源 隊列大小 " + queue.size());
take(obj);
}
} catch (InterruptedException ex) {
System.out.println("消費者 中斷");
}
}
void take(Object obj) {
try {
Thread.sleep(100); // simulate time passing
} catch (InterruptedException ex) {
System.out.println("消費者 讀 中斷");
}
System.out.println("消費對象 " + obj);
}
}
測試該解決方案是否運行正常
package io.ymq.example.thread;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
* 描述: 測試
*
* @author yanpenglei
* @create 2018-03-14 15:58
**/
public class ProducerConsumerExample {
public static void main(String[] args) throws InterruptedException {
int numProducers = 4;
int numConsumers = 3;
BlockingQueue<Object> myQueue = new LinkedBlockingQueue<Object>(5);
for (int i = 0; i < numProducers; i++) {
new Thread(new Producer(myQueue)).start();
}
for (int i = 0; i < numConsumers; i++) {
new Thread(new Consumer(myQueue)).start();
}
Thread.sleep(1000);
System.exit(0);
}
}
運行結果
生產者資源隊列大小= 1 生產者資源隊列大小= 1 消費者 資源 隊列大小 1 生產者資源隊列大小= 1 消費者 資源 隊列大小 1 消費者 資源 隊列大小 1 生產者資源隊列大小= 1 生產者資源隊列大小= 3 消費對象 java.lang.Object@1e1aa52b 生產者資源隊列大小= 2 生產者資源隊列大小= 5 消費對象 java.lang.Object@6e740a76 消費對象 java.lang.Object@697853f6 ...... 消費對象 java.lang.Object@41a10cbc 消費對象 java.lang.Object@4963c8d1 消費者 資源 隊列大小 5 生產者資源隊列大小= 5 生產者資源隊列大小= 5 消費者 資源 隊列大小 4 消費對象 java.lang.Object@3e49c35d 消費者 資源 隊列大小 4 生產者資源隊列大小= 5
從輸出結果中,我們可以發現隊列大小永遠不會超過5，消費者線程消費了生產者生產的資源。