一、什么是BlockingQueue

BlockingQueue即阻塞隊(duì)列，從阻塞這個(gè)詞可以看出，在某些情況下對阻塞隊(duì)列的訪問可能會(huì)造成阻塞。被阻塞的情況主要有如下兩種：

1. 當(dāng)隊(duì)列滿了的時(shí)候進(jìn)行入隊(duì)列操作
2. 當(dāng)隊(duì)列空了的時(shí)候進(jìn)行出隊(duì)列操作
1
2
3

因此，當(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ì)列操作。

在Java中，BlockingQueue的接口位于java.util.concurrent?包中(在Java5版本開始提供)，由上面介紹的阻塞隊(duì)列的特性可知，阻塞隊(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接口中的方法

阻塞隊(duì)列一共有四套方法分別用來進(jìn)行insert、remove和examine，當(dāng)每套方法對應(yīng)的操作不能馬上執(zhí)行時(shí)會(huì)有不同的反應(yīng)，下面這個(gè)表格就分類列出了這些方法：?
asdf

-	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()	?	?

這四套方法對應(yīng)的特點(diǎn)分別是：

1. ThrowsException：如果操作不能馬上進(jìn)行，則拋出異常
2. SpecialValue：如果操作不能馬上進(jìn)行，將會(huì)返回一個(gè)特殊的值，一般是true或者false
3. Blocks:如果操作不能馬上進(jìn)行，操作會(huì)被阻塞
4. TimesOut:如果操作不能馬上進(jìn)行，操作會(huì)被阻塞指定的時(shí)間，如果指定時(shí)間沒執(zhí)行，則返回一個(gè)特殊值，一般是true或者false
1
2
3
4
5

需要注意的是，我們不能向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)類主要有以下幾種：

1. ArrayBlockingQueue
2. DelayQueue
3. LinkedBlockingQueue
4. PriorityBlockingQueue
5. SynchronousQueue
1
2
3
4
5
6

下面我們就分別介紹這幾個(gè)實(shí)現(xiàn)類。

4.1 ArrayBlockingQueue

ArrayBlockingQueue是一個(gè)有邊界的阻塞隊(duì)列，它的內(nèi)部實(shí)現(xiàn)是一個(gè)數(shù)組。有邊界的意思是它的容量是有限的，我們必須在其初始化的時(shí)候指定它的容量大小，容量大小一旦指定就不可改變。

ArrayBlockingQueue是以先進(jìn)先出的方式存儲(chǔ)數(shù)據(jù)，最新插入的對象是尾部，最新移出的對象是頭部。下面是一個(gè)初始化和使用ArrayBlockingQueue的例子：

BlockingQueue queue = new ArrayBlockingQueue(1024);
queue.put("1");
Object object = queue.take();
1
2
3
4

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);
}
1
2
3
4

getDelay()方法的返回值就是隊(duì)列元素被釋放前的保持時(shí)間，如果返回0或者一個(gè)負(fù)值，就意味著該元素已經(jīng)到期需要被釋放，此時(shí)DelayedQueue會(huì)通過其take()方法釋放此對象。

從上面Delayed 接口定義可以看到，它還繼承了Comparable接口，這是因?yàn)镈elayedQueue中的元素需要進(jìn)行排序，一般情況，我們都是按元素過期時(shí)間的優(yōu)先級(jí)進(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());}@Overridepublic int compareTo(Delayed o) {DelayedElement cached=(DelayedElement) o;return cached.getExpired()> expired?1:-1;}@Overridepublic long getDelay(TimeUnit unit) {return ( expired - System. currentTimeMillis());}@Overridepublic String toString() {return "DelayedElement [delay=" + delay + ", name=" + name + "]";}public long getExpired() {return expired;}}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

設(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());}
1
2
3
4
5
6
7
8
9

}

運(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();}@Overridepublic void run() {System. out.println( name + " 交卷,用時(shí)" + costTime /1000);}@Overridepublic long getDelay(TimeUnit unit) {return ( finishedTime - System. currentTimeMillis());}@Overridepublic int compareTo(Delayed o) {Student other = (Student) o;return costTime >= other. costTime?1:-1;}}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

