生產(chǎn)者消費(fèi)者模式的幾種實(shí)現(xiàn)方式
拿我們生活中的例子來(lái)說(shuō)，工廠生產(chǎn)出來(lái)的產(chǎn)品總是要輸出到外面使用的，這就是生產(chǎn)與消費(fèi)的概念。
在我們實(shí)際的軟件開(kāi)發(fā)過(guò)程中，經(jīng)常會(huì)碰到如下場(chǎng)景：某個(gè)模塊負(fù)責(zé)產(chǎn)生數(shù)據(jù)，這些數(shù)據(jù)由另一個(gè)模塊來(lái)負(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)者模式
相信大家都有去吃過(guò)日本料理。有個(gè)很誘人的餐食就是烤肉，烤肉師父會(huì)站在一邊一直烤肉，再將烤好的肉放在一個(gè)盤子中;而流著口水的我們這些食客會(huì)坐在一邊，只要盤子里有肉我們就會(huì)一直去吃。
在這個(gè)生活案例中，烤肉師父就是生產(chǎn)者，他就負(fù)責(zé)烤肉，烤完了就把肉放在盤子里，而不是直接遞給食客(即不用通知食客去吃肉)，如果盤子肉滿，師父就會(huì)停一會(huì)，直到有人去食用烤肉后再去進(jìn)行生產(chǎn)肉;而食客的我們就只是盯著盤子，一旦盤子有肉我們就負(fù)責(zé)去吃就行;
整個(gè)過(guò)程中食客與烤肉師父都不是直接打交道的，而是都與盤子進(jìn)行交互。
盤子充當(dāng)了一個(gè)緩沖區(qū)的概念，有東西生產(chǎn)出來(lái)就把東西放進(jìn)去，盤子也是有大小限制，超過(guò)盤子大小就會(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是無(wú)界的，而arrayblockingqueue 是有界的。
下面使用阻塞隊(duì)列實(shí)現(xiàn)生產(chǎn)者消費(fèi)者：
生產(chǎn)者：
|
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
|
import java.util.concurrent.blockingqueue;public class producer implements runnable{ private final blockingqueue blockingqueue; //設(shè)置隊(duì)列緩存的大小。生產(chǎn)過(guò)程中超過(guò)這個(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)出來(lái)的產(chǎn)品放在隊(duì)列緩存中 blockingqueue.put(task); ++task; //讓其停止一會(huì)，便于查看效果 thread.sleep(1000); } catch (interruptedexception e) { e.printstacktrace(); } } }} |
消費(fèi)者：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
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è)試：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
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)閘inkedblockingqueue是無(wú)界隊(duì)列，所以生產(chǎn)者會(huì)不斷去生產(chǎn)，將生產(chǎn)出的任務(wù)放在隊(duì)列中，消費(fèi)者去隊(duì)列中去消費(fèi)：
如果改用有界阻塞隊(duì)列arrayblockingqueue，就可以初始化隊(duì)列的大小。則隊(duì)列中元素超過(guò)隊(duì)列大小的時(shí)候，生產(chǎn)者就會(huì)等待消費(fèi)者消費(fèi)一個(gè)再去生產(chǎn)一個(gè)：
測(cè)試代碼：
初始化一個(gè)大小為10的arrayblockingqueue：
|
1
2
3
4
5
6
7
8
9
|
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)。可以看到生產(chǎn)出來(lái)的產(chǎn)品序號(hào)與消費(fèi)的產(chǎn)品序號(hào)差始終為10(隊(duì)列的大小)：
總結(jié)
以上就是本文關(guān)于生產(chǎn)消費(fèi)者模式實(shí)現(xiàn)方式和線程安全問(wèn)題代碼示例的全部?jī)?nèi)容，希望對(duì)大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站其他相關(guān)專題，如有不足之處，歡迎留言指出。感謝朋友們對(duì)本站的支持！
原文鏈接：http://blog.csdn.net/noaman_wgs/article/details/66969905
