Java多線程 BlockingQueue實現(xiàn)生產(chǎn)者消費者模型詳解
BlockingQueue
BlockingQueue、解決了多線程中，如何高效安全“傳輸”數(shù)據(jù)的問題。程序員無需關(guān)心什么時候阻塞線程，什么時候喚醒線程，該喚醒哪個線程。
方法介紹
BlockingQueue是Queue的子類
void put(E e)
插入指定元素，當(dāng)BlockingQueue為滿，則線程阻塞，進(jìn)入Waiting狀態(tài)，直到BlockingQueue有空閑空間再繼續(xù)。
這里以ArrayBlockingQueue為例進(jìn)行分析
void take()
隊首出隊，當(dāng)BlockingQueue為空，則線程阻塞，進(jìn)入Waiting狀態(tài)，直到BlockingQueue不為空再繼續(xù)。
int drainTo(Collection<? super E> c)
從隊列中批量取出數(shù)據(jù)，并放入到另一個集合中，返回轉(zhuǎn)移數(shù)據(jù)的數(shù)量，只需一次加鎖和解鎖。
BlockingQueue的實現(xiàn)類
ArrayBlockingQueue
/* * Concurrency control uses the classic two-condition algorithm * found in any textbook. */ /** Main lock guarding all access */ final ReentrantLock lock; /** Condition for waiting takes */ private final Condition notEmpty; /** Condition for waiting puts */ private final Condition notFull;
基于數(shù)組實現(xiàn)的BlockingQueue，需要指定隊列容量，可以指定是否為公平鎖；只有一個ReentrantLock，生產(chǎn)者和消費者不能異步執(zhí)行。
LinkedBlockingQueue
/** Lock held by take, poll, etc */ private final ReentrantLock takeLock = new ReentrantLock(); /** Wait queue for waiting takes */ private final Condition notEmpty = takeLock.newCondition(); /** Lock held by put, offer, etc */ private final ReentrantLock putLock = new ReentrantLock(); /** Wait queue for waiting puts */ private final Condition notFull = putLock.newCondition();
基于鏈表實現(xiàn)的BlockingQueue，可以指定隊列容量，不指定隊列容量默認(rèn)為Integer.MAX_VALUE；有兩個ReentrantLock，生產(chǎn)者和消費者可以異步執(zhí)行。
BlockingQueue實現(xiàn)生產(chǎn)者消費者模型
緩沖區(qū)可以存放大量數(shù)據(jù)
生產(chǎn)者和消費者速度各不相同
public class MyThread42 {
public static void main(String[] args)
{
final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
Runnable producerRunnable = new Runnable()
{
int i = 0;
public void run()
{
while (true)
{
try
{
System.out.println("我生產(chǎn)了一個" + i++);
bq.put(i + "");
Thread.sleep(1000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
}
};
Runnable customerRunnable = new Runnable()
{
public void run()
{
while (true)
{
try
{
System.out.println("我消費了一個" + bq.take());
Thread.sleep(3000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
}
};
Thread producerThread = new Thread(producerRunnable);
Thread customerThread = new Thread(customerRunnable);
producerThread.start();
customerThread.start();
}
}
輸出結(jié)果如下
我生產(chǎn)了一個0 我消費了一個1 我生產(chǎn)了一個1 我生產(chǎn)了一個2 我消費了一個2 我生產(chǎn)了一個3 我生產(chǎn)了一個4 我生產(chǎn)了一個5 我消費了一個3 我生產(chǎn)了一個6 我生產(chǎn)了一個7 我生產(chǎn)了一個8 我消費了一個4 我生產(chǎn)了一個9 我生產(chǎn)了一個10 我生產(chǎn)了一個11 我消費了一個5 我生產(chǎn)了一個12 我生產(chǎn)了一個13 我生產(chǎn)了一個14 我消費了一個6 我生產(chǎn)了一個15 我生產(chǎn)了一個16 我消費了一個7 我生產(chǎn)了一個17 我消費了一個8 我生產(chǎn)了一個18 我消費了一個9 我生產(chǎn)了一個19 我消費了一個10 我生產(chǎn)了一個20 我消費了一個11 我生產(chǎn)了一個21 我消費了一個12 我生產(chǎn)了一個22 我消費了一個13 我生產(chǎn)了一個23 我消費了一個14 我生產(chǎn)了一個24 ······
生產(chǎn)者沒有生產(chǎn)到BlockingQueue的容量（極限是10）之前，生產(chǎn)3個，消費1個，再生產(chǎn)到BlockingQueue的容量之后，生產(chǎn)一個消費一個，因為不能超過BlockingQueue的容量。
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
淺談js文件引用方式及其同步執(zhí)行與異步執(zhí)行
下面小編就為大家?guī)硪黄獪\談js文件引用方式及其同步執(zhí)行與異步執(zhí)行。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2016-10-10
這篇文章主要介紹了Spring注入Map集合實現(xiàn)策略模式詳解,Spring提供通過@Resource注解將相同類型的對象注入到Map集合,并將對象的名字作為key,對象作為value封裝進(jìn)入Map,需要的朋友可以參考下2023-11-11
這篇文章主要為大家介紹了Java并發(fā)工具Fork/Join原理，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-06-06
這篇文章主要介紹了Java中生成微信小程序太陽碼的實現(xiàn)方案,本文講解了如何生成微信小程序太陽碼，通過微信提供的兩種方案都可以實現(xiàn)，在實際的項目中建議采用第二種方案，需要的朋友可以參考下2022-05-05
前提是centos6.3系統(tǒng)已經(jīng)安裝好，在這里以64位系統(tǒng)為例，下面是jdk，tomcat，mysql下載安裝步驟，有需要的小伙伴可以參考下2016-10-10
這篇文章主要介紹了javaWeb 四大域?qū)ο笤敿?xì)介紹的相關(guān)資料,需要的朋友可以參考下2016-11-11
這篇文章主要為大家介紹了Spring Cloud Stream 高級特性使用詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-09-09
基于Java實現(xiàn)無向環(huán)和有向環(huán)的檢測
這篇文章主要介紹了如何在?Java?中實現(xiàn)無向環(huán)和有向環(huán)的檢測，文中的示例代碼講解詳細(xì)，對我們學(xué)習(xí)Java有一定的幫助，需要的可以參考一下2022-04-04
SpringSecurity實現(xiàn)動態(tài)加載權(quán)限信息的方法
這篇文章主要介紹了SpringSecurity實現(xiàn)動態(tài)加載權(quán)限信息,本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定需要的朋友可以參考下2022-01-01
這篇文章主要為大家詳細(xì)介紹了Java的lamda表達(dá)式實現(xiàn)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-06-06