然后在構(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();}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

我們看一下運(yùn)行結(jié)果：

學(xué)生2 交卷,用時(shí)3
學(xué)生1 交卷,用時(shí)5
學(xué)生5 交卷,用時(shí)7
學(xué)生4 交卷,用時(shí)8
學(xué)生3 交卷,用時(shí)11
1
2
3
4
5
6

通過運(yùn)行結(jié)果我們可以發(fā)現(xiàn)，每個(gè)學(xué)生在指定開始時(shí)間到達(dá)之后就會(huì)“交卷”（取決于getDelay()方法），并且是先做完的先交卷（取決于compareTo()方法）。

通過查看其源碼可以看到，DelayQueue內(nèi)部實(shí)現(xiàn)用的是PriorityQueue和一個(gè)Lock：

4.3 LinkedBlockingQueue

LinkedBlockingQueue阻塞隊(duì)列大小的配置是可選的，如果我們初始化時(shí)指定一個(gè)大小，它就是有邊界的，如果不指定，它就是無邊界的。說是無邊界，其實(shí)是采用了默認(rèn)大小為Integer.MAX_VALUE的容量 。它的內(nèi)部實(shí)現(xiàn)是一個(gè)鏈表。

和ArrayBlockingQueue一樣，LinkedBlockingQueue 也是以先進(jìn)先出的方式存儲(chǔ)數(shù)據(jù)，最新插入的對象是尾部，最新移出的對象是頭部。下面是一個(gè)初始化和使LinkedBlockingQueue的例子：

BlockingQueue<String> unbounded = new LinkedBlockingQueue<String>();
BlockingQueue<String> bounded   = new LinkedBlockingQueue<String>(1024);
bounded.put("Value");
String value = bounded.take();
1
2
3
4
5

4.4 PriorityBlockingQueue

PriorityBlockingQueue是一個(gè)沒有邊界的隊(duì)列，它的排序規(guī)則和?java.util.PriorityQueue一樣。需要注意，PriorityBlockingQueue中允許插入null對象。

所有插入PriorityBlockingQueue的對象必須實(shí)現(xiàn)?java.lang.Comparable接口，隊(duì)列優(yōu)先級(jí)的排序規(guī)則就是按照我們對這個(gè)接口的實(shí)現(xiàn)來定義的。

另外，我們可以從PriorityBlockingQueue獲得一個(gè)迭代器Iterator，但這個(gè)迭代器并不保證按照優(yōu)先級(jí)順序進(jìn)行迭代。

下面我們舉個(gè)例子來說明一下，首先我們定義一個(gè)對象類型，這個(gè)對象需要實(shí)現(xiàn)Comparable接口：

public class PriorityElement implements Comparable<PriorityElement> {
private int priority;//定義優(yōu)先級(jí)
PriorityElement(int priority) {//初始化優(yōu)先級(jí)this.priority = priority;
}
@Override
public int compareTo(PriorityElement o) {//按照優(yōu)先級(jí)大小進(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 + "]";
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

然后我們把這些元素隨機(jī)設(shè)置優(yōu)先級(jí)放入隊(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());}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

看一下運(yùn)行結(jié)果：

PriorityElement [priority=3]
PriorityElement [priority=4]
PriorityElement [priority=5]
PriorityElement [priority=8]
PriorityElement [priority=9]
1
2
3
4
5
6

4.5 SynchronousQueue

SynchronousQueue隊(duì)列內(nèi)部僅允許容納一個(gè)元素。當(dāng)一個(gè)線程插入一個(gè)元素后會(huì)被阻塞，除非這個(gè)元素被另一個(gè)線程消費(fèi)。

轉(zhuǎn)自：http://blog.csdn.net/suifeng3051/article/details/48807423

總結(jié)

以上是生活随笔為你收集整理的Java并发编程之——BlockingQueue（队列）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。