生產者消費者問題

生產者消費者問題，也稱有限緩衝問題，是一個多程序同步問題的經典案例。該問題描述了共享固定大小緩衝區的兩個程序——即所謂的“生產者”和“消費者”——在實際執行時會發生的問題。生產者的主要作用是生成一定量的資料放到緩衝區中，然後重複此過程。與此同時，消費者也在緩衝區消耗這些資料。該問題的關鍵就是要保證生產者不會在緩衝區滿時加入資料，消費者也不會在緩衝區中空時消耗資料。要解決該問題，就必須讓生產者在緩衝區滿時休眠，等到下次消費者消耗緩衝區中的資料的時候，生產者才能被喚醒，開始往緩衝區新增資料。同樣，也可以讓消費者在緩衝區空時進入休眠，等到生產者往緩衝區新增資料之後，再喚醒消費者。通常採用程序間通訊的方法解決該問題，常用的方法有訊號燈法等。如果解決方法不夠完善，則容易出現死鎖的情況。出現死鎖時，兩個執行緒都會陷入休眠，等待對方喚醒自己。該問題也能被推廣到多個生產者和消費者的情形。實現.不完善的實現.下面這個解決方法會導致競爭條件。如果程式設計師不夠小心，那麼他就有可能寫出下面這種演算法。該演算法使用了兩個系統庫函式，sleep 和 wakeup。呼叫 sleep 的程序會被阻斷，直到有另一個程序用 wakeup 喚醒之。程式碼中的 itemCount 用於記錄緩衝區中的資料項數。int itemCount = 0;procedure producer() {while (true) {item = produceItem();if (itemCount == BUFFER_SIZE) {sleep();putItemIntoBuffer(item);itemCount = itemCount + 1;if (itemCount == 1) {wakeup(consumer);procedure consumer() {while (true) {if (itemCount == 0) {sleep();item = removeItemFromBuffer();itemCount = itemCount - 1;if (itemCount == BUFFER_SIZE - 1) {wakeup(producer);consumeItem(item);上面程式碼中的問題在於它可能導致競爭條件，進而引發死鎖。考慮下面的情形：由於兩個程序都進入了永遠的休眠，死鎖情況出現了。因此，該演算法是不完善的。使用訊號燈的演算法.訊號燈可以避免上述喚醒指令不起作用的情況。該方法，emptyCount 用於記錄緩衝區中空閒空間數。當有新資料項被放入緩衝區時，fillCount 增加，emptyCount 減少。如果在生產者嘗試減少 emptyCount 的時候發現其值為零，那麼生產者就進入休眠。等到有資料項被消耗，emptyCount 增加的時候，生產者才被喚醒。消費者的行為類似。semaphore fillCount = 0; // 生產的專案semaphore emptyCount = BUFFER_SIZE; // 剩餘空間procedure producer() {while (true) {item = produceItem();down(emptyCount);putItemIntoBuffer(item);up(fillCount);procedure consumer() {while (true) {down(fillCount);item = removeItemFromBuffer();up(emptyCount);consumeItem(item);上述方法在只有一個生產者和一個消費者時能解決問題。對於多個生產者或者多個消費者共享緩衝區的情況，該演算法也會導致競爭條件，出現兩個或以上的程序同時讀或寫同一個緩衝區槽的情況。為了說明這種情況是如何發生的，可以假設 putItemIntoBuffer() 的一種可能的實現：先尋找下一個可用空槽，然後寫入資料項。下列情形是可能出現的：為了解決這個問題，需要在保證同一時刻只有一個生產者能夠執行 putItemIntoBuffer()。也就是說，需要尋找一種方法來互斥地執行臨界區的程式碼。為了達到這個目的，可引入一個二值訊號燈 mutex，其值只能為 1 或者 0。如果把執行緒放入 down(mutex) 和 up(mutex) 之間，就可以限制只有一個執行緒能被執行。多生產者、消費者的解決演算法如下：semaphore mutex = 1;semaphore fillCount = 0;semaphore emptyCount = BUFFER_SIZE;procedure producer() {while (true) {item = produceItem();down(emptyCount);down(mutex);putItemIntoBuffer(item);up(mutex);up(fillCount);procedure consumer() {while (true) {down(fillCount);down(mutex);item = removeItemFromBuffer();up(mutex);up(emptyCount);consumeItem(item);使用管程的演算法.下列虛擬碼展示的是使用管程來解決生產者消費者問題的辦法。由於管程一定能保證互斥，不必特地考慮保護臨界區。也就是說，下面這個方法不用修改就可以推廣適用於任意數量的生產者和消費者的情況。monitor ProducerConsumer {int itemCount;condition full;condition empty;procedure add(item) {while (itemCount == BUFFER_SIZE)wait(full);putItemIntoBuffer(item);itemCount = itemCount + 1;if (itemCount == 1)notify(empty);procedure remove() {while (itemCount == 0)wait(empty);item = removeItemFromBuffer();itemCount = itemCount - 1;if (itemCount == BUFFER_SIZE - 1)notify(full);return item;procedure producer() {while (true) {item = produceItem()ProducerConsumer.add(item)procedure consumer() {while (true) {item = ProducerConsumer.remove()consumeItem(item)注意程式碼中 while 語句的用法，都是用在測試緩衝區是否已滿或空的時候。當存在多個消費者時，有可能造成競爭條件的情況是：某一消費者在一項資料被放入緩衝區中時被喚醒，但是另一消費者已經在管程上等待了一段時間並移除了這項資料。如果 while 語句被改成 if，則會出現放入緩衝區的資料項過多，或移除空緩衝區中的元素的情況。不使用訊號燈或者管程.對於生產者消費者問題來說，特別是當只有一個生產者和一個消費者時，實現一個先進先出結構或者通訊通道非常重要。這樣，生產者-消費者模式就可以在不依賴訊號燈、互斥變數或管程的的情況下高效地傳輸資料。但如果採用這種模式，效能可能下降，因為實現這種模式的代價比較高。人們喜歡用先進先出結構或者通訊通道，只是因為可以避免端與端之間的原子性同步。用 C 語言舉例如下，請注意：volatile unsigned int produceCount, consumeCount;TokenType buffer[BUFFER_SIZE];void producer(void) {while (1) {while (produceCount - consumeCount == BUFFER_SIZE)sched_yield(); // 緩衝區滿buffer[produceCount % BUFFER_SIZE] = produceToken();produceCount += 1;void consumer(void) {while (1) {while (produceCount - consumeCount == 0)sched_yield(); // 緩衝區空consumeToken( buffer[consumeCount % BUFFER_SIZE]);consumeCount += 1;Java 中的例子.import java.util.Stack;import java.util.concurrent.atomic.AtomicInteger;1個生產者 3個消費者 生產、消費10次@作者 ptpublic class ProducerConsumer {Stack();final static int NO_ITEMS = 10;public static void main(String args[]) {ProducerConsumer pc = new ProducerConsumer();Thread t1 = new Thread(pc.new Producer());Consumer consumer = pc.new Consumer();Thread t2 = new Thread(consumer);Thread t3 = new Thread(consumer);Thread t4 = new Thread(consumer);t1.start();try {Thread.sleep(100);} catch (InterruptedException e1) {e1.printStackTrace();t2.start();t3.start();t4.start();try {t2.join();t3.join();t4.join();} catch (InterruptedException e) {e.printStackTrace();class Producer implements Runnable {public void produce(int i) {System.out.println("Producing " + i);items.push(new Integer(i));@Overridepublic void run() {int i = 0;// 生產10次while (i++ = NO_ITEMS;@Overridepublic void run() {while (!theEnd()) {synchronized (items) {while (items.isEmpty() &amp;&amp; (!theEnd())) {try {items.wait(10);} catch (InterruptedException e) {Thread.interrupted();consume();