生產(chǎn)者消費者之Java簡單實現(xiàn)

HollisChuang 發(fā)布于2019-08-16 10:53 / 1294人閱讀

摘要：為了解決這個問題于是引入了生產(chǎn)者和消費者模式。代碼實現(xiàn)多生產(chǎn)者和多消費者實現(xiàn)阻塞隊列，將生產(chǎn)者和消費者解耦。已經(jīng)滿了等待用使用于多個生產(chǎn)者的情況說明中有元素可以取用使用于多個消費者的情況。

為什么要使用生產(chǎn)者和消費者模式

在線程世界里，生產(chǎn)者就是生產(chǎn)數(shù)據(jù)的線程，消費者就是消費數(shù)據(jù)的線程。

在多線程開發(fā)當中，如果生產(chǎn)者處理速度很快，而消費者處理速度很慢，那么生產(chǎn)者就必須等待消費者處理完，才能繼續(xù)生產(chǎn)數(shù)據(jù)。同樣的道理，如果消費者的處理能力大于生產(chǎn)者，那么消費者就必須等待生產(chǎn)者。為了解決這個問題于是引入了生產(chǎn)者和消費者模式。

什么是生產(chǎn)者消費者模式

生產(chǎn)者消費者模式是通過一個容器來解決生產(chǎn)者和消費者的強耦合問題。生產(chǎn)者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產(chǎn)者要數(shù)據(jù)，而是直接從阻塞隊列里取，阻塞隊列就相當于一個緩沖區(qū)，平衡了生產(chǎn)者和消費者的處理能力。

這個阻塞隊列就是用來給生產(chǎn)者和消費者解耦的。縱觀大多數(shù)設計模式，都會找一個第三者出來進行解耦，如工廠模式的第三者是工廠類，模板模式的第三者是模板類。

代碼實現(xiàn)（多生產(chǎn)者和多消費者）

QueueBuffer : 實現(xiàn)阻塞隊列，將生產(chǎn)者和消費者解耦。它底層是一個數(shù)組，構造的時候指定數(shù)組的大小。由于實現(xiàn)的時多生產(chǎn)這和多消費者的模型，所以注意一下 put 和 get 中對阻塞條件的描述用的是while循環(huán)，這是為了生產(chǎn)者之間或者消費者之間他們的內(nèi)部競爭所造成的數(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使用于多個生產(chǎn)者的情況
            try {
                wait();
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }

        notifyAll(); //說明buffer中有元素 可以取
        buffer[count++] = value;
        System.out.println("Put "+value+" current size = "+count);
    }

    public synchronized int get(){
        while(count == 0){//用while使用于多個消費者的情況。
            try {
                wait();//buffer為空，需要等到put進元素
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
//        notify() 只是去通知其他的線程，但是synchronized 方法里面的代碼還是會執(zhí)行完畢的。
//        synchronized方法本來就加了鎖。代碼的執(zhí)行跟你的notify()也無關，代碼的執(zhí)行是跟你的
//        synchronized綁定一起而已。

        notifyAll(); //說明剛剛從buffer中取出了元素 有空位可以加進新的元素
        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();
        }
    }
}

注意事項

調用obj的wait(), notify()方法前，必須獲得obj對象的鎖，也就是必須寫在synchronized(obj) {…} 代碼段內(nèi)。

調用obj.wait()后，線程A就釋放了obj的鎖，否則線程B無法獲得obj鎖，也就無法在synchronized(obj) {…} 代碼段內(nèi)喚醒A。

當obj.wait()方法返回后，線程A需要再次獲得obj鎖，才能繼續(xù)執(zhí)行。

如果A1,A2,A3都在obj.wait()，則B調用obj.notify()只能喚醒A1,A2,A3中的一個（具體哪一個由JVM決定）。

obj.notifyAll()則能全部喚醒A1,A2,A3，但是要繼續(xù)執(zhí)行obj.wait()的下一條語句，必須獲得obj鎖，因此，A1,A2,A3只有一個有機會獲得鎖繼續(xù)執(zhí)行，例如A1，其余的需要等待A1釋放obj鎖之后才能繼續(xù)執(zhí)行。

