生產(chǎn)者-消費者Java實現(xiàn)

2017-07-27

1 概述

---

?

生產(chǎn)者消費者問題是多線程的一個經(jīng)典問題，它描述是有一塊緩沖區(qū)作為倉庫，生產(chǎn)者可以將產(chǎn)品放入倉庫，消費者則可以從倉庫中取走產(chǎn)品。

解決生產(chǎn)者/消費者問題的方法可分為兩類：

• 采用某種機制保護生產(chǎn)者和消費者之間的同步；
• 在生產(chǎn)者和消費者之間建立一個管道。

第一種方式有較高的效率，并且易于實現(xiàn)，代碼的可控制性較好，屬于常用的模式。第二種管道緩沖區(qū)不易控制，被傳輸數(shù)據(jù)對象不易于封裝等，實用性不強。

在Java中有四種方法支持同步，其中前三個是同步方法，一個是管道方法。

• wait() / notify()方法
• await() / signal()方法
• BlockingQueue阻塞隊列方法
• PipedInputStream / PipedOutputStream

本文只介紹前三種。

2 實現(xiàn)

---

?

2.1 wait() / notify()方法

wait() / nofity()方法是基類Object的兩個方法：

• wait()方法：當(dāng)緩沖區(qū)已滿/空時，生產(chǎn)者/消費者線程停止自己的執(zhí)行，放棄鎖，使自己處于等等狀態(tài)，讓其他線程執(zhí)行。
• notify()方法：當(dāng)生產(chǎn)者/消費者向緩沖區(qū)放入/取出一個產(chǎn)品時，向其他等待的線程發(fā)出可執(zhí)行的通知，同時放棄鎖，使自己處于等待狀態(tài)。

緩沖區(qū)Storage.java代碼如下：

