- 云服務(wù)器優(yōu)惠
- 阿里云優(yōu)惠券
- 騰訊云優(yōu)惠券
- 京東云優(yōu)惠券
- 站點(diǎn)信息
- 問(wèn)題反饋
- 郵箱：[email protected]
關(guān)注微信
?2016-2026 Matools All rights reserved,
粵ICP備17059708號(hào)
不知道你是否遇到過(guò)面試官讓你手寫生產(chǎn)者消費(fèi)者代碼。別說(shuō)，前段時(shí)間有小伙伴還真的遇到了這種情況。當(dāng)時(shí)是一臉懵逼。
但是，俗話說(shuō)，從哪里跌倒就要從哪里爬起來(lái)。既然這次被問(wèn)到了，那就回去好好研究一下，爭(zhēng)取下一次不再被虐唄。
于是，今天我決定手敲一個(gè)生產(chǎn)者消費(fèi)者模式壓壓驚。（因?yàn)槲乙膊幌胍院蟊幻嬖嚬傺鞍。?br>
生產(chǎn)者消費(fèi)者模式，其實(shí)很簡(jiǎn)單。無(wú)非就是生產(chǎn)者不停的生產(chǎn)數(shù)據(jù)，消費(fèi)者不停的消費(fèi)數(shù)據(jù)。（這不廢話嗎，字面意思我也知道啊）
咳咳。其實(shí)，我們可以拿水池來(lái)舉例。
比如，現(xiàn)在要用多個(gè)注水管往水池里邊注水，那這些注水管就認(rèn)為是生產(chǎn)者。從水池里邊抽水的抽水管就是消費(fèi)者。水池本身就是一個(gè)緩沖區(qū)，用于生產(chǎn)者消費(fèi)者之間的通訊。
好的，跟著我的思路。
既然生產(chǎn)者是生產(chǎn)數(shù)據(jù)的，那總得定義一個(gè)數(shù)據(jù)類吧（Data）
public class Data { private int id; private int num; public int getId() {
return id; } public void setId(int id) { this.id = id; } public int getNum() {
return num; } public void setNum(int num) { this.num = num; } public Data(int
id, int num) { this.id = id; this.num = num; } public Data() { } }
以上數(shù)據(jù)，假設(shè)注水管每次注水的id和注水容量num(單位是升)都是遞增的。并且，單次出水管的出水量和注水管的注水量是一一對(duì)應(yīng)的。
生產(chǎn)者的類Producer和消費(fèi)者類Consumer內(nèi)部都需要維護(hù)一個(gè)阻塞隊(duì)列，來(lái)存儲(chǔ)緩沖區(qū)的數(shù)據(jù)。
public class Producer implements Runnable{ //共享阻塞隊(duì)列 private
BlockingDeque<Data> queue; //是否還在運(yùn)行 private volatile boolean isRunning = true;
//id生成器 private static AtomicInteger count = new AtomicInteger(); //生成隨機(jī)數(shù)
private static Random random = new Random(); public
Producer(BlockingDeque<Data> queue){ this.queue = queue; } @Override public
void run() { try { while(isRunning){ //模擬注水耗時(shí)
Thread.sleep(random.nextInt(1000)); int num = count.incrementAndGet(); Data
data = new Data(num, num);
System.out.println("當(dāng)前>>注水管:"+Thread.currentThread().getName()+"注水容量(L):"+num);
if(!queue.offer(data,2, TimeUnit.SECONDS)){ System.out.println("注水失敗..."); } }
}catch (Exception e){ e.printStackTrace(); } } public void stop(){ isRunning =
false; } }
消費(fèi)者：
public class Consumer implements Runnable{ private BlockingDeque<Data> queue ;
private static Random random = new Random(); public
Consumer(BlockingDeque<Data> queue){ this.queue = queue; } @Override public
void run() { while (true){ try { Data data = queue.take(); //模擬抽水耗時(shí)
Thread.sleep(random.nextInt(1000)); if(data != null){
System.out.println("當(dāng)前<<抽水管:"+Thread.currentThread().getName()+",抽取水容量(L):"+data.getNum());
} }catch (Exception e){ e.printStackTrace(); } } } }
測(cè)試類，假設(shè)有三個(gè)注水管和三個(gè)出水管（即六個(gè)線程）同時(shí)運(yùn)行。等一定時(shí)間后，所有注水管停止注水，則當(dāng)水池空（阻塞隊(duì)列為空）的時(shí)候，出水管也將不再出水。
public class TestProC { public static void main(String[] args) throws
InterruptedException { BlockingDeque<Data> queue = new
LinkedBlockingDeque<>(10); Producer producer1 = new Producer(queue); Producer
producer2 = new Producer(queue); Producer producer3 = new Producer(queue);
Consumer consumer1 = new Consumer(queue); Consumer consumer2 = new
Consumer(queue); Consumer consumer3 = new Consumer(queue); ExecutorService
service = Executors.newCachedThreadPool(); service.execute(producer1);
service.execute(producer2); service.execute(producer3);
service.execute(consumer1); service.execute(consumer2);
service.execute(consumer3); Thread.sleep(3000); producer1.stop();
producer2.stop(); producer3.stop(); Thread.sleep(1000); service.shutdown(); } }
運(yùn)行結(jié)果如下：
到最后一次注水20L的時(shí)候，所有注水管都停止注水了，但此時(shí)水池還沒(méi)空。于是，所有出水管繼續(xù)消費(fèi)水資源，直到最后20L也被消費(fèi)完。
以上，就是一個(gè)典型的生產(chǎn)者消費(fèi)者模式。
可以看到，這種模式有很多優(yōu)點(diǎn)：
1）可以解耦消費(fèi)者和生產(chǎn)者，因?yàn)樗鼈兪莾蓚€(gè)不同的類，互相之間不會(huì)產(chǎn)生影響。
2）支持并發(fā)。生產(chǎn)者只管生產(chǎn)數(shù)據(jù)就行了，生產(chǎn)完直接把數(shù)據(jù)丟到緩沖區(qū)，而不需要等消費(fèi)者消費(fèi)完數(shù)據(jù)才可以生產(chǎn)下一個(gè)數(shù)據(jù)。否則會(huì)造成阻塞，從而影響效率。
3）允許生產(chǎn)者和消費(fèi)者有不同的處理速度。如，當(dāng)生產(chǎn)者生產(chǎn)數(shù)據(jù)比較快的時(shí)候，會(huì)把消費(fèi)者還沒(méi)來(lái)得及處理的數(shù)據(jù)先放到緩沖區(qū)。等有空閑的消費(fèi)者了，再去緩沖區(qū)拿去數(shù)據(jù)。
另外，以上的緩沖區(qū)，我們一般會(huì)使用阻塞隊(duì)列。就像上邊用的LinkedBlockingDeque。
這樣，當(dāng)隊(duì)列滿的時(shí)候，會(huì)阻塞生產(chǎn)者繼續(xù)往隊(duì)列添加數(shù)據(jù)，直到有消費(fèi)者來(lái)消費(fèi)了隊(duì)列中的數(shù)據(jù)。當(dāng)隊(duì)列空的時(shí)候，也會(huì)阻塞消費(fèi)者從隊(duì)列獲取數(shù)據(jù)，直到有生產(chǎn)者把數(shù)據(jù)放入到隊(duì)列中。
阻塞隊(duì)列最好使用有界隊(duì)列（代碼中指定的容量為10）。因?yàn)?，如果生產(chǎn)者的速度遠(yuǎn)遠(yuǎn)大于消費(fèi)者時(shí)，就會(huì)有可能造成隊(duì)列的元素一直增加，直到內(nèi)存耗盡。當(dāng)然，這也需要看實(shí)際的業(yè)務(wù)情況。如果能保證生產(chǎn)者的數(shù)量在可控范圍內(nèi)，不會(huì)給內(nèi)存造成壓力，用無(wú)界隊(duì)列，也未嘗不可。
熱門工具 換一換