1.介紹
阻塞隊(duì)列 (BlockingQueue)是Java util.concurrent包下重要的數(shù)據(jù)結(jié)構(gòu)，BlockingQueue提供了線程安全的隊(duì)列訪問方式：當(dāng)阻塞隊(duì)列進(jìn)行插入數(shù)據(jù)時(shí)，如果隊(duì)列已滿，線程將會(huì)阻塞等待直到隊(duì)列非滿；從阻塞隊(duì)列取數(shù)據(jù)時(shí)，如果隊(duì)列已空，線程將會(huì)阻塞等待直到隊(duì)列非空。并發(fā)包下很多高級(jí)同步類的實(shí)現(xiàn)都是基于BlockingQueue實(shí)現(xiàn)的。
JDK7提供了以下7個(gè)阻塞隊(duì)列：
ArrayBlockingQueue ：由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列。
LinkedBlockingQueue ：由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列。
PriorityBlockingQueue ：支持優(yōu)先級(jí)排序的無(wú)界阻塞隊(duì)列。
DelayQueue：使用優(yōu)先級(jí)隊(duì)列實(shí)現(xiàn)的無(wú)界阻塞隊(duì)列。
SynchronousQueue：不存儲(chǔ)元素的阻塞隊(duì)列。
LinkedTransferQueue：鏈表結(jié)構(gòu)組成的無(wú)界阻塞隊(duì)列。
LinkedBlockingDeque：鏈表結(jié)構(gòu)組成的雙向阻塞隊(duì)列。
阻塞隊(duì)列提供了下列四種處理方法:
這4類方法中，在隊(duì)列已滿（或?yàn)榭眨┑那闆r下，有些會(huì)拋出異常，有些則返回true/false，有些則一直阻塞，還有些則可以設(shè)置超時(shí)時(shí)間，時(shí)間到了后，自動(dòng)退出阻塞狀態(tài)，實(shí)際項(xiàng)目中可根據(jù)需要選取適合的方法。
2.隊(duì)列
隊(duì)列(Queue)與棧(Stack)是數(shù)據(jù)結(jié)構(gòu)中的二種常用結(jié)構(gòu)，隊(duì)列的特點(diǎn)是先進(jìn)先出(FirstInFirstOut)，而Stack是先進(jìn)后出(FirstInLastOut)，說(shuō)得通俗點(diǎn)：Queue就是電影院入場(chǎng)時(shí)人們排起來(lái)的進(jìn)場(chǎng)隊(duì)伍，先來(lái)的人（即：前排在前面的人）先入場(chǎng)，而Statck則是一隊(duì)人依次進(jìn)入了一個(gè)死胡同想出來(lái)，先進(jìn)去(最里面)的人，必須等后面的人(后進(jìn)入的人)出來(lái)了，自己才能出來(lái)。
阻塞隊(duì)列與普通隊(duì)列的區(qū)別在于，當(dāng)隊(duì)列是空的時(shí)，從隊(duì)列中獲取元素的操作將會(huì)被阻塞，或者當(dāng)隊(duì)列是滿時(shí)，往隊(duì)列里添加元素的操作會(huì)被阻塞。試圖從空的阻塞隊(duì)列中獲取元素的線程將會(huì)被阻塞，直到其他的線程往空的隊(duì)列插入新的元素。同樣，試圖往已滿的阻塞隊(duì)列中添加新元素的線程同樣也會(huì)被阻塞，直到其他的線程使隊(duì)列重新變得空閑起來(lái)，如從隊(duì)列中移除一個(gè)或者多個(gè)元素，或者完全清空隊(duì)列.
隊(duì)列在多線程應(yīng)用中，常用于生產(chǎn)-消費(fèi)場(chǎng)景，打個(gè)通俗的比方：很多人早上喜歡去買油條，買油條的人相當(dāng)于消費(fèi)者，做油條的師傅則是生產(chǎn)者。而油鍋邊上用于放油條的鐵架子，可以看成一個(gè)共享的隊(duì)列，師傅做好油條后，一根一根的撈出來(lái)放在架子上，而顧客則按排隊(duì)的順序一根根的付好錢從架子上拿。即：隊(duì)列的一頭，不斷有人在放入東西（生產(chǎn)元素），另一頭不斷有人的消費(fèi)(拿走元素)。這里就有一個(gè)很有趣的現(xiàn)象，如果買的人多，師傅來(lái)不及做，那么第一個(gè)顧客就會(huì)一直等著（后面的所有人也得等著，或稱為阻塞了后面的人），直到師傅炸好一根，然后第一個(gè)顧客買完走了，后面的人才能頂上來(lái)，類似的道理，如果架子放滿了，沒有人來(lái)買，師傅就會(huì)停下來(lái)，等有人來(lái)買了，才會(huì)繼續(xù)做，這就是所謂的隊(duì)列阻塞，而能產(chǎn)生阻塞行為的隊(duì)列稱為阻塞隊(duì)列。
從剛才的描述可以看出，發(fā)生阻塞起碼得滿足下面至少一個(gè)條件：(前提：隊(duì)列是有界的)
1.從隊(duì)列里取元素時(shí)，如果隊(duì)列為空，則代碼一直等在這里（即阻塞），直到隊(duì)列里有東西了，拿到元素了，后面的代碼才能繼續(xù)；
2.向隊(duì)列里放元素時(shí)，如果隊(duì)列滿了（即放不下更多元素），則代碼也會(huì)卡住，直到隊(duì)列里的東西被取走了（即：有空位可以放新元素了），后面的代碼才能繼續(xù)；
3.模擬生產(chǎn)者消費(fèi)者
模擬了買油條的場(chǎng)景，1個(gè)老板在做油條，3個(gè)顧客在排隊(duì)買：
|
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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
|
package com.tl.skyLine.thread;import java.util.concurrent.ArrayBlockingQueue;/** * Created by tl on 17/3/3. */public class BlockingQueueTest { //final成員變量表示常量，只能被賦值一次，賦值后值不再改變。 private static final int queueSize = 5; private static final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(queueSize); private static final int produceSpeed = 2000;//生產(chǎn)速度(越小越快) private static final int consumeSpeed = 10;//消費(fèi)速度(越小越快) //生產(chǎn)者 public static class Producer implements Runnable { @Override public void run() { while (true) { try { System.out.println("老板準(zhǔn)備炸油條了，架子上還能放：" + (queueSize - queue.size()) + "根油條"); queue.put("1根油條"); System.out.println("老板炸好了1根油條，架子上還能放：" + (queueSize - queue.size()) + "根油條"); Thread.sleep(produceSpeed); } catch (InterruptedException e) { e.printStackTrace(); } } } } //消費(fèi)者 public static class Consumer implements Runnable { @Override public void run() { while (true) { try { System.out.println("A 準(zhǔn)備買油條了，架子上還剩" + queue.size() + "根油條"); queue.take(); System.out.println("A 買到1根油條，架子上還剩" + queue.size() + "根油條"); Thread.sleep(consumeSpeed); System.out.println("B 準(zhǔn)備買油條了，架子上還剩" + queue.size() + "根油條"); queue.take(); System.out.println("B 買到1根油條，架子上還剩" + queue.size() + "根油條"); Thread.sleep(consumeSpeed); System.out.println("C 準(zhǔn)備買油條了，架子上還剩" + queue.size() + "根油條"); queue.take(); System.out.println("C 買到1根油條，架子上還剩" + queue.size() + "根油條"); Thread.sleep(consumeSpeed); } catch (InterruptedException e) { e.printStackTrace(); } } } } public static void main(String[] args) { Thread producer = new Thread(new Producer()); Thread consumer = new Thread(new Consumer()); producer.start(); consumer.start(); }} |
結(jié)果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
老板準(zhǔn)備炸油條了，架子上還能放：5根油條老板炸好了1根油條，架子上還能放：4根油條A 準(zhǔn)備買油條了，架子上還剩1根油條A 買到1根油條，架子上還剩0根油條B 準(zhǔn)備買油條了，架子上還剩0根油條老板準(zhǔn)備炸油條了，架子上還能放：5根油條老板炸好了1根油條，架子上還能放：4根油條B 買到1根油條，架子上還剩0根油條C 準(zhǔn)備買油條了，架子上還剩0根油條老板準(zhǔn)備炸油條了，架子上還能放：5根油條老板炸好了1根油條，架子上還能放：4根油條C 買到1根油條，架子上還剩0根油條A 準(zhǔn)備買油條了，架子上還剩0根油條老板準(zhǔn)備炸油條了，架子上還能放：5根油條老板炸好了1根油條，架子上還能放：4根油條A 買到1根油條，架子上還剩0根油條B 準(zhǔn)備買油條了，架子上還剩0根油條老板準(zhǔn)備炸油條了，架子上還能放：5根油條老板炸好了1根油條，架子上還能放：4根油條 |
上面設(shè)置消費(fèi)速度快于生產(chǎn)速度，所有會(huì)發(fā)現(xiàn)控制臺(tái)中，老板炸油條時(shí)，架子上油條賣得快，所以能放4根或者5根油條，而顧客由于購(gòu)買速度快，所以會(huì)經(jīng)常出現(xiàn)架子上還剩0根油條，當(dāng)我們把生產(chǎn)速度produceSpeed跟消費(fèi)速度consumeSpeed對(duì)換一下，又會(huì)發(fā)現(xiàn)相反的結(jié)果。
4.原理分析
我們簡(jiǎn)單模擬一下BlockingQueue，熟悉一下背后的原理：
|
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
|
public class BlockingQueue { private List queue = new LinkedList(); private int limit = 10; public BlockingQueue(int limit){ this.limit = limit; } public synchronized void enqueue(Object item) throws InterruptedException { while(this.queue.size() == this.limit) { wait(); } if(this.queue.size() == 0) { notifyAll(); } this.queue.add(item); } public synchronized Object dequeue() throws InterruptedException{ while(this.queue.size() == 0){ wait(); } if(this.queue.size() == this.limit){ notifyAll(); } return this.queue.remove(0); }} |
我們看一下jdk的實(shí)現(xiàn)過(guò)程：
ArrayBlockingQueue的部分源碼：
|
1
2
3
4
5
6
|
/** Main lock guarding all access */final ReentrantLock lock; /** Condition for waiting takes */private final Condition notEmpty; /** Condition for waiting puts */private final Condition notFull; |
這3個(gè)變量很重要，ReentrantLock重入鎖，notEmpty檢查不為空的Condition以及notFull用來(lái)檢查隊(duì)列未滿的Condition
Condition是一個(gè)接口，里面有二個(gè)重要的方法：
await()：Causesthecurrentthreadtowaituntilitissignalledorinterrupted.即阻塞當(dāng)前線程，直到被通知（喚醒）或中斷
singal():Wakesuponewaitingthread.喚醒阻塞的線程
再來(lái)看put方法：(jdk1.8)
|
1
2
3
4
5
6
7
8
9
10
11
12
|
public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == items.length) notFull.await(); enqueue(e); } finally { lock.unlock(); }} |
1.先獲取鎖
2.然后用while循環(huán)檢測(cè)元素個(gè)數(shù)是否等于items長(zhǎng)度，如果相等，表示隊(duì)列滿了,調(diào)用notFull的await()方法阻塞線程
3.否則調(diào)用enqueue()方法添加元素
4.最后解鎖
這是添加元素的代碼（jdk 1.8）,注意最后一行notEmpty.signal()方法，表示添加完元素后，調(diào)用singal()通知等待（從隊(duì)列中取元素）的線程，隊(duì)列不空(有值)啦，可以來(lái)取東西了。
類似的take()與dequeue()方法則相當(dāng)于逆過(guò)程（注：同樣都是jdk 1.8）
|
1
2
3
4
5
6
7
8
9
10
11
|
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); }} |
類似的：
1. 先加鎖
2. 如果元素個(gè)數(shù)為空，表示隊(duì)列已空，調(diào)用notEmpty的await()阻塞線程，直接隊(duì)列里又有新元素加入為止
3. 然后調(diào)用dequeue 從隊(duì)列里刪除元素
4. 解鎖
dequeue方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
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(); notFull.signal(); return x;} |
倒數(shù)第2行，元素移除后，調(diào)用notFull.singnal喚醒等待（向隊(duì)列添加元素的）線程，隊(duì)列有空位了，可以向里面添加元素了。
總結(jié)
以上就是本文關(guān)于java并發(fā)學(xué)習(xí)之BlockingQueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者詳解的全部?jī)?nèi)容，希望對(duì)大家有所幫助。有什么問題可以隨時(shí)留言，小編會(huì)及時(shí)回復(fù)大家的。感謝朋友們對(duì)本站的支持！
原文鏈接：http://blog.csdn.net/tomcat_2014/article/details/60135152