import java.util.LinkedList;public class Storage {// 倉庫最大存儲量private final int MAX_SIZE = 100;// 倉庫存儲的載體private LinkedList<Object> list = new LinkedList<Object>();// 生產(chǎn)產(chǎn)品public void produce(String producer){synchronized (list){// 如果倉庫已滿while (list.size() == MAX_SIZE){System.out.println("倉庫已滿，【"+producer+"】： 暫時不能執(zhí)行生產(chǎn)任務(wù)!");try{// 由于條件不滿足，生產(chǎn)阻塞 list.wait();}catch (InterruptedException e){e.printStackTrace();}}// 生產(chǎn)產(chǎn)品 list.add(new Object()); System.out.println("【"+producer+"】：生產(chǎn)了一個產(chǎn)品\t【現(xiàn)倉儲量為】:" + list.size());list.notifyAll();}}// 消費產(chǎn)品public void consume(String consumer){synchronized (list){//如果倉庫存儲量不足while (list.size()==0){System.out.println("倉庫已空，【"+consumer+"】： 暫時不能執(zhí)行消費任務(wù)!");try{// 由于條件不滿足，消費阻塞 list.wait();}catch (InterruptedException e){e.printStackTrace();}}list.remove();System.out.println("【"+consumer+"】：消費了一個產(chǎn)品\t【現(xiàn)倉儲量為】:" + list.size());list.notifyAll();}}public LinkedList<Object> getList(){return list;}public void setList(LinkedList<Object> list){this.list = list;}public int getMAX_SIZE(){return MAX_SIZE;} } View Code

Test.java

public class Test {public static void main(String[] args){Storage storage=new Storage();for(int i=1;i<6;i++){int finalI = i;new Thread(new Runnable() {@Overridepublic void run() {storage.produce(String.format("生成者%d:", finalI));}}).start();}for(int i=1;i<4;i++){int finalI = i;new Thread(()-> storage.consume(String.format("消費者%d:", finalI))).start();}} } View Code

結(jié)果如下：

倉庫已空，【消費者1】： 暫時不能執(zhí)行消費任務(wù)! 【生產(chǎn)者3】：生產(chǎn)了一個產(chǎn)品 【現(xiàn)倉儲量為】:1 【消費者2】：消費了一個產(chǎn)品 【現(xiàn)倉儲量為】:0 倉庫已空，【消費者3】： 暫時不能執(zhí)行消費任務(wù)! 【生產(chǎn)者1】：生產(chǎn)了一個產(chǎn)品 【現(xiàn)倉儲量為】:1 【生產(chǎn)者4】：生產(chǎn)了一個產(chǎn)品 【現(xiàn)倉儲量為】:2 【生產(chǎn)者2】：生產(chǎn)了一個產(chǎn)品 【現(xiàn)倉儲量為】:3 【生產(chǎn)者5】：生產(chǎn)了一個產(chǎn)品 【現(xiàn)倉儲量為】:4 【消費者1】：消費了一個產(chǎn)品 【現(xiàn)倉儲量為】:3 【消費者3】：消費了一個產(chǎn)品 【現(xiàn)倉儲量為】:2

2.2 await() / signal()方法

await()和signal()的功能基本上和wait() / nofity()相同，完全可以取代它們，但是它們和新引入的鎖定機制Lock直接掛鉤，具有更大的靈活性。通過在Lock對象上調(diào)用newCondition()方法，將條件變量和一個鎖對象進行綁定，進而控制并發(fā)程序訪問競爭資源的安全。

緩沖區(qū)Storage.java代碼如下：

import java.util.LinkedList; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock;public class Storage {// 倉庫最大存儲量private final int MAX_SIZE = 100;// 倉庫存儲的載體private LinkedList<Object> list = new LinkedList<Object>();// 鎖private final Lock lock = new ReentrantLock();// 倉庫滿的條件變量private final Condition full = lock.newCondition();// 倉庫空的條件變量private final Condition empty = lock.newCondition();// 生產(chǎn)產(chǎn)品public void produce(String producer) {lock.lock();// 如果倉庫已滿while (list.size() == MAX_SIZE) {System.out.println("倉庫已滿，【" + producer + "】： 暫時不能執(zhí)行生產(chǎn)任務(wù)!");try {// 由于條件不滿足，生產(chǎn)阻塞 full.await();} catch (InterruptedException e) {e.printStackTrace();}}// 生產(chǎn)產(chǎn)品list.add(new Object());System.out.println("【" + producer + "】：生產(chǎn)了一個產(chǎn)品\t【現(xiàn)倉儲量為】:" + list.size());empty.signalAll();// 釋放鎖 lock.unlock();}// 消費產(chǎn)品public void consume(String consumer) {// 獲得鎖 lock.lock();// 如果倉庫存儲量不足while (list.size() == 0) {System.out.println("倉庫已空，【" + consumer + "】： 暫時不能執(zhí)行消費任務(wù)!");try {// 由于條件不滿足，消費阻塞 empty.await();} catch (InterruptedException e) {e.printStackTrace();}}list.remove();System.out.println("【" + consumer + "】：消費了一個產(chǎn)品\t【現(xiàn)倉儲量為】:" + list.size());full.signalAll();// 釋放鎖 lock.unlock();}public LinkedList<Object> getList() {return list;}public void setList(LinkedList<Object> list) {this.list = list;}public int getMAX_SIZE() {return MAX_SIZE;} } View Code

2.3 BlockingQueue

它是一個已經(jīng)在內(nèi)部實現(xiàn)了同步的隊列，實現(xiàn)方式采用的是我們第2種await() / signal()方法。它可以在生成對象時指定容量大小。它用于阻塞操作的是put()和take()方法：

put()方法：類似于我們上面的生產(chǎn)者線程，容量達到最大時，自動阻塞。

take()方法：類似于我們上面的消費者線程，容量為0時，自動阻塞。

import java.util.concurrent.LinkedBlockingQueue;public class Storage {// 倉庫最大存儲量private final int MAX_SIZE = 100;// 倉庫存儲的載體private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(100); // 生產(chǎn)產(chǎn)品public void produce(String producer) {// 如果倉庫已滿if (list.size() == MAX_SIZE) {System.out.println("倉庫已滿，【" + producer + "】： 暫時不能執(zhí)行生產(chǎn)任務(wù)!"); }// 生產(chǎn)產(chǎn)品try {list.put(new Object());} catch (InterruptedException e) {// TODO Auto-generated catch block e.printStackTrace();}System.out.println("【" + producer + "】：生產(chǎn)了一個產(chǎn)品\t【現(xiàn)倉儲量為】:" + list.size());}// 消費產(chǎn)品public void consume(String consumer) {// 如果倉庫存儲量不足if (list.size() == 0) {System.out.println("倉庫已空，【" + consumer + "】： 暫時不能執(zhí)行消費任務(wù)!"); }try {list.take();} catch (InterruptedException e) {// TODO Auto-generated catch block e.printStackTrace();}System.out.println("【" + consumer + "】：消費了一個產(chǎn)品\t【現(xiàn)倉儲量為】:" + list.size()); }public LinkedBlockingQueue<Object> getList() {return list;}public void setList(LinkedBlockingQueue<Object> list) {this.list = list;}public int getMAX_SIZE() {return MAX_SIZE;} } View Code

?

轉(zhuǎn)載于:https://www.cnblogs.com/Ming8006/p/7243858.html

總結(jié)

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

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