引言
生產者和消費者問題是線程模型中的經典問題：生產者和消費者在同一時間段內共用同一個存儲空間，如下圖所示，生產者向空間里存放數據，而消費者取用數據，如果不加以協調可能會出現以下情況：
生產者消費者圖
存儲空間已滿，而生產者占用着它，消費者等着生產者讓出空間從而去除產品，生產者等着消費者消費產品，從而向空間中添加產品。互相等待，從而發生死鎖。
生產者消費者問題是研究多線程程序時繞不開的經典問題之一，它描述是有一塊緩沖區作為倉庫，生產者可以將產品放入倉庫，消費者則可以從倉庫中取走產品。解決生產者/消費者問題的方法可分為兩類：
（1）采用某種機制保護生產者和消費者之間的同步；
（2）在生產者和消費者之間建立一個管道。
第一種方式有較高的效率，並且易於實現，代碼的可控制性較好，屬於常用的模式。第二種管道緩沖區不易控制，被傳輸數據對象不易於封裝等，實用性不強。因此本文只介紹同步機制實現的生產者/消費者問題。
同步問題核心在於：如何保證同一資源被多個線程並發訪問時的完整性。常用的同步方法是采用信號或加鎖機制，保證資源在任意時刻至多被一個線程訪問。Java語言在多線程編程上實現了完全對象化，提供了對同步機制的良好支持。在Java中一共有四種方法支持同步，其中前三個是同步方法，一個是管道方法。
（1）wait() / notify()方法
（2）await() / signal()方法
（3）BlockingQueue阻塞隊列方法
（4）PipedInputStream / PipedOutputStream
本文只介紹最常用的前兩種種，第三、四種暫不做討論，有興趣的讀者可以自己去網上找答案。
一、wait() / notify()方法
wait() / nofity()方法是基類Object的兩個方法，也就意味着所有Java類都會擁有這兩個方法，這樣，我們就可以為任何對象實現同步機制。
wait()方法：當緩沖區已滿/空時，生產者/消費者線程停止自己的執行，放棄鎖，使自己處於等等狀態，讓其他線程執行。
notify()方法：當生產者/消費者向緩沖區放入/取出一個產品時，向其他等待的線程發出可執行的通知，同時放棄鎖，使自己處於等待狀態。
代碼實現：
1、倉庫類
import java.util.LinkedList; /** * 倉庫類Storage實現緩沖區 * * @author zcr */ public class Storage { // 倉庫最大存儲量 private final int MAX_SIZE = 100; // 倉庫存儲的載體 private LinkedList<Object> list = new LinkedList<Object>(); /** * 生產num個產品 * @param 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個產品 * @param 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; } }
2、生產者
/** * 生產者類Producer繼承線程類Thread * * * @author zcr * */ 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; } }
3、消費者
/** * 消費者類Consumer繼承線程類Thread * * * @author zcr * */ 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; } }
4、測試類
/** * 測試類Test * @author zcr * */ 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(); } }
5、結果：
【要消費的產品數量】:50 【庫存量】:0 暫時不能執行生產任務! 【要消費的產品數量】:20 【庫存量】:0 暫時不能執行生產任務! 【已經生產產品數】:10 【現倉儲量為】:10 【要消費的產品數量】:20 【庫存量】:10 暫時不能執行生產任務! 【要消費的產品數量】:50 【庫存量】:10 暫時不能執行生產任務! 【已經生產產品數】:10 【現倉儲量為】:20 【已經生產產品數】:10 【現倉儲量為】:30 【要消費的產品數量】:50 【庫存量】:30 暫時不能執行生產任務! 【已經消費產品數】:20 【現倉儲量為】:10 【已經生產產品數】:10 【現倉儲量為】:20 【已經生產產品數】:10 【現倉儲量為】:30 【要消費的產品數量】:50 【庫存量】:30 暫時不能執行生產任務! 【已經生產產品數】:10 【現倉儲量為】:40 【已經消費產品數】:30 【現倉儲量為】:10 【要消費的產品數量】:50 【庫存量】:10 暫時不能執行生產任務! 【已經生產產品數】:80 【現倉儲量為】:90 【已經消費產品數】:50 【現倉儲量為】: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()方法，將條件變量和一個鎖對象進行綁定，進而控制並發程序訪問競爭資源的安全。
下面來看代碼：
只需要更新倉庫類Storage的代碼即可，生產者Producer、消費者Consumer、測試類Test的代碼均不需要進行任何更改。
倉庫類
import java.util.LinkedList; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * 倉庫類Storage實現緩沖區 * * * @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/t【庫存量】:0/t暫時不能執行生產任務! 【已經生產產品數】:10/t【現倉儲量為】:10 【已經生產產品數】:10/t【現倉儲量為】:20 【要消費的產品數量】:30/t【庫存量】:20/t暫時不能執行生產任務! 【已經消費產品數】:20/t【現倉儲量為】:0 【已經生產產品數】:10/t【現倉儲量為】:10 【已經生產產品數】:10/t【現倉儲量為】:20 【要消費的產品數量】:50/t【庫存量】:20/t暫時不能執行生產任務! 【已經生產產品數】:10/t【現倉儲量為】:30 【已經生產產品數】:10/t【現倉儲量為】:40 【要生產的產品數量】:80/t【庫存量】:40/t暫時不能執行生產任務! 【已經消費產品數】:30/t【現倉儲量為】:10 【要消費的產品數量】:50/t【庫存量】:10/t暫時不能執行生產任務! 【已經生產產品數】:80/t【現倉儲量為】:90 【已經消費產品數】:50/t【現倉儲量為】:40
這樣我們就知道為神馬我要在Storage類中定義public void produce(int num);和public void consume(int num);方法，並在生產者類Producer和消費者類Consumer中調用Storage類中的實現了吧。將可能發生的變化集中到一個類中，不影響原有的構架設計，同時無需修改其他業務層代碼。
總結
兩種方式原理一致，都是對獨占空間加鎖，阻塞和喚醒線程，第一種方式比較傳統，第二種方式速度比較快。
致謝：感謝您的耐心閱讀！