一、什么是BlockingQueue
BlockingQueue即阻塞隊(duì)列，從阻塞這個(gè)詞可以看出，在某些情況下對阻塞隊(duì)列的訪問可能會(huì)造成阻塞。被阻塞的情況主要有如下兩種：
- 當(dāng)隊(duì)列滿了的時(shí)候進(jìn)行入隊(duì)列操作
- 當(dāng)隊(duì)列空了的時(shí)候進(jìn)行出隊(duì)列操作
因此，當(dāng)一個(gè)線程試圖對一個(gè)已經(jīng)滿了的隊(duì)列進(jìn)行入隊(duì)列操作時(shí)，它將會(huì)被阻塞，除非有另一個(gè)線程做了出隊(duì)列操作；同樣，當(dāng)一個(gè)線程試圖對一個(gè)空隊(duì)列進(jìn)行出隊(duì)列操作時(shí)，它將會(huì)被阻塞，除非有另一個(gè)線程進(jìn)行了入隊(duì)列操作。
二、BlockingQueue的用法
阻塞隊(duì)列主要用在生產(chǎn)者/消費(fèi)者的場景，下面這幅圖展示了一個(gè)線程生產(chǎn)、一個(gè)線程消費(fèi)的場景：
負(fù)責(zé)生產(chǎn)的線程不斷的制造新對象并插入到阻塞隊(duì)列中，直到達(dá)到這個(gè)隊(duì)列的上限值。隊(duì)列達(dá)到上限值之后生產(chǎn)線程將會(huì)被阻塞，直到消費(fèi)的線程對這個(gè)隊(duì)列進(jìn)行消費(fèi)。同理，負(fù)責(zé)消費(fèi)的線程不斷的從隊(duì)列中消費(fèi)對象，直到這個(gè)隊(duì)列為空，當(dāng)隊(duì)列為空時(shí)，消費(fèi)線程將會(huì)被阻塞，除非隊(duì)列中有新的對象被插入。
三、BlockingQueue接口中的方法
| - | Throws Exception | Special Value | Blocks | Times Out |
|---|---|---|---|---|
| Insert | add(o) | offer(o) | put(o) | offer(o, timeout, timeunit) |
| Remove | remove(o) | poll() | take() | poll(timeout, timeunit) |
| Examine | element() | peek() |
阻塞隊(duì)列一共有四套方法分別用來進(jìn)行insert、remove和examine，當(dāng)每套方法對應(yīng)的操作不能馬上執(zhí)行時(shí)會(huì)有不同的反應(yīng)，下面這個(gè)表格就分類列出了這些方法：
| - | Throws Exception | Special Value | Blocks | Times Out |
|---|---|---|---|---|
| Insert | add(o) | offer(o) | put(o) | offer(o, timeout, timeunit) |
| Remove | remove(o) | poll() | take() | poll(timeout, timeunit) |
| Examine | element() | peek() |
- ThrowsException：如果操作不能馬上進(jìn)行，則拋出異常
- SpecialValue：如果操作不能馬上進(jìn)行，將會(huì)返回一個(gè)特殊的值，一般是true或者false
- Blocks:如果操作不能馬上進(jìn)行，操作會(huì)被阻塞
- TimesOut:如果操作不能馬上進(jìn)行，操作會(huì)被阻塞指定的時(shí)間，如果指定時(shí)間沒執(zhí)行，則返回一個(gè)特殊值，一般是true或者false
需要注意的是，我們不能向BlockingQueue中插入null，否則會(huì)報(bào)NullPointerException。
四、BlockingQueue的實(shí)現(xiàn)類
BlockingQueue只是java.util.concurrent包中的一個(gè)接口，而在具體使用時(shí)，我們用到的是它的實(shí)現(xiàn)類，當(dāng)然這些實(shí)現(xiàn)類也位于java.util.concurrent包中。在Java6中，BlockingQueue的實(shí)現(xiàn)類主要有以下幾種：
- ArrayBlockingQueue
- DelayQueue
- LinkedBlockingQueue
- PriorityBlockingQueue
- SynchronousQueue
4.1 ArrayBlockingQueue
ArrayBlockingQueue是一個(gè)有邊界的阻塞隊(duì)列，它的內(nèi)部實(shí)現(xiàn)是一個(gè)數(shù)組。有邊界的意思是它的容量是有限的，我們必須在其初始化的時(shí)候指定它的容量大小，容量大小一旦指定就不可改變。
ArrayBlockingQueue是以先進(jìn)先出的方式存儲數(shù)據(jù)，最新插入的對象是尾部，最新移出的對象是頭部。下面是一個(gè)初始化和使用ArrayBlockingQueue的例子：
BlockingQueue queue = new ArrayBlockingQueue(1024);
queue.put("1");
Object object = queue.take();
4.2 DelayQueue
DelayQueue阻塞的是其內(nèi)部元素，DelayQueue中的元素必須實(shí)現(xiàn) java.util.concurrent.Delayed接口，這個(gè)接口的定義非常簡單：
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}
getDelay()方法的返回值就是隊(duì)列元素被釋放前的保持時(shí)間，如果返回0或者一個(gè)負(fù)值，就意味著該元素已經(jīng)到期需要被釋放，此時(shí)DelayedQueue會(huì)通過其take()方法釋放此對象。
從上面Delayed 接口定義可以看到，它還繼承了Comparable接口，這是因?yàn)镈elayedQueue中的元素需要進(jìn)行排序，一般情況，我們都是按元素過期時(shí)間的優(yōu)先級進(jìn)行排序。
例1：為一個(gè)對象指定過期時(shí)間
首先，我們先定義一個(gè)元素，這個(gè)元素要實(shí)現(xiàn)Delayed接口
public class DelayedElement implements Delayed {
private long expired;
private long delay;
private String name;
DelayedElement(String elementName, long delay) {
this. name = elementName;
this. delay= delay;
expired = ( delay + System. currentTimeMillis());
}
@Override
public int compareTo(Delayed o) {
DelayedElement cached=(DelayedElement) o;
return cached.getExpired()> expired?1:-1;
}
@Override
public long getDelay(TimeUnit unit) {
return ( expired - System. currentTimeMillis());
}
@Override
public String toString() {
return "DelayedElement [delay=" + delay + ", name=" + name + "]";
}
public long getExpired() {
return expired;
}
}
設(shè)置這個(gè)元素的過期時(shí)間為3s
public class DelayQueueExample {
public static void main(String[] args) throws InterruptedException {
DelayQueue<DelayedElement> queue= new DelayQueue<>();
DelayedElement ele= new DelayedElement( "cache 3 seconds",3000);
queue.put( ele);
System. out.println( queue.take());
}
}
運(yùn)行這個(gè)main函數(shù)，我們可以發(fā)現(xiàn)，我們需要等待3s之后才會(huì)打印這個(gè)對象。
其實(shí)DelayQueue應(yīng)用場景很多，比如定時(shí)關(guān)閉連接、緩存對象，超時(shí)處理等各種場景，下面我們就拿學(xué)生考試為例讓大家更深入的理解DelayQueue的使用。
例2：把所有考試的學(xué)生看做是一個(gè)DelayQueue，誰先做完題目釋放誰
首先，我們構(gòu)造一個(gè)學(xué)生對象
public class Student implements Runnable,Delayed{
private String name; //姓名
private long costTime;//做試題的時(shí)間
private long finishedTime;//完成時(shí)間
public Student(String name, long costTime) {
this. name = name;
this. costTime= costTime;
finishedTime = costTime + System. currentTimeMillis();
}
@Override
public void run() {
System. out.println( name + " 交卷,用時(shí)" + costTime /1000);
}
@Override
public long getDelay(TimeUnit unit) {
return ( finishedTime - System. currentTimeMillis());
}
@Override
public int compareTo(Delayed o) {
Student other = (Student) o;
return costTime >= other. costTime?1:-1;
}
}
然后在構(gòu)造一個(gè)教師對象對學(xué)生進(jìn)行考試
public class Teacher {
static final int STUDENT_SIZE = 30;
public static void main(String[] args) throws InterruptedException {
Random r = new Random();
//把所有學(xué)生看做一個(gè)延遲隊(duì)列
DelayQueue<Student> students = new DelayQueue<Student>();
//構(gòu)造一個(gè)線程池用來讓學(xué)生們“做作業(yè)”
ExecutorService exec = Executors.newFixedThreadPool(STUDENT_SIZE);
for ( int i = 0; i < STUDENT_SIZE; i++) {
//初始化學(xué)生的姓名和做題時(shí)間
students.put( new Student( "學(xué)生" + (i + 1), 3000 + r.nextInt(10000)));
}
//開始做題
while(! students.isEmpty()){
exec.execute( students.take());
}
exec.shutdown();
}
}
我們看一下運(yùn)行結(jié)果：
學(xué)生2 交卷,用時(shí)3
學(xué)生1 交卷,用時(shí)5
學(xué)生5 交卷,用時(shí)7
學(xué)生4 交卷,用時(shí)8
學(xué)生3 交卷,用時(shí)11
通過運(yùn)行結(jié)果我們可以發(fā)現(xiàn)，每個(gè)學(xué)生在指定開始時(shí)間到達(dá)之后就會(huì)“交卷”（取決于getDelay()方法），并且是先做完的先交卷（取決于compareTo()方法）。
4.3 LinkedBlockingQueue
LinkedBlockingQueue阻塞隊(duì)列大小的配置是可選的，如果我們初始化時(shí)指定一個(gè)大小，它就是有邊界的，如果不指定，它就是無邊界的。說是無邊界，其實(shí)是采用了默認(rèn)大小為Integer.MAX_VALUE的容量 。它的內(nèi)部實(shí)現(xiàn)是一個(gè)鏈表。
和ArrayBlockingQueue一樣，LinkedBlockingQueue 也是以先進(jìn)先出的方式存儲數(shù)據(jù)，最新插入的對象是尾部，最新移出的對象是頭部。下面是一個(gè)初始化和使LinkedBlockingQueue的例子：
BlockingQueue<String> unbounded = new LinkedBlockingQueue<String>();
BlockingQueue<String> bounded = new LinkedBlockingQueue<String>(1024);
bounded.put("Value");
String value = bounded.take();
4.4 PriorityBlockingQueue
PriorityBlockingQueue是一個(gè)沒有邊界的隊(duì)列，它的排序規(guī)則和 java.util.PriorityQueue一樣。需要注意，PriorityBlockingQueue中允許插入null對象。
所有插入PriorityBlockingQueue的對象必須實(shí)現(xiàn) java.lang.Comparable接口，隊(duì)列優(yōu)先級的排序規(guī)則就是按照我們對這個(gè)接口的實(shí)現(xiàn)來定義的。
另外，我們可以從PriorityBlockingQueue獲得一個(gè)迭代器Iterator，但這個(gè)迭代器并不保證按照優(yōu)先級順序進(jìn)行迭代。
下面我們舉個(gè)例子來說明一下，首先我們定義一個(gè)對象類型，這個(gè)對象需要實(shí)現(xiàn)Comparable接口：
public class PriorityElement implements Comparable<PriorityElement> {
private int priority;//定義優(yōu)先級
PriorityElement(int priority) {
//初始化優(yōu)先級
this.priority = priority;
}
@Override
public int compareTo(PriorityElement o) {
//按照優(yōu)先級大小進(jìn)行排序
return priority >= o.getPriority() ? 1 : -1;
}
public int getPriority() {
return priority;
}
public void setPriority(int priority) {
this.priority = priority;
}
@Override
public String toString() {
return "PriorityElement [priority=" + priority + "]";
}
}
然后我們把這些元素隨機(jī)設(shè)置優(yōu)先級放入隊(duì)列中
public class PriorityBlockingQueueExample {
public static void main(String[] args) throws InterruptedException {
PriorityBlockingQueue<PriorityElement> queue = new PriorityBlockingQueue<>();
for (int i = 0; i < 5; i++) {
Random random=new Random();
PriorityElement ele = new PriorityElement(random.nextInt(10));
queue.put(ele);
}
while(!queue.isEmpty()){
System.out.println(queue.take());
}
}
}
看一下運(yùn)行結(jié)果：
PriorityElement [priority=3]
PriorityElement [priority=4]
PriorityElement [priority=5]
PriorityElement [priority=8]
PriorityElement [priority=9]
4.5 SynchronousQueue
SynchronousQueue隊(duì)列內(nèi)部僅允許容納一個(gè)元素。當(dāng)一個(gè)線程插入一個(gè)元素后會(huì)被阻塞，除非這個(gè)元素被另一個(gè)線程消費(fèi)。
