實質上，很多后臺服務程序并發控制的基本原理都可以歸納為生產者/消費者模式，而這是恰恰是在本科操作系統課堂上老師反復講解，而我們卻視而不見不以為然的。在博文《一種面向作業流(工作流)的輕量級可復用的異步流水開發框架的設計與實現》中將介紹一種生產者/消費者模式的具體應用。

生產者消費者問題是研究多線程程序時繞不開的經典問題之一，它描述是有一塊緩沖區作為倉庫，生產者可以將產品放入倉庫，消費者則可以從倉庫中取走產品。解決生產者/消費者問題的方法可分為兩類：（1）采用某種機制保護生產者和消費者之間的同步；（2）在生產者和消費者之間建立一個管道。第一種方式有較高的效率，并且易于實現，代碼的可控制性較好，屬于常用的模式。第二種管道緩沖區不易控制，被傳輸數據對象不易于封裝等，實用性不強。因此本文只介紹同步機制實現的生產者/消費者問題。

同步問題核心在于：如何保證同一資源被多個線程并發訪問時的完整性。常用的同步方法是采用信號或加鎖機制，保證資源在任意時刻至多被一個線程訪問。Java語言在多線程編程上實現了完全對象化，提供了對同步機制的良好支持。在Java中一共有四種方法支持同步，其中前三個是同步方法，一個是管道方法。

（1）wait() / notify()方法

（2）await() / signal()方法

（3）BlockingQueue阻塞隊列方法

（4）PipedInputStream / PipedOutputStream

本文只介紹最常用的前三種，第四種暫不做討論，有興趣的讀者可以自己去網上找答案。

?

一、wait() / notify()方法

wait() / nofity()方法是基類Object的兩個方法，也就意味著所有Java類都會擁有這兩個方法，這樣，我們就可以為任何對象實現同步機制。

wait()方法：當緩沖區已滿/空時，生產者/消費者線程停止自己的執行，放棄鎖，使自己處于等等狀態，讓其他線程執行。

notify()方法：當生產者/消費者向緩沖區放入/取出一個產品時，向其他等待的線程發出可執行的通知，同時放棄鎖，使自己處于等待狀態。

光看文字可能不太好理解，咱來段代碼就明白了：

