這篇文章主要為大家展示了“java中BlockingQueue如何實現(xiàn)生產(chǎn)者消費者”，內(nèi)容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領(lǐng)大家一起研究并學(xué)習(xí)一下“java中BlockingQueue如何實現(xiàn)生產(chǎn)者消費者”這篇文章吧。
創(chuàng)新互聯(lián)主營高平網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司,主營網(wǎng)站建設(shè)方案,成都app開發(fā),高平h5微信平臺小程序開發(fā)搭建,高平網(wǎng)站營銷推廣歡迎高平等地區(qū)企業(yè)咨詢
1.介紹
阻塞隊列 (BlockingQueue)是Java util.concurrent包下重要的數(shù)據(jù)結(jié)構(gòu)，BlockingQueue提供了線程安全的隊列訪問方式：當(dāng)阻塞隊列進行插入數(shù)據(jù)時，如果隊列已滿，線程將會阻塞等待直到隊列非滿；從阻塞隊列取數(shù)據(jù)時，如果隊列已空，線程將會阻塞等待直到隊列非空。并發(fā)包下很多高級同步類的實現(xiàn)都是基于BlockingQueue實現(xiàn)的。
JDK7提供了以下7個阻塞隊列：
ArrayBlockingQueue ：由數(shù)組結(jié)構(gòu)組成的有界阻塞隊列。
LinkedBlockingQueue ：由鏈表結(jié)構(gòu)組成的有界阻塞隊列。
PriorityBlockingQueue ：支持優(yōu)先級排序的無界阻塞隊列。
DelayQueue：使用優(yōu)先級隊列實現(xiàn)的無界阻塞隊列。
SynchronousQueue：不存儲元素的阻塞隊列。
LinkedTransferQueue：鏈表結(jié)構(gòu)組成的無界阻塞隊列。
LinkedBlockingDeque：鏈表結(jié)構(gòu)組成的雙向阻塞隊列。
阻塞隊列提供了下列四種處理方法:
這4類方法中，在隊列已滿（或為空）的情況下，有些會拋出異常，有些則返回true/false，有些則一直阻塞，還有些則可以設(shè)置超時時間，時間到了后，自動退出阻塞狀態(tài)，實際項目中可根據(jù)需要選取適合的方法。
2.隊列
隊列(Queue)與棧(Stack)是數(shù)據(jù)結(jié)構(gòu)中的二種常用結(jié)構(gòu)，隊列的特點是先進先出(FirstInFirstOut)，而Stack是先進后出(FirstInLastOut)，說得通俗點：Queue就是電影院入場時人們排起來的進場隊伍，先來的人（即：前排在前面的人）先入場，而Statck則是一隊人依次進入了一個死胡同想出來，先進去(最里面)的人，必須等后面的人(后進入的人)出來了，自己才能出來。
阻塞隊列與普通隊列的區(qū)別在于，當(dāng)隊列是空的時，從隊列中獲取元素的操作將會被阻塞，或者當(dāng)隊列是滿時，往隊列里添加元素的操作會被阻塞。試圖從空的阻塞隊列中獲取元素的線程將會被阻塞，直到其他的線程往空的隊列插入新的元素。同樣，試圖往已滿的阻塞隊列中添加新元素的線程同樣也會被阻塞，直到其他的線程使隊列重新變得空閑起來，如從隊列中移除一個或者多個元素，或者完全清空隊列.
隊列在多線程應(yīng)用中，常用于生產(chǎn)-消費場景，打個通俗的比方：很多人早上喜歡去買油條，買油條的人相當(dāng)于消費者，做油條的師傅則是生產(chǎn)者。而油鍋邊上用于放油條的鐵架子，可以看成一個共享的隊列，師傅做好油條后，一根一根的撈出來放在架子上，而顧客則按排隊的順序一根根的付好錢從架子上拿。即：隊列的一頭，不斷有人在放入東西（生產(chǎn)元素），另一頭不斷有人的消費(拿走元素)。這里就有一個很有趣的現(xiàn)象，如果買的人多，師傅來不及做，那么第一個顧客就會一直等著（后面的所有人也得等著，或稱為阻塞了后面的人），直到師傅炸好一根，然后第一個顧客買完走了，后面的人才能頂上來，類似的道理，如果架子放滿了，沒有人來買，師傅就會停下來，等有人來買了，才會繼續(xù)做，這就是所謂的隊列阻塞，而能產(chǎn)生阻塞行為的隊列稱為阻塞隊列。
從剛才的描述可以看出，發(fā)生阻塞起碼得滿足下面至少一個條件：(前提：隊列是有界的)
1.從隊列里取元素時，如果隊列為空，則代碼一直等在這里（即阻塞），直到隊列里有東西了，拿到元素了，后面的代碼才能繼續(xù)；
2.向隊列里放元素時，如果隊列滿了（即放不下更多元素），則代碼也會卡住，直到隊列里的東西被取走了（即：有空位可以放新元素了），后面的代碼才能繼續(xù)；
3.模擬生產(chǎn)者消費者
模擬了買油條的場景，1個老板在做油條，3個顧客在排隊買：
重慶分公司
重慶分公司
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 queue = new ArrayBlockingQueue(queueSize);
private static final int produceSpeed = 2000;//生產(chǎn)速度(越小越快)
private static final int consumeSpeed = 10;//消費速度(越小越快)
//生產(chǎn)者
public static class Producer implements Runnable {
@Override
public void run() {
while (true) {
try {
System.out.println("老板準備炸油條了，架子上還能放：" + (queueSize - queue.size()) + "根油條");
queue.put("1根油條");
System.out.println("老板炸好了1根油條，架子上還能放：" + (queueSize - queue.size()) + "根油條");
Thread.sleep(produceSpeed);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
//消費者
public static class Consumer implements Runnable {
@Override
public void run() {
while (true) {
try {
System.out.println("A 準備買油條了，架子上還剩" + queue.size() + "根油條");
queue.take();
System.out.println("A 買到1根油條，架子上還剩" + queue.size() + "根油條");
Thread.sleep(consumeSpeed);
System.out.println("B 準備買油條了，架子上還剩" + queue.size() + "根油條");
queue.take();
System.out.println("B 買到1根油條，架子上還剩" + queue.size() + "根油條");
Thread.sleep(consumeSpeed);
System.out.println("C 準備買油條了，架子上還剩" + 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é)果：
老板準備炸油條了，架子上還能放：5根油條 老板炸好了1根油條，架子上還能放：4根油條 A 準備買油條了，架子上還剩1根油條 A 買到1根油條，架子上還剩0根油條 B 準備買油條了，架子上還剩0根油條 老板準備炸油條了，架子上還能放：5根油條 老板炸好了1根油條，架子上還能放：4根油條 B 買到1根油條，架子上還剩0根油條 C 準備買油條了，架子上還剩0根油條 老板準備炸油條了，架子上還能放：5根油條 老板炸好了1根油條，架子上還能放：4根油條 C 買到1根油條，架子上還剩0根油條 A 準備買油條了，架子上還剩0根油條 老板準備炸油條了，架子上還能放：5根油條 老板炸好了1根油條，架子上還能放：4根油條 A 買到1根油條，架子上還剩0根油條 B 準備買油條了，架子上還剩0根油條 老板準備炸油條了，架子上還能放：5根油條 老板炸好了1根油條，架子上還能放：4根油條
上面設(shè)置消費速度快于生產(chǎn)速度，所有會發(fā)現(xiàn)控制臺中，老板炸油條時，架子上油條賣得快，所以能放4根或者5根油條，而顧客由于購買速度快，所以會經(jīng)常出現(xiàn)架子上還剩0根油條，當(dāng)我們把生產(chǎn)速度produceSpeed跟消費速度consumeSpeed對換一下，又會發(fā)現(xiàn)相反的結(jié)果。
4.原理分析
我們簡單模擬一下BlockingQueue，熟悉一下背后的原理：
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的實現(xiàn)過程：
ArrayBlockingQueue的部分源碼：
/** 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個變量很重要，ReentrantLock重入鎖，notEmpty檢查不為空的Condition以及notFull用來檢查隊列未滿的Condition
Condition是一個接口，里面有二個重要的方法：
await()：Causesthecurrentthreadtowaituntilitissignalledorinterrupted.即阻塞當(dāng)前線程，直到被通知（喚醒）或中斷
singal():Wakesuponewaitingthread.喚醒阻塞的線程
再來看put方法：(jdk1.8)
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)檢測元素個數(shù)是否等于items長度，如果相等，表示隊列滿了,調(diào)用notFull的await()方法阻塞線程
3.否則調(diào)用enqueue()方法添加元素
4.最后解鎖
這是添加元素的代碼（jdk 1.8）,注意最后一行notEmpty.signal()方法，表示添加完元素后，調(diào)用singal()通知等待（從隊列中取元素）的線程，隊列不空(有值)啦，可以來取東西了。
類似的take()與dequeue()方法則相當(dāng)于逆過程（注：同樣都是jdk 1.8）
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. 如果元素個數(shù)為空，表示隊列已空，調(diào)用notEmpty的await()阻塞線程，直接隊列里又有新元素加入為止
3. 然后調(diào)用dequeue 從隊列里刪除元素
4. 解鎖
dequeue方法：
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喚醒等待（向隊列添加元素的）線程，隊列有空位了，可以向里面添加元素了。
以上是“java中BlockingQueue如何實現(xiàn)生產(chǎn)者消費者”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學(xué)習(xí)更多知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！
