資訊專欄INFORMATION COLUMN
摘要：為了解決這個(gè)問(wèn)題于是引入了生產(chǎn)者和消費(fèi)者模式。代碼實(shí)現(xiàn)多生產(chǎn)者和多消費(fèi)者實(shí)現(xiàn)阻塞隊(duì)列，將生產(chǎn)者和消費(fèi)者解耦。已經(jīng)滿了等待用使用于多個(gè)生產(chǎn)者的情況說(shuō)明中有元素可以取用使用于多個(gè)消費(fèi)者的情況。
為什么要使用生產(chǎn)者和消費(fèi)者模式
在線程世界里，生產(chǎn)者就是生產(chǎn)數(shù)據(jù)的線程，消費(fèi)者就是消費(fèi)數(shù)據(jù)的線程。
在多線程開(kāi)發(fā)當(dāng)中，如果生產(chǎn)者處理速度很快，而消費(fèi)者處理速度很慢，那么生產(chǎn)者就必須等待消費(fèi)者處理完，才能繼續(xù)生產(chǎn)數(shù)據(jù)。同樣的道理，如果消費(fèi)者的處理能力大于生產(chǎn)者，那么消費(fèi)者就必須等待生產(chǎn)者。為了解決這個(gè)問(wèn)題于是引入了生產(chǎn)者和消費(fèi)者模式。
什么是生產(chǎn)者消費(fèi)者模式生產(chǎn)者消費(fèi)者模式是通過(guò)一個(gè)容器來(lái)解決生產(chǎn)者和消費(fèi)者的強(qiáng)耦合問(wèn)題。生產(chǎn)者和消費(fèi)者彼此之間不直接通訊，而通過(guò)阻塞隊(duì)列來(lái)進(jìn)行通訊，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費(fèi)者處理，直接扔給阻塞隊(duì)列，消費(fèi)者不找生產(chǎn)者要數(shù)據(jù)，而是直接從阻塞隊(duì)列里取，阻塞隊(duì)列就相當(dāng)于一個(gè)緩沖區(qū)，平衡了生產(chǎn)者和消費(fèi)者的處理能力。
這個(gè)阻塞隊(duì)列就是用來(lái)給生產(chǎn)者和消費(fèi)者解耦的?？v觀大多數(shù)設(shè)計(jì)模式，都會(huì)找一個(gè)第三者出來(lái)進(jìn)行解耦，如工廠模式的第三者是工廠類，模板模式的第三者是模板類。
代碼實(shí)現(xiàn)（多生產(chǎn)者 和 多消費(fèi)者）QueueBuffer : 實(shí)現(xiàn)阻塞隊(duì)列，將生產(chǎn)者和消費(fèi)者解耦。它底層是一個(gè)數(shù)組，構(gòu)造的時(shí)候指定數(shù)組的大小。由于實(shí)現(xiàn)的時(shí)多生產(chǎn)這和多消費(fèi)者的模型，所以注意一下 put 和 get 中對(duì)阻塞條件的描述用的是while循環(huán)，這是為了生產(chǎn)者之間或者消費(fèi)者之間他們的內(nèi)部競(jìng)爭(zhēng)所造成的數(shù)組越界異常。
package concurrency;
public class QueueBuffer {
private final int SIZE;
private int count = 0;
private int[] buffer;
public QueueBuffer(int size){
this.SIZE = size;
buffer = new int[SIZE];
}
public int getSIZE(){
return SIZE;
}
public synchronized void put(int value){
while (count == SIZE){ //buffer已經(jīng)滿了 等待get ,用while使用于多個(gè)生產(chǎn)者的情況
try {
wait();
}catch (InterruptedException e){
e.printStackTrace();
}
}
notifyAll(); //說(shuō)明buffer中有元素 可以取
buffer[count++] = value;
System.out.println("Put "+value+" current size = "+count);
}
public synchronized int get(){
while(count == 0){//用while使用于多個(gè)消費(fèi)者的情況。
try {
wait();//buffer為空，需要等到put進(jìn)元素
}catch (InterruptedException e){
e.printStackTrace();
}
}
// notify() 只是去通知其他的線程，但是synchronized 方法里面的代碼還是會(huì)執(zhí)行完畢的。
// synchronized方法本來(lái)就加了鎖。代碼的執(zhí)行跟你的notify()也無(wú)關(guān)，代碼的執(zhí)行是跟你的
// synchronized綁定一起而已。
notifyAll(); //說(shuō)明剛剛從buffer中取出了元素 有空位可以加進(jìn)新的元素
int result = buffer[--count];
System.out.println("Get "+result+" current size = "+count);
return result;
}
}
class Test{
public static void main(String[] args){
QueueBuffer q = new QueueBuffer(10);
new Producer(q);
new Producer(q);
new Producer(q);
new Consumer(q);
new Consumer(q);
new Consumer(q);
System.out.println("Press Control-C to stop.");
}
}
Producer
package concurrency;
import java.util.Random;
public class Producer implements Runnable {
Random rand = new Random(47);
private QueueBuffer q;
Producer(QueueBuffer q) {
this.q = q;
new Thread(this, "Producer").start();
}
public void run() {
while (true) {
q.put(rand.nextInt(q.getSIZE()));
Thread.yield();
}
}
}
Consumer
package concurrency;
public class Consumer implements Runnable {
private QueueBuffer q;
Consumer(QueueBuffer q) {
this.q = q;
new Thread(this, "Consumer").start();
}
public void run() {
while (true){
q.get();
Thread.yield();
}
}
}
注意事項(xiàng)
調(diào)用obj的wait(), notify()方法前，必須獲得obj對(duì)象的鎖，也就是必須寫在synchronized(obj) {…} 代碼段內(nèi)。
調(diào)用obj.wait()后，線程A就釋放了obj的鎖，否則線程B無(wú)法獲得obj鎖，也就無(wú)法在synchronized(obj) {…} 代碼段內(nèi)喚醒A。
當(dāng)obj.wait()方法返回后，線程A需要再次獲得obj鎖，才能繼續(xù)執(zhí)行。
如果A1,A2,A3都在obj.wait()，則B調(diào)用obj.notify()只能喚醒A1,A2,A3中的一個(gè)（具體哪一個(gè)由JVM決定）。
obj.notifyAll()則能全部喚醒A1,A2,A3，但是要繼續(xù)執(zhí)行obj.wait()的下一條語(yǔ)句，必須獲得obj鎖，因此，A1,A2,A3只有一個(gè)有機(jī)會(huì)獲得鎖繼續(xù)執(zhí)行，例如A1，其余的需要等待A1釋放obj鎖之后才能繼續(xù)執(zhí)行。
當(dāng)B調(diào)用obj.notify/notifyAll的時(shí)候，B正持有obj鎖，因此，A1,A2,A3雖被喚醒，但是仍無(wú)法獲得obj鎖。直到B退出synchronized塊，釋放obj鎖后，A1,A2,A3中的一個(gè)才有機(jī)會(huì)獲得鎖繼續(xù)執(zhí)行。這一點(diǎn)很重要，并不是調(diào)用 notify 或者 notifyAll 之后馬上釋放鎖，而是執(zhí)行完相應(yīng)的synchronized代碼段。
文章版權(quán)歸作者所有，未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址：http://systransis.cn/yun/70641.html
摘要：前言這篇主要來(lái)講解多線程中一個(gè)非常經(jīng)典的設(shè)計(jì)模式包括它的基礎(chǔ)到拓展希望大家能夠有所收獲生產(chǎn)者消費(fèi)者模式簡(jiǎn)述此設(shè)計(jì)模式中主要分兩類線程生產(chǎn)者線程和消費(fèi)者線程生產(chǎn)者提供數(shù)據(jù)和任務(wù)消費(fèi)者處理數(shù)據(jù)和任務(wù)該模式的核心就是數(shù)據(jù)和任務(wù)的交互點(diǎn)共享內(nèi)存緩 前言 這篇主要來(lái)講解多線程中一個(gè)非常經(jīng)典的設(shè)計(jì)模式包括它的基礎(chǔ)到拓展希望大家能夠有所收獲 生產(chǎn)者-消費(fèi)者模式簡(jiǎn)述 此設(shè)計(jì)模式中主要分兩類線程：生產(chǎn)者...
摘要：微軟的雖然引入了事件機(jī)制，可以在隊(duì)列收到消息時(shí)觸發(fā)事件，通知訂閱者。由微軟作為主要貢獻(xiàn)者的，則對(duì)以及做了進(jìn)一層包裝，并能夠很好地實(shí)現(xiàn)這一模式。 在分布式服務(wù)框架中，一個(gè)最基礎(chǔ)的問(wèn)題就是遠(yuǎn)程服務(wù)是怎么通訊的，在Java領(lǐng)域中有很多可實(shí)現(xiàn)遠(yuǎn)程通訊的技術(shù)，例如：RMI、MINA、ESB、Burlap、Hessian、SOAP、EJB和JMS等，這些名詞之間到底是些什么關(guān)系呢，它們背后到底是基...
摘要：微軟的雖然引入了事件機(jī)制，可以在隊(duì)列收到消息時(shí)觸發(fā)事件，通知訂閱者。由微軟作為主要貢獻(xiàn)者的，則對(duì)以及做了進(jìn)一層包裝，并能夠很好地實(shí)現(xiàn)這一模式。 在分布式服務(wù)框架中，一個(gè)最基礎(chǔ)的問(wèn)題就是遠(yuǎn)程服務(wù)是怎么通訊的，在Java領(lǐng)域中有很多可實(shí)現(xiàn)遠(yuǎn)程通訊的技術(shù)，例如：RMI、MINA、ESB、Burlap、Hessian、SOAP、EJB和JMS等，這些名詞之間到底是些什么關(guān)系呢，它們背后到底是基...
摘要：直接上代碼注意在使用實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模型時(shí)候，泛型使用若等對(duì)象時(shí)候會(huì)發(fā)現(xiàn)消費(fèi)者出現(xiàn)異常，這是由于傳值和傳引用的區(qū)別，而由于的自動(dòng)裝箱不會(huì)出現(xiàn)此類問(wèn)題，具體可自行嘗試生產(chǎn)者消費(fèi)者獲取失敗主線程 直接上代碼注意在使用blockingqueue實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模型時(shí)候，BlockingQueue泛型使用若atomic等對(duì)象時(shí)候會(huì)發(fā)現(xiàn)消費(fèi)者出現(xiàn)異常，這是由于傳值和傳引用的區(qū)別，而Integer...
摘要：前文回顧上一篇文章重點(diǎn)嘮叨了中協(xié)調(diào)線程間通信的機(jī)制，它有力的保證了線程間通信的安全性以及便利性。所以同一時(shí)刻廚師線程和服務(wù)員線程不會(huì)同時(shí)在等待隊(duì)列中。對(duì)于在操作系統(tǒng)中線程的阻塞狀態(tài)，語(yǔ)言中用和這三個(gè)狀態(tài)分別表示。 前文回顧 上一篇文章重點(diǎn)嘮叨了java中協(xié)調(diào)線程間通信的wait/notify機(jī)制，它有力的保證了線程間通信的安全性以及便利性。本篇將介紹wait/notify機(jī)制的一個(gè)應(yīng)用...
閱讀 1927·2021-11-23 09:51
閱讀 1566·2021-11-19 09:40
閱讀 3238·2021-11-11 11:01
閱讀 1141·2021-09-27 13:34
閱讀 1874·2021-09-22 15:56
閱讀 2150·2019-08-30 15:52
閱讀 1088·2019-08-30 14:13
閱讀 3501·2019-08-30 14:10