import java.util.LinkedList;/*** 倉庫類Storage實現緩沖區* * Email:530025983@qq.com* * @author MONKEY.D.MENG 2011-03-15* */ public class Storage {// 倉庫最大存儲量private final int MAX_SIZE = 100;// 倉庫存儲的載體private LinkedList<Object> list = new LinkedList<Object>();// 生產num個產品public void produce(int num){// 同步代碼段synchronized (list){// 如果倉庫剩余容量不足while (list.size() + num > MAX_SIZE){System.out.println("【要生產的產品數量】:" + num + "/t【庫存量】:"+ list.size() + "/t暫時不能執行生產任務!");try{// 由于條件不滿足，生產阻塞 list.wait();}catch (InterruptedException e){e.printStackTrace();}}// 生產條件滿足情況下，生產num個產品for (int i = 1; i <= num; ++i){list.add(new Object());}System.out.println("【已經生產產品數】:" + num + "/t【現倉儲量為】:" + list.size());list.notifyAll();}}// 消費num個產品public void consume(int num){// 同步代碼段synchronized (list){// 如果倉庫存儲量不足while (list.size() < num){System.out.println("【要消費的產品數量】:" + num + "/t【庫存量】:"+ list.size() + "/t暫時不能執行生產任務!");try{// 由于條件不滿足，消費阻塞 list.wait();}catch (InterruptedException e){e.printStackTrace();}}// 消費條件滿足情況下，消費num個產品for (int i = 1; i <= num; ++i){list.remove();}System.out.println("【已經消費產品數】:" + num + "/t【現倉儲量為】:" + list.size());list.notifyAll();}}// get/set方法public LinkedList<Object> getList(){return list;}public void setList(LinkedList<Object> list){this.list = list;}public int getMAX_SIZE(){return MAX_SIZE;} } /*** 生產者類Producer繼承線程類Thread* * Email:530025983@qq.com* * @author MONKEY.D.MENG 2011-03-15* */ public class Producer extends Thread {// 每次生產的產品數量private int num;// 所在放置的倉庫private Storage storage;// 構造函數，設置倉庫public Producer(Storage storage){this.storage = storage;}// 線程run函數public void run(){produce(num);}// 調用倉庫Storage的生產函數public void produce(int num){storage.produce(num);}// get/set方法public int getNum(){return num;}public void setNum(int num){this.num = num;}public Storage getStorage(){return storage;}public void setStorage(Storage storage){this.storage = storage;} } /*** 消費者類Consumer繼承線程類Thread* * Email:530025983@qq.com* * @author MONKEY.D.MENG 2011-03-15* */ public class Consumer extends Thread {// 每次消費的產品數量private int num;// 所在放置的倉庫private Storage storage;// 構造函數，設置倉庫public Consumer(Storage storage){this.storage = storage;}// 線程run函數public void run(){consume(num);}// 調用倉庫Storage的生產函數public void consume(int num){storage.consume(num);}// get/set方法public int getNum(){return num;}public void setNum(int num){this.num = num;}public Storage getStorage(){return storage;}public void setStorage(Storage storage){this.storage = storage;} } /*** 測試類Test* * Email:530025983@qq.com* * @author MONKEY.D.MENG 2011-03-15* */ public class Test {public static void main(String[] args){// 倉庫對象Storage storage = new Storage();// 生產者對象Producer p1 = new Producer(storage);Producer p2 = new Producer(storage);Producer p3 = new Producer(storage);Producer p4 = new Producer(storage);Producer p5 = new Producer(storage);Producer p6 = new Producer(storage);Producer p7 = new Producer(storage);// 消費者對象Consumer c1 = new Consumer(storage);Consumer c2 = new Consumer(storage);Consumer c3 = new Consumer(storage);// 設置生產者產品生產數量p1.setNum(10);p2.setNum(10);p3.setNum(10);p4.setNum(10);p5.setNum(10);p6.setNum(10);p7.setNum(80);// 設置消費者產品消費數量c1.setNum(50);c2.setNum(20);c3.setNum(30);// 線程開始執行 c1.start();c2.start();c3.start();p1.start();p2.start();p3.start();p4.start();p5.start();p6.start();p7.start();} } 【要消費的產品數量】:50 【庫存量】:0 暫時不能執行生產任務! 【要消費的產品數量】:30 【庫存量】:0 暫時不能執行生產任務! 【要消費的產品數量】:20 【庫存量】:0 暫時不能執行生產任務! 【已經生產產品數】:10 【現倉儲量為】:10 【要消費的產品數量】:20 【庫存量】:10 暫時不能執行生產任務! 【要消費的產品數量】:30 【庫存量】:10 暫時不能執行生產任務! 【要消費的產品數量】:50 【庫存量】:10 暫時不能執行生產任務! 【已經生產產品數】:10 【現倉儲量為】:20 【要消費的產品數量】:50 【庫存量】:20 暫時不能執行生產任務! 【要消費的產品數量】:30 【庫存量】:20 暫時不能執行生產任務! 【已經消費產品數】:20 【現倉儲量為】:0 【已經生產產品數】:10 【現倉儲量為】:10 【已經生產產品數】:10 【現倉儲量為】:20 【已經生產產品數】:80 【現倉儲量為】:100 【要生產的產品數量】:10 【庫存量】:100 暫時不能執行生產任務! 【已經消費產品數】:30 【現倉儲量為】:70 【已經消費產品數】:50 【現倉儲量為】:20 【已經生產產品數】:10 【現倉儲量為】:30 【已經生產產品數】:10 【現倉儲量為】:40

看完上述代碼，對wait() / notify()方法實現的同步有了了解。你可能會對Storage類中為什么要定義public void produce(int num);和public void consume(int num);方法感到不解，為什么不直接在生產者類Producer和消費者類Consumer中實現這兩個方法，卻要調用Storage類中的實現呢？淡定，后文會有解釋。我們先往下走。

?

二、await() / signal()方法

在JDK5.0之后，Java提供了更加健壯的線程處理機制，包括同步、鎖定、線程池等，它們可以實現更細粒度的線程控制。await()和signal()就是其中用來做同步的兩種方法，它們的功能基本上和wait() / nofity()相同，完全可以取代它們，但是它們和新引入的鎖定機制Lock直接掛鉤，具有更大的靈活性。通過在Lock對象上調用newCondition()方法，將條件變量和一個鎖對象進行綁定，進而控制并發程序訪問競爭資源的安全。下面來看代碼：

import java.util.LinkedList; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock;/*** 倉庫類Storage實現緩沖區* * Email:530025983@qq.com* * @author MONKEY.D.MENG 2011-03-15* */ 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();// 生產num個產品public void produce(int num){// 獲得鎖 lock.lock();// 如果倉庫剩余容量不足while (list.size() + num > MAX_SIZE){System.out.println("【要生產的產品數量】:" + num + "/t【庫存量】:" + list.size()+ "/t暫時不能執行生產任務!");try{// 由于條件不滿足，生產阻塞 full.await();}catch (InterruptedException e){e.printStackTrace();}}// 生產條件滿足情況下，生產num個產品for (int i = 1; i <= num; ++i){list.add(new Object());}System.out.println("【已經生產產品數】:" + num + "/t【現倉儲量為】:" + list.size());// 喚醒其他所有線程 full.signalAll();empty.signalAll();// 釋放鎖 lock.unlock();}// 消費num個產品public void consume(int num){// 獲得鎖 lock.lock();// 如果倉庫存儲量不足while (list.size() < num){System.out.println("【要消費的產品數量】:" + num + "/t【庫存量】:" + list.size()+ "/t暫時不能執行生產任務!");try{// 由于條件不滿足，消費阻塞 empty.await();}catch (InterruptedException e){e.printStackTrace();}}// 消費條件滿足情況下，消費num個產品for (int i = 1; i <= num; ++i){list.remove();}System.out.println("【已經消費產品數】:" + num + "/t【現倉儲量為】:" + list.size());// 喚醒其他所有線程 full.signalAll();empty.signalAll();// 釋放鎖 lock.unlock();}// set/get方法public int getMAX_SIZE(){return MAX_SIZE;}public LinkedList<Object> getList(){return list;}public void setList(LinkedList<Object> list){this.list = list;} } 【要消費的產品數量】:50 【庫存量】:0 暫時不能執行生產任務! 【要消費的產品數量】:30 【庫存量】:0 暫時不能執行生產任務! 【已經生產產品數】:10 【現倉儲量為】:10 【已經生產產品數】:10 【現倉儲量為】:20 【要消費的產品數量】:50 【庫存量】:20 暫時不能執行生產任務! 【要消費的產品數量】:30 【庫存量】:20 暫時不能執行生產任務! 【已經生產產品數】:10 【現倉儲量為】:30 【要消費的產品數量】:50 【庫存量】:30 暫時不能執行生產任務! 【已經消費產品數】:20 【現倉儲量為】:10 【已經生產產品數】:10 【現倉儲量為】:20 【要消費的產品數量】:30 【庫存量】:20 暫時不能執行生產任務! 【已經生產產品數】:80 【現倉儲量為】:100 【要生產的產品數量】:10 【庫存量】:100 暫時不能執行生產任務! 【已經消費產品數】:50 【現倉儲量為】:50 【已經生產產品數】:10 【現倉儲量為】:60 【已經消費產品數】:30 【現倉儲量為】:30 【已經生產產品數】:10 【現倉儲量為】:40

只需要更新倉庫類Storage的代碼即可，生產者Producer、消費者Consumer、測試類Test的代碼均不需要進行任何更改。這樣我們就知道為神馬我要在Storage類中定義public void produce(int num);和public void consume(int num);方法，并在生產者類Producer和消費者類Consumer中調用Storage類中的實現了吧。將可能發生的變化集中到一個類中，不影響原有的構架設計，同時無需修改其他業務層代碼。無意之中，我們好像使用了某種設計模式，具體是啥我忘記了，啊哈哈，等我想起來再告訴大家~

?

三、BlockingQueue阻塞隊列方法

BlockingQueue是JDK5.0的新增內容，它是一個已經在內部實現了同步的隊列，實現方式采用的是我們第2種await() / signal()方法。它可以在生成對象時指定容量大小。它用于阻塞操作的是put()和take()方法。

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

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

關于BlockingQueue的內容網上有很多，大家可以自己搜，我在這不多介紹。下面直接看代碼，跟以往一樣，我們只需要更改倉庫類Storage的代碼即可：

import java.util.concurrent.LinkedBlockingQueue;/*** 倉庫類Storage實現緩沖區* * Email:530025983@qq.com* * @author MONKEY.D.MENG 2011-03-15* */ public class Storage {// 倉庫最大存儲量private final int MAX_SIZE = 100;// 倉庫存儲的載體private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(100);// 生產num個產品public void produce(int num){// 如果倉庫剩余容量為0if (list.size() == MAX_SIZE){System.out.println("【庫存量】:" + MAX_SIZE + "/t暫時不能執行生產任務!");}// 生產條件滿足情況下，生產num個產品for (int i = 1; i <= num; ++i){try{// 放入產品，自動阻塞list.put(new Object());}catch (InterruptedException e){e.printStackTrace();}System.out.println("【現倉儲量為】:" + list.size());}}// 消費num個產品public void consume(int num){// 如果倉庫存儲量不足if (list.size() == 0){System.out.println("【庫存量】:0/t暫時不能執行生產任務!");}// 消費條件滿足情況下，消費num個產品for (int i = 1; i <= num; ++i){try{// 消費產品，自動阻塞 list.take();}catch (InterruptedException e){e.printStackTrace();}}System.out.println("【現倉儲量為】:" + list.size());}// set/get方法public LinkedBlockingQueue<Object> getList(){return list;}public void setList(LinkedBlockingQueue<Object> list){this.list = list;}public int getMAX_SIZE(){return MAX_SIZE;} } 【庫存量】:0 暫時不能執行生產任務! 【庫存量】:0 暫時不能執行生產任務! 【現倉儲量為】:1 【現倉儲量為】:1 【現倉儲量為】:3 【現倉儲量為】:4 【現倉儲量為】:5 【現倉儲量為】:6 【現倉儲量為】:7 【現倉儲量為】:8 【現倉儲量為】:9 【現倉儲量為】:10 【現倉儲量為】:11 【現倉儲量為】:1 【現倉儲量為】:2 【現倉儲量為】:13 【現倉儲量為】:14 【現倉儲量為】:17 【現倉儲量為】:19 【現倉儲量為】:20 【現倉儲量為】:21 【現倉儲量為】:22 【現倉儲量為】:23 【現倉儲量為】:24 【現倉儲量為】:25 【現倉儲量為】:26 【現倉儲量為】:12 【現倉儲量為】:1 【現倉儲量為】:1 【現倉儲量為】:2 【現倉儲量為】:3 【現倉儲量為】:4 【現倉儲量為】:5 【現倉儲量為】:6 【現倉儲量為】:7 【現倉儲量為】:27 【現倉儲量為】:8 【現倉儲量為】:6 【現倉儲量為】:18 【現倉儲量為】:2 【現倉儲量為】:3 【現倉儲量為】:4 【現倉儲量為】:5 【現倉儲量為】:6 【現倉儲量為】:7 【現倉儲量為】:8 【現倉儲量為】:9 【現倉儲量為】:10 【現倉儲量為】:16 【現倉儲量為】:11 【現倉儲量為】:12 【現倉儲量為】:13 【現倉儲量為】:14 【現倉儲量為】:15 【現倉儲量為】:1 【現倉儲量為】:2 【現倉儲量為】:3 【現倉儲量為】:3 【現倉儲量為】:15 【現倉儲量為】:1 【現倉儲量為】:0 【現倉儲量為】:1 【現倉儲量為】:1 【現倉儲量為】:1 【現倉儲量為】:2 【現倉儲量為】:3 【現倉儲量為】:4 【現倉儲量為】:0 【現倉儲量為】:1 【現倉儲量為】:5 【現倉儲量為】:6 【現倉儲量為】:7 【現倉儲量為】:8 【現倉儲量為】:9 【現倉儲量為】:10 【現倉儲量為】:11 【現倉儲量為】:12 【現倉儲量為】:13 【現倉儲量為】:14 【現倉儲量為】:15 【現倉儲量為】:16 【現倉儲量為】:17 【現倉儲量為】:1 【現倉儲量為】:1 【現倉儲量為】:2 【現倉儲量為】:3 【現倉儲量為】:4 【現倉儲量為】:5 【現倉儲量為】:6 【現倉儲量為】:3 【現倉儲量為】:3 【現倉儲量為】:1 【現倉儲量為】:2 【現倉儲量為】:3 【現倉儲量為】:4 【現倉儲量為】:5 【現倉儲量為】:6 【現倉儲量為】:7 【現倉儲量為】:8 【現倉儲量為】:9 【現倉儲量為】:10 【現倉儲量為】:11 【現倉儲量為】:12 【現倉儲量為】:13 【現倉儲量為】:14 【現倉儲量為】:15 【現倉儲量為】:16 【現倉儲量為】:17 【現倉儲量為】:18 【現倉儲量為】:19 【現倉儲量為】:6 【現倉儲量為】:7 【現倉儲量為】:8 【現倉儲量為】:9 【現倉儲量為】:10 【現倉儲量為】:11 【現倉儲量為】:12 【現倉儲量為】:13 【現倉儲量為】:14 【現倉儲量為】:15 【現倉儲量為】:16 【現倉儲量為】:17 【現倉儲量為】:18 【現倉儲量為】:19 【現倉儲量為】:20 【現倉儲量為】:21 【現倉儲量為】:22 【現倉儲量為】:23 【現倉儲量為】:24 【現倉儲量為】:25 【現倉儲量為】:26 【現倉儲量為】:27 【現倉儲量為】:28 【現倉儲量為】:29 【現倉儲量為】:30 【現倉儲量為】:31 【現倉儲量為】:32 【現倉儲量為】:33 【現倉儲量為】:34 【現倉儲量為】:35 【現倉儲量為】:36 【現倉儲量為】:37 【現倉儲量為】:38 【現倉儲量為】:39 【現倉儲量為】:40

當然，你會發現這時對于public void produce(int num);和public void consume(int num);方法業務邏輯上的實現跟前面兩個例子不太一樣，沒關系，這個例子只是為了說明BlockingQueue阻塞隊列的使用。

有時使用BlockingQueue可能會出現put()和System.out.println()輸出不匹配的情況，這是由于它們之間沒有同步造成的。當緩沖區已滿，生產者在put()操作時，put()內部調用了await()方法，放棄了線程的執行，然后消費者線程執行，調用take()方法，take()內部調用了signal()方法，通知生產者線程可以執行，致使在消費者的println()還沒運行的情況下生產者的println()先被執行，所以有了輸出不匹配的情況。

對于BlockingQueue大家可以放心使用，這可不是它的問題，只是在它和別的對象之間的同步有問題。

對于Java實現生產者/消費者問題的方法先總結到這里面吧，過幾天實現一下C++版本的，接下來要馬上著手于基于生產者/消費者模式的《異步工作流服務框架的設計與實現》，請持續關注本博客。

轉自：http://blog.csdn.net/monkey_d_meng/article/details/6251879

?

?

?

?

轉載于:https://www.cnblogs.com/davidwang456/p/3705593.html

總結

以上是生活随笔為你收集整理的生产者/消费者问题的多种Java实现方式--转的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。