生産者消費者問題是研究多線程程式時繞不開的經典問題之一，它描述是有一塊緩沖區作為倉庫，生産者可以将産品放入倉庫，消費者則可以從倉庫中取走産品。解決生産者/消費者問題的方法可分為兩類：（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:[email protected]
* @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)
// 如果倉庫存儲量不足
while (list.size() < num)
System.out.println("【要消費的産品數量】:" + num + "/t【庫存量】:"
// 由于條件不滿足，消費阻塞
// 消費條件滿足情況下，消費num個産品
list.remove();
System.out.println("【已經消費産品數】:" + num + "/t【現倉儲量為】:" + list.size());
// 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
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的生産函數
storage.produce(num);
public int getNum()
return num;
public void setNum(int num)
this.num = num;
public Storage getStorage()
return storage;
public void setStorage(Storage storage)
* 消費者類Consumer繼承線程類Thread
public class Consumer extends Thread
// 每次消費的産品數量
public Consumer(Storage storage)
consume(num);
storage.consume(num);
* 測試類Test
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
【已經生産産品數】: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.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
// 鎖
private final Lock lock = new ReentrantLock();
// 倉庫滿的條件變量
private final Condition full = lock.newCondition();
// 倉庫空的條件變量
private final Condition empty = lock.newCondition();
// 獲得鎖
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();
// 如果倉庫存儲量不足
while (list.size() < num)
System.out.println("【要消費的産品數量】:" + num + "/t【庫存量】:" + list.size()
// 由于條件不滿足，消費阻塞
empty.await();
// 消費條件滿足情況下，消費num個産品
list.remove();
System.out.println("【已經消費産品數】:" + num + "/t【現倉儲量為】:" + list.size());
// set/get方法
【要消費的産品數量】:50 【庫存量】:30 暫時不能執行生産任務!
【已經消費産品數】:20 【現倉儲量為】:10
【已經消費産品數】:50 【現倉儲量為】:50
【已經生産産品數】:10 【現倉儲量為】:60
【已經消費産品數】:30 【現倉儲量為】:30
隻需要更新倉庫類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;
private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(
100);
// 如果倉庫剩餘容量為0
if (list.size() == MAX_SIZE)
System.out.println("【庫存量】:" + MAX_SIZE + "/t暫時不能執行生産任務!");
// 放入産品，自動阻塞
list.put(new Object());
System.out.println("【現倉儲量為】:" + list.size());
if (list.size() == 0)
System.out.println("【庫存量】:0/t暫時不能執行生産任務!");
// 消費産品，自動阻塞
list.take();
System.out.println("【現倉儲量為】:" + list.size());
public LinkedBlockingQueue<Object> getList()
public void setList(LinkedBlockingQueue<Object> list)
【庫存量】:0 暫時不能執行生産任務!
【現倉儲量為】:1
【現倉儲量為】:3
【現倉儲量為】:4
【現倉儲量為】:5
【現倉儲量為】:6
【現倉儲量為】:7
【現倉儲量為】:8
【現倉儲量為】:9
【現倉儲量為】:10
【現倉儲量為】:11
【現倉儲量為】:2
【現倉儲量為】:13
【現倉儲量為】:14
【現倉儲量為】:17
【現倉儲量為】:19
【現倉儲量為】:20
【現倉儲量為】:21
【現倉儲量為】:22
【現倉儲量為】:23
【現倉儲量為】:24
【現倉儲量為】:25
【現倉儲量為】:26
【現倉儲量為】:12
【現倉儲量為】:27
【現倉儲量為】:18
【現倉儲量為】:16
【現倉儲量為】:15
【現倉儲量為】:0
【現倉儲量為】: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大家可以放心使用，這可不是它的問題，隻是在它和别的對象之間的同步有問題。
轉http://blog.csdn.net/monkey_d_meng/article/details/6251879
QQ：519841366
本頁版權歸作者和部落格園所有，歡迎轉載，但未經作者同意必須保留此段聲明，
且在文章頁面明顯位置給出原文連結，否則保留追究法律責任的權利
![Java小案例——随機數猜測随機數猜測[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)