當B調用obj.notify/notifyAll的時候，B正持有obj鎖，因此，A1,A2,A3雖被喚醒，但是仍無法獲得obj鎖。直到B退出synchronized塊，釋放obj鎖后，A1,A2,A3中的一個才有機會獲得鎖繼續(xù)執(zhí)行。這一點很重要，并不是調用 notify 或者 notifyAll 之后馬上釋放鎖，而是執(zhí)行完相應的synchronized代碼段。

GPU云服務器云服務器對集群生產(chǎn)消費實現(xiàn)數(shù)據(jù)遷移線程java生產(chǎn)者消費者 java生產(chǎn)者消費生產(chǎn)者消費者

文章版權歸作者所有，未經(jīng)允許請勿轉載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/70641.html

java高并發(fā)之從零到放棄（五）

摘要：前言這篇主要來講解多線程中一個非常經(jīng)典的設計模式包括它的基礎到拓展希望大家能夠有所收獲生產(chǎn)者消費者模式簡述此設計模式中主要分兩類線程生產(chǎn)者線程和消費者線程生產(chǎn)者提供數(shù)據(jù)和任務消費者處理數(shù)據(jù)和任務該模式的核心就是數(shù)據(jù)和任務的交互點共享內(nèi)存緩前言這篇主要來講解多線程中一個非常經(jīng)典的設計模式包括它的基礎到拓展希望大家能夠有所收獲生產(chǎn)者-消費者模式簡述此設計模式中主要分兩類線程：生產(chǎn)者...

meislzhua 2019-08-16 10:55 評論0 收藏0
分布式服務框架之遠程通訊技術及原理分析

摘要：微軟的雖然引入了事件機制，可以在隊列收到消息時觸發(fā)事件，通知訂閱者。由微軟作為主要貢獻者的，則對以及做了進一層包裝，并能夠很好地實現(xiàn)這一模式。在分布式服務框架中，一個最基礎的問題就是遠程服務是怎么通訊的，在Java領域中有很多可實現(xiàn)遠程通訊的技術，例如：RMI、MINA、ESB、Burlap、Hessian、SOAP、EJB和JMS等，這些名詞之間到底是些什么關系呢，它們背后到底是基...

sorra 2019-06-24 18:08 評論0 收藏0
分布式服務框架之遠程通訊技術及原理分析

摘要：微軟的雖然引入了事件機制，可以在隊列收到消息時觸發(fā)事件，通知訂閱者。由微軟作為主要貢獻者的，則對以及做了進一層包裝，并能夠很好地實現(xiàn)這一模式。在分布式服務框架中，一個最基礎的問題就是遠程服務是怎么通訊的，在Java領域中有很多可實現(xiàn)遠程通訊的技術，例如：RMI、MINA、ESB、Burlap、Hessian、SOAP、EJB和JMS等，這些名詞之間到底是些什么關系呢，它們背后到底是基...

0xE7A38A 2019-08-19 10:58 評論0 收藏0
java之BlockingQueue實現(xiàn)生產(chǎn)者消費者

摘要：直接上代碼注意在使用實現(xiàn)生產(chǎn)者消費者模型時候，泛型使用若等對象時候會發(fā)現(xiàn)消費者出現(xiàn)異常，這是由于傳值和傳引用的區(qū)別，而由于的自動裝箱不會出現(xiàn)此類問題，具體可自行嘗試生產(chǎn)者消費者獲取失敗主線程直接上代碼注意在使用blockingqueue實現(xiàn)生產(chǎn)者消費者模型時候，BlockingQueue泛型使用若atomic等對象時候會發(fā)現(xiàn)消費者出現(xiàn)異常，這是由于傳值和傳引用的區(qū)別，而Integer...

李義 2019-08-15 13:44 評論0 收藏0
Java并發(fā)編程之線程間通訊（下）-生產(chǎn)者與消費者

摘要：前文回顧上一篇文章重點嘮叨了中協(xié)調線程間通信的機制，它有力的保證了線程間通信的安全性以及便利性。所以同一時刻廚師線程和服務員線程不會同時在等待隊列中。對于在操作系統(tǒng)中線程的阻塞狀態(tài)，語言中用和這三個狀態(tài)分別表示。前文回顧上一篇文章重點嘮叨了java中協(xié)調線程間通信的wait/notify機制，它有力的保證了線程間通信的安全性以及便利性。本篇將介紹wait/notify機制的一個應用...

lufficc 2019-08-16 16:46 評論0 收藏0