您好,登錄后才能下訂單哦!
生產(chǎn)者消費(fèi)者模式的幾種實(shí)現(xiàn)方式
拿我們生活中的例子來說,工廠生產(chǎn)出來的產(chǎn)品總是要輸出到外面使用的,這就是生產(chǎn)與消費(fèi)的概念。
在我們實(shí)際的軟件開發(fā)過程中,經(jīng)常會(huì)碰到如下場(chǎng)景:某個(gè)模塊負(fù)責(zé)產(chǎn)生數(shù)據(jù),這些數(shù)據(jù)由另一個(gè)模塊來負(fù)責(zé)處理(此處的模塊是廣義的,可以是類、函數(shù)、線程、進(jìn)程等)。
產(chǎn)生數(shù)據(jù)的模塊,就形象地稱為生產(chǎn)者;而處理數(shù)據(jù)的模塊,就稱為消費(fèi)者。
第一種:采用wait—notify實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式
1. 一生產(chǎn)者與一消費(fèi)者:
2. 一生產(chǎn)者與多消費(fèi)者:
第二種: 采用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式
3. 使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式
相信大家都有去吃過日本料理。有個(gè)很誘人的餐食就是烤肉,烤肉師父會(huì)站在一邊一直烤肉,再將烤好的肉放在一個(gè)盤子中;而流著口水的我們這些食客會(huì)坐在一邊,只要盤子里有肉我們就會(huì)一直去吃。
在這個(gè)生活案例中,烤肉師父就是生產(chǎn)者,他就負(fù)責(zé)烤肉,烤完了就把肉放在盤子里,而不是直接遞給食客(即不用通知食客去吃肉),如果盤子肉滿,師父就會(huì)停一會(huì),直到有人去食用烤肉后再去進(jìn)行生產(chǎn)肉;而食客的我們就只是盯著盤子,一旦盤子有肉我們就負(fù)責(zé)去吃就行;
整個(gè)過程中食客與烤肉師父都不是直接打交道的,而是都與盤子進(jìn)行交互。
盤子充當(dāng)了一個(gè)緩沖區(qū)的概念,有東西生產(chǎn)出來就把東西放進(jìn)去,盤子也是有大小限制,超過盤子大小就會(huì)阻塞生產(chǎn)者生產(chǎn),等待消費(fèi)者去消費(fèi);當(dāng)盤子為空的時(shí)候 ,即阻塞消費(fèi)者消費(fèi),等待生產(chǎn)者去生產(chǎn)。
編程中阻塞隊(duì)列即可以實(shí)現(xiàn)盤子這個(gè)功能。
阻塞隊(duì)列的特點(diǎn):
當(dāng)隊(duì)列元素已滿的時(shí)候,阻塞插入操作;
當(dāng)隊(duì)列元素為空的時(shí)候,阻塞獲取操作。
ArrayBlockingQueue 與 LinkedBlockingQueue都是支持FIFO(先進(jìn)先出),但是LinkedBlockingQueue是無界的,而ArrayBlockingQueue 是有界的。
下面使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者:
生產(chǎn)者:
import java.util.concurrent.BlockingQueue; public class Producer implements Runnable{ private final BlockingQueue blockingQueue; //設(shè)置隊(duì)列緩存的大小。生產(chǎn)過程中超過這個(gè)大小就暫時(shí)停止生產(chǎn) private final int QUEUE_SIZE = 10; public Producer(BlockingQueue blockingQueue){ this.blockingQueue = blockingQueue; } int task = 1; @Override public void run() { while(true){ try { System.out.println("正在生產(chǎn):" + task); //將生產(chǎn)出來的產(chǎn)品放在隊(duì)列緩存中 blockingQueue.put(task); ++task; //讓其停止一會(huì),便于查看效果 Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } }
消費(fèi)者:
import java.util.concurrent.BlockingQueue; //消費(fèi)者 public class Consumer implements Runnable{ private final BlockingQueue blockingQueue; public Consumer(BlockingQueue blockingQueue){ this.blockingQueue = blockingQueue; } @Override public void run() { //只要阻塞隊(duì)列中有任務(wù),就一直去消費(fèi) while(true){ try { System.out.println("正在消費(fèi): " + blockingQueue.take()); //讓其停止一會(huì),便于查看效果 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } } }
測(cè)試:
import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; /** * 生產(chǎn)者消費(fèi)者模式 * 使用阻塞隊(duì)列BlockingQueue * @author wanggenshen * */ public class TestConPro { public static void main(String[] args){ BlockingQueue blockingQueue = new LinkedBlockingQueue(5); Producer p = new Producer(blockingQueue); Consumer c = new Consumer(blockingQueue); Thread tp = new Thread(p); Thread tc= new Thread(c); tp.start(); tc.start(); } }
因?yàn)長(zhǎng)inkedBlockingQueue是無界隊(duì)列,所以生產(chǎn)者會(huì)不斷去生產(chǎn),將生產(chǎn)出的任務(wù)放在隊(duì)列中,消費(fèi)者去隊(duì)列中去消費(fèi):
如果改用有界阻塞隊(duì)列ArrayBlockingQueue,就可以初始化隊(duì)列的大小。則隊(duì)列中元素超過隊(duì)列大小的時(shí)候,生產(chǎn)者就會(huì)等待消費(fèi)者消費(fèi)一個(gè)再去生產(chǎn)一個(gè):
測(cè)試代碼:
初始化一個(gè)大小為10的ArrayBlockingQueue:
public static void main(String[] args){ BlockingQueue blockingQueue = new ArrayBlockingQueue(10); Producer p = new Producer(blockingQueue); Consumer c = new Consumer(blockingQueue); Thread tp = new Thread(p); Thread tc= new Thread(c); tp.start(); tc.start(); }
測(cè)試中,讓生產(chǎn)者生產(chǎn)速度略快,而消費(fèi)者速度略慢一點(diǎn)??梢钥吹缴a(chǎn)出來的產(chǎn)品序號(hào)與消費(fèi)的產(chǎn)品序號(hào)差始終為10(隊(duì)列的大小):
總結(jié)
以上就是本文關(guān)于生產(chǎn)消費(fèi)者模式實(shí)現(xiàn)方式和線程安全問題代碼示例的全部?jī)?nèi)容,希望對(duì)大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站其他相關(guān)專題,如有不足之處,歡迎留言指出。感謝朋友們對(duì)本站的支持!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。