? ? ? ? 生產(chǎn)者和消費者問題是線程模型中的經(jīng)典問題，生產(chǎn)者和消費者在同一時間段共用同一個存儲空間，這個存儲空間是一個緩沖區(qū)的倉庫，生產(chǎn)者可以將產(chǎn)品放入倉庫，消費者可以從倉庫中取出產(chǎn)品。

? ? ? ?

?生產(chǎn)者/消費者模型是基于等待/通知機制，主要關(guān)注以下幾點:

生產(chǎn)者生產(chǎn)的時候消費者不能消費

消費者消費的時候生產(chǎn)者不能生產(chǎn)

緩沖區(qū)空時消費者不能消費

緩沖區(qū)滿時生產(chǎn)者不能生產(chǎn)

主要優(yōu)點:

• 解耦。因為多了一個緩沖區(qū)，所以生產(chǎn)者和消費者并不直接相互調(diào)用，這一點很容易想到，這樣生產(chǎn)者和消費者的代碼發(fā)生變化，都不會對對方產(chǎn)生影響，這樣其實就把生產(chǎn)者和消費者之間的強耦合解開，變?yōu)榱松a(chǎn)者和緩沖區(qū)/消費者和緩沖區(qū)之間的弱耦合
• 通過平衡生產(chǎn)者和消費者的處理能力來提高整體處理數(shù)據(jù)的速度，這是生產(chǎn)者/消費者模型最重要的一個優(yōu)點。如果消費者直接從生產(chǎn)者這里拿數(shù)據(jù)，如果生產(chǎn)者生產(chǎn)的速度很慢，但消費者消費的速度很快，那消費者就得占用CPU的時間片白白等在那邊。有了生產(chǎn)者/消費者模型，生產(chǎn)者和消費者就是兩個獨立的并發(fā)體，生產(chǎn)者把生產(chǎn)出來的數(shù)據(jù)往緩沖區(qū)一丟就好了，不必管消費者；消費者也是，從緩沖區(qū)去拿數(shù)據(jù)就好了，也不必管生產(chǎn)者，緩沖區(qū)滿了就不生產(chǎn)，緩沖區(qū)空了就不消費，使生產(chǎn)者/消費者的處理能力達到一個動態(tài)的平衡

實現(xiàn)生產(chǎn)者和消費者的5種方式?

wait()和notify()方法的實現(xiàn):

這也是最簡單最基礎(chǔ)的實現(xiàn)，緩沖區(qū)滿和為空時都調(diào)用wait()方法等待，當(dāng)生產(chǎn)者生產(chǎn)了一個產(chǎn)品或者消費者消費了一個產(chǎn)品之后會喚醒所有線程。

package org.example;import java.sql.SQLOutput;public class Test {private static Integer count = 0;private static final Integer Total = 10;private static String flag = "agree";public static void main(String[] args) {Test test = new Test();new Thread(test.new Producer()).start();new Thread(test.new Consumer()).start();new Thread(test.new Producer()).start();new Thread(test.new Consumer()).start();new Thread(test.new Producer()).start();new Thread(test.new Consumer()).start();new Thread(test.new Producer()).start();new Thread(test.new Consumer()).start();}public class Producer implements Runnable{@Overridepublic void run() {for (int i = 0;i < 5; i++){try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}synchronized (flag) {while (count == Total) {try {flag.wait();} catch (InterruptedException e) {e.printStackTrace();}}count++;System.out.println(Thread.currentThread().getName()+"生產(chǎn)者共有" + count);flag.notifyAll();}}}}public class Consumer implements Runnable{@Overridepublic void run() {for(int i = 0; i < 5; i++){try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}synchronized (flag){while(count == 0){try {flag.wait();} catch (InterruptedException e) {e.printStackTrace();}}count--;System.out.println(Thread.currentThread().getName() + "消費者消費，目前總共有" + count);flag.notifyAll();}}}} }

運行結(jié)果:

Thread-6生產(chǎn)者共有1 Thread-1消費者消費，目前總共有0 Thread-0生產(chǎn)者共有1 Thread-4生產(chǎn)者共有2 Thread-5消費者消費，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-2生產(chǎn)者共有1 Thread-3消費者消費，目前總共有0 Thread-4生產(chǎn)者共有1 Thread-6生產(chǎn)者共有2 Thread-7消費者消費，目前總共有1 Thread-0生產(chǎn)者共有2 Thread-1消費者消費，目前總共有1 Thread-5消費者消費，目前總共有0 Thread-2生產(chǎn)者共有1 Thread-3消費者消費，目前總共有0 Thread-4生產(chǎn)者共有1 Thread-5消費者消費，目前總共有0 Thread-2生產(chǎn)者共有1 Thread-3消費者消費，目前總共有0 Thread-6生產(chǎn)者共有1 Thread-7消費者消費，目前總共有0 Thread-0生產(chǎn)者共有1 Thread-1消費者消費，目前總共有0 Thread-6生產(chǎn)者共有1 Thread-3消費者消費，目前總共有0 Thread-4生產(chǎn)者共有1 Thread-0生產(chǎn)者共有2 Thread-1消費者消費，目前總共有1 Thread-2生產(chǎn)者共有2 Thread-5消費者消費，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-0生產(chǎn)者共有1 Thread-3消費者消費，目前總共有0 Thread-6生產(chǎn)者共有1 Thread-7消費者消費，目前總共有0 Thread-4生產(chǎn)者共有1 Thread-2生產(chǎn)者共有2 Thread-5消費者消費，目前總共有1 Thread-1消費者消費，目前總共有0Process finished with exit code 0

可重入鎖ReentrantLock的實現(xiàn)

java.util.concurrent.lock 中的 Lock 框架是鎖定的一個抽象，通過對lock的lock()方法和unlock()方法實現(xiàn)了對鎖的顯示控制，而synchronize()則是對鎖的隱性控制。
可重入鎖，也叫做遞歸鎖，指的是同一線程 外層函數(shù)獲得鎖之后 ，內(nèi)層遞歸函數(shù)仍然有獲取該鎖的代碼，但不受影響，簡單來說，該鎖維護這一個與獲取鎖相關(guān)的計數(shù)器，如果擁有鎖的某個線程再次得到鎖，那么獲取計數(shù)器就加1，函數(shù)調(diào)用結(jié)束計數(shù)器就減1，然后鎖需要被釋放兩次才能獲得真正釋放。已經(jīng)獲取鎖的線程進入其他需要相同鎖的同步代碼塊不會被阻塞。

import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /*** 生產(chǎn)者和消費者，ReentrantLock的實現(xiàn)*/ public class Test{private static Integer count = 0;private static final Integer FULL = 10;//創(chuàng)建一個鎖對象private Lock lock = new ReentrantLock();//創(chuàng)建兩個條件變量，一個為緩沖區(qū)非滿，一個為緩沖區(qū)非空private final Condition notFull = lock.newCondition();private final Condition notEmpty = lock.newCondition();public static void main(String[] args) {Test test2 = new Test();new Thread(test2.new Producer()).start();new Thread(test2.new Consumer()).start();new Thread(test2.new Producer()).start();new Thread(test2.new Consumer()).start();new Thread(test2.new Producer()).start();new Thread(test2.new Consumer()).start();new Thread(test2.new Producer()).start();new Thread(test2.new Consumer()).start();}class Producer implements Runnable {@Overridepublic void run() {for (int i = 0; i < 10; i++) {try {Thread.sleep(3000);} catch (Exception e) {e.printStackTrace();}//獲取鎖lock.lock();try {while (count == FULL) {try {notFull.await();} catch (InterruptedException e) {e.printStackTrace();}}count++;System.out.println(Thread.currentThread().getName()+ "生產(chǎn)者生產(chǎn)，目前總共有" + count);//喚醒消費者notEmpty.signal();} finally {//釋放鎖lock.unlock();}}}}class Consumer implements Runnable {@Overridepublic void run() {for (int i = 0; i < 10; i++) {try {Thread.sleep(3000);} catch (InterruptedException e1) {e1.printStackTrace();}lock.lock();try {while (count == 0) {try {notEmpty.await();} catch (Exception e) {e.printStackTrace();}}count--;System.out.println(Thread.currentThread().getName()+ "消費者消費，目前總共有" + count);notFull.signal();} finally {lock.unlock();}}}} }

?運行結(jié)果

Thread-2生產(chǎn)者生產(chǎn)，目前總共有1 Thread-6生產(chǎn)者生產(chǎn)，目前總共有2 Thread-4生產(chǎn)者生產(chǎn)，目前總共有3 Thread-5消費者消費，目前總共有2 Thread-1消費者消費，目前總共有1 Thread-0生產(chǎn)者生產(chǎn)，目前總共有2 Thread-3消費者消費，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-4生產(chǎn)者生產(chǎn)，目前總共有1 Thread-3消費者消費，目前總共有0 Thread-6生產(chǎn)者生產(chǎn)，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-0生產(chǎn)者生產(chǎn)，目前總共有1 Thread-5消費者消費，目前總共有0 Thread-2生產(chǎn)者生產(chǎn)，目前總共有1 Thread-1消費者消費，目前總共有0 Thread-4生產(chǎn)者生產(chǎn)，目前總共有1 Thread-1消費者消費，目前總共有0 Thread-6生產(chǎn)者生產(chǎn)，目前總共有1 Thread-5消費者消費，目前總共有0 Thread-0生產(chǎn)者生產(chǎn)，目前總共有1 Thread-2生產(chǎn)者生產(chǎn)，目前總共有2 Thread-3消費者消費，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-4生產(chǎn)者生產(chǎn)，目前總共有1 Thread-5消費者消費，目前總共有0 Thread-0生產(chǎn)者生產(chǎn)，目前總共有1 Thread-6生產(chǎn)者生產(chǎn)，目前總共有2 Thread-3消費者消費，目前總共有1 Thread-2生產(chǎn)者生產(chǎn)，目前總共有2 Thread-1消費者消費，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-6生產(chǎn)者生產(chǎn)，目前總共有1 Thread-3消費者消費，目前總共有0 Thread-4生產(chǎn)者生產(chǎn)，目前總共有1 Thread-0生產(chǎn)者生產(chǎn)，目前總共有2 Thread-2生產(chǎn)者生產(chǎn)，目前總共有3 Thread-1消費者消費，目前總共有2 Thread-7消費者消費，目前總共有1 Thread-5消費者消費，目前總共有0 Thread-4生產(chǎn)者生產(chǎn)，目前總共有1 Thread-6生產(chǎn)者生產(chǎn)，目前總共有2 Thread-3消費者消費，目前總共有1 Thread-2生產(chǎn)者生產(chǎn)，目前總共有2 Thread-7消費者消費，目前總共有1 Thread-0生產(chǎn)者生產(chǎn)，目前總共有2 Thread-5消費者消費，目前總共有1 Thread-1消費者消費，目前總共有0 Thread-6生產(chǎn)者生產(chǎn)，目前總共有1 Thread-4生產(chǎn)者生產(chǎn)，目前總共有2 Thread-3消費者消費，目前總共有1 Thread-2生產(chǎn)者生產(chǎn)，目前總共有2 Thread-5消費者消費，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-0生產(chǎn)者生產(chǎn)，目前總共有1 Thread-1消費者消費，目前總共有0 Thread-6生產(chǎn)者生產(chǎn)，目前總共有1 Thread-4生產(chǎn)者生產(chǎn)，目前總共有2 Thread-3消費者消費，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-0生產(chǎn)者生產(chǎn)，目前總共有1 Thread-2生產(chǎn)者生產(chǎn)，目前總共有2 Thread-5消費者消費，目前總共有1 Thread-1消費者消費，目前總共有0 Thread-6生產(chǎn)者生產(chǎn)，目前總共有1 Thread-3消費者消費，目前總共有0 Thread-4生產(chǎn)者生產(chǎn)，目前總共有1 Thread-1消費者消費，目前總共有0 Thread-0生產(chǎn)者生產(chǎn)，目前總共有1 Thread-5消費者消費，目前總共有0 Thread-2生產(chǎn)者生產(chǎn)，目前總共有1 Thread-7消費者消費，目前總共有0 Thread-4生產(chǎn)者生產(chǎn)，目前總共有1 Thread-6生產(chǎn)者生產(chǎn)，目前總共有2 Thread-3消費者消費，目前總共有1 Thread-2生產(chǎn)者生產(chǎn)，目前總共有2 Thread-0生產(chǎn)者生產(chǎn)，目前總共有3 Thread-5消費者消費，目前總共有2 Thread-7消費者消費，目前總共有1 Thread-1消費者消費，目前總共有0Process finished with exit code 0

阻塞隊列BlockingQueue的實現(xiàn)

BlockingQueue即阻塞隊列，從阻塞這個詞可以看出，在某些情況下對阻塞隊列的訪問可能會造成阻塞。被阻塞的情況主要有如下兩種:

當(dāng)隊列滿了的時候進行入隊列操作

當(dāng)隊列空了的時候進行出隊列操作
因此，當(dāng)一個線程對已經(jīng)滿了的阻塞隊列進行入隊操作時會阻塞，除非有另外一個線程進行了出隊操作，當(dāng)一個線程對一個空的阻塞隊列進行出隊操作時也會阻塞，除非有另外一個線程進行了入隊操作。
從上可知，阻塞隊列是線程安全的。
下面是BlockingQueue接口的一些方法:

import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; /*** 使用BlockingQueue實現(xiàn)生產(chǎn)者消費者模型*/ public class Test3 {private static Integer count = 0;//創(chuàng)建一個阻塞隊列final BlockingQueue blockingQueue = new ArrayBlockingQueue<>(10);public static void main(String[] args) {Test3 test3 = new Test3();new Thread(test3.new Producer()).start();new Thread(test3.new Consumer()).start();new Thread(test3.new Producer()).start();new Thread(test3.new Consumer()).start();new Thread(test3.new Producer()).start();new Thread(test3.new Consumer()).start();new Thread(test3.new Producer()).start();new Thread(test3.new Consumer()).start();}class Producer implements Runnable {@Overridepublic void run() {for (int i = 0; i < 10; i++) {try {Thread.sleep(3000);} catch (Exception e) {e.printStackTrace();}try {blockingQueue.put(1);count++;System.out.println(Thread.currentThread().getName()+ "生產(chǎn)者生產(chǎn)，目前總共有" + count);} catch (InterruptedException e) {e.printStackTrace();}}}}class Consumer implements Runnable {@Overridepublic void run() {for (int i = 0; i < 10; i++) {try {Thread.sleep(3000);} catch (InterruptedException e1) {e1.printStackTrace();}try {blockingQueue.take();count--;System.out.println(Thread.currentThread().getName()+ "消費者消費，目前總共有" + count);} catch (InterruptedException e) {e.printStackTrace();}}}} }

?

信號量Semaphore的實現(xiàn)?

Semaphore（信號量）是用來控制同時訪問特定資源的線程數(shù)量，它通過協(xié)調(diào)各個線程，以保證合理的使用公共資源，在操作系統(tǒng)中是一個非常重要的問題，可以用來解決哲學(xué)家就餐問題。Java中的Semaphore維護了一個許可集，一開始先設(shè)定這個許可集的數(shù)量，可以使用acquire()方法獲得一個許可，當(dāng)許可不足時會被阻塞，release()添加一個許可。在下列代碼中，還加入了另外一個mutex信號量，維護生產(chǎn)者消費者之間的同步關(guān)系，保證生產(chǎn)者和消費者之間的交替進行

import java.util.concurrent.Semaphore; /*** 使用semaphore信號量實現(xiàn)*/ public class Test4 {private static Integer count = 0;//創(chuàng)建三個信號量final Semaphore notFull = new Semaphore(10);final Semaphore notEmpty = new Semaphore(0);final Semaphore mutex = new Semaphore(1);public static void main(String[] args) {Test4 test4 = new Test4();new Thread(test4.new Producer()).start();new Thread(test4.new Consumer()).start();new Thread(test4.new Producer()).start();new Thread(test4.new Consumer()).start();new Thread(test4.new Producer()).start();new Thread(test4.new Consumer()).start();new Thread(test4.new Producer()).start();new Thread(test4.new Consumer()).start();}class Producer implements Runnable {@Overridepublic void run() {for (int i = 0; i < 10; i++) {try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}try {notFull.acquire();mutex.acquire();count++;System.out.println(Thread.currentThread().getName()+ "生產(chǎn)者生產(chǎn)，目前總共有" + count);} catch (InterruptedException e) {e.printStackTrace();} finally {mutex.release();notEmpty.release();}}}}class Consumer implements Runnable {@Overridepublic void run() {for (int i = 0; i < 10; i++) {try {Thread.sleep(3000);} catch (InterruptedException e1) {e1.printStackTrace();}try {notEmpty.acquire();mutex.acquire();count--;System.out.println(Thread.currentThread().getName()+ "消費者消費，目前總共有" + count);} catch (InterruptedException e) {e.printStackTrace();} finally {mutex.release();notFull.release();}}}} }

管道輸入輸出流PipedInputStream和PipedOutputStream實現(xiàn)

?

在java的io包下，PipedOutputStream和PipedInputStream分別是管道輸出流和管道輸入流。
它們的作用是讓多線程可以通過管道進行線程間的通訊。在使用管道通信時，必須將PipedOutputStream和PipedInputStream配套使用。
使用方法：先創(chuàng)建一個管道輸入流和管道輸出流，然后將輸入流和輸出流進行連接，用生產(chǎn)者線程往管道輸出流中寫入數(shù)據(jù)，消費者在管道輸入流中讀取數(shù)據(jù)，這樣就可以實現(xiàn)了不同線程間的相互通訊，但是這種方式在生產(chǎn)者和生產(chǎn)者、消費者和消費者之間不能保證同步，也就是說在一個生產(chǎn)者和一個消費者的情況下是可以生產(chǎn)者和消費者之間交替運行的，多個生成者和多個消費者者之間則不行

/*** 使用管道實現(xiàn)生產(chǎn)者消費者模型*/ public class Test5 {final PipedInputStream pis = new PipedInputStream();final PipedOutputStream pos = new PipedOutputStream();{try {pis.connect(pos);} catch (IOException e) {e.printStackTrace();}}class Producer implements Runnable {@Overridepublic void run() {try {while(true) {Thread.sleep(1000);int num = (int) (Math.random() * 255);System.out.println(Thread.currentThread().getName() + "生產(chǎn)者生產(chǎn)了一個數(shù)字，該數(shù)字為： " + num);pos.write(num);pos.flush();} } catch (Exception e) {e.printStackTrace();} finally {try {pos.close();pis.close();} catch (IOException e) {e.printStackTrace();}}}}class Consumer implements Runnable {@Overridepublic void run() {try {while(true) {Thread.sleep(1000);int num = pis.read();System.out.println("消費者消費了一個數(shù)字，該數(shù)字為：" + num);}} catch (Exception e) {e.printStackTrace();} finally {try {pos.close();pis.close();} catch (IOException e) {e.printStackTrace();}}}}public static void main(String[] args) {Test5 test5 = new Test5();new Thread(test5.new Producer()).start();new Thread(test5.new Consumer()).start();} }

參考

《Thinking In Java》

Java實現(xiàn)生產(chǎn)者和消費者的5種方式

總結(jié)

以上是生活随笔為你收集整理的Java中生产者和消费者总结的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。