來源：SpringForAll社區；

原文鏈接：https://dzone.com/articles/the-evolution-of-producer-consumer-problem-in-java；

作者：Ioan Tinca；

譯者：liumapp

想要了解更多關於Java生產者消費者問題的演變嗎？那就看看這篇文章吧，我們分別用舊方法和新方法來處理這個問題。

生產者消費者問題是一個典型的多進程同步問題。

對於大多數人來說，這個問題可能是我們在學校，執行第一次並行算法所遇到的第一個同步問題。

雖然它很簡單，但一直是並行計算中的最大挑戰 - 多個進程共享一個資源。

問題陳述

生產者和消費者兩個程序,共享一個大小有限的公共緩衝區。

假設一個生產者"生產"一份數據並將其存儲在緩衝區中，而一個消費者"消費"這份數據，並將這份數據從緩衝區中刪除。

再假設現在這兩個程序在併發地運行，我們需要確保當緩衝區的數據已滿時，生產者不會放置新數據進來，也要確保當緩衝區的數據爲空時，消費者不會試圖刪除數據緩衝區的數據。

解決方案

爲了解決上述的併發問題，生產者和消費者將不得不相互通信。

如果緩衝區已滿，生產者將處於睡眠狀態，直到有通知信息喚醒。

在消費者將一些數據從緩衝區刪除後，消費者將通知生產者，隨後生產者將重新開始填充數據到緩衝區中。

如果緩衝區內容爲空的化，那麼情況是一樣的，只不過，消費者會先等待生產者的通知。

但如果這種溝通做得不恰當，在進程彼此等待的位置可能導致程序死鎖。

經典的方法

首先來看一個典型的Java方案來解決這個問題。

package ProducerConsumer;
import java.util.LinkedList;
import java.util.Queue;
public class ClassicProducerConsumerExample {
 public static void main(String[] args) throws InterruptedException {
 Buffer buffer = new Buffer(2);
 Thread producerThread = new Thread(new Runnable() {
 @Override
 public void run() {
 try {
 buffer.produce();
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 });
 Thread consumerThread = new Thread(new Runnable() {
 @Override
 public void run() {
 try {
 buffer.consume();
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 });
 producerThread.start();
 consumerThread.start();
 producerThread.join();
 consumerThread.join();
 }
 static class Buffer {
 private Queue list;
 private int size;
 public Buffer(int size) {
 this.list = new LinkedList<>();
 this.size = size;
 }
 public void produce() throws InterruptedException {
 int value = 0;
 while (true) {
 synchronized (this) {
 while (list.size() >= size) {
 // wait for the consumer
 wait();
 }
 list.add(value);
 System.out.println("Produced " + value);
 value++;
 // notify the consumer
 notify();
 Thread.sleep(1000);
 }
 }
 }
 public void consume() throws InterruptedException {
 while (true) {
 synchronized (this) {
 while (list.size() == 0) {
 // wait for the producer
 wait();
 }
 int value = list.poll();
 System.out.println("Consume " + value);
 // notify the producer
 notify();
 Thread.sleep(1000);
 }
 }
 }
 }
}

這裏我們有生產者和消費者兩個線程，它們共享一個公共緩衝區。生產者線程開始產生新的元素並將它們存儲在緩衝區。如果緩衝區已滿，那麼生產者線程進入睡眠狀態，直到有通知喚醒。否則，生產者線程將會在緩衝區創建一個新元素然後通知消費者。就像我之前說的，這個過程也適用於消費者。如果緩衝區爲空，那麼消費者將等待生產者的通知。否則，消費者將從緩衝區刪除一個元素並通知生產者。

正如你所看到的，在之前的例子中，生產者和消費者的工作都是管理緩衝區的對象。這些線程僅僅調用了buffer.produce()和buffer.consume()兩個方法就搞定了一切。

對於緩衝區是否應該負責創建或者刪除元素，一直都是一個有爭議的話題，但在我看來，緩衝區不應該做這種事情。當然，這取決於你想要達到的目的，但在這種情況下，緩衝區應該只是負責以線程安全的形式存儲合併元素，而不是生產新的元素。

所以，讓我們把生產和消費的邏輯從緩衝對象中進行解耦。

package ProducerConsumer;
import java.util.LinkedList;
import java.util.Queue;
public class ProducerConsumerExample2 {
 public static void main(String[] args) throws InterruptedException {
 Buffer buffer = new Buffer(2);
 Thread producerThread = new Thread(() -> {
 try {
 int value = 0;
 while (true) {
 buffer.add(value);
 System.out.println("Produced " + value);
 value ++;
 Thread.sleep(1000);
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 });
 Thread consumerThread = new Thread(() -> {
 try {
 while (true) {
 int value = buffer.poll();
 System.out.println("Consume " + value);
 Thread.sleep(1000);
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 });
 producerThread.start();
 consumerThread.start();
 producerThread.join();
 consumerThread.join();
 }
 static class Buffer {
 private Queue list;
 private int size;
 public Buffer(int size) {
 this.list = new LinkedList<>();
 this.size = size;
 }
 public void add(int value) throws InterruptedException {
 synchronized (this) {
 while (list.size() >= size) {
 wait();
 }
 list.add(value);
 notify();
 }
 }
 public int poll() throws InterruptedException {
 synchronized (this) {
 while (list.size() == 0) {
 wait();
 }
 int value = list.poll();
 notify();
 return value;
 }
 }
 }
}

這樣好多了，至少現在緩衝區僅僅負責以線程安全的形式來存儲和刪除元素。

隊列阻塞(BlockingQueue)

不過，我們還可以進一步改善。

在前面的例子中，我們已經創建了一個緩衝區，每當存儲一個元素之前，緩衝區將等待是否有可用的一個槽以防止沒有足夠的存儲空間，並且，在合併之前，緩衝區也會等待一個新的元素出現，以確保存儲和刪除的操作是線程安全的。

但是，Java本身的庫已經整合了這些操作。它被稱之爲BlockingQueue，在這裏可以查看它的詳細文檔。

BlockingQueue是一個以線程安全的形式存入和取出實例的隊列。而這就是我們所需要的。

所以,如果我們在示例中使用BlockingQueue，我們就不需要再去實現等待和通知的機制。

接下來，我們來看看具體的代碼。

package ProducerConsumer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
public class ProducerConsumerWithBlockingQueue {
 public static void main(String[] args) throws InterruptedException {
 BlockingQueue blockingQueue = new LinkedBlockingDeque<>(2);
 Thread producerThread = new Thread(() -> {
 try {
 int value = 0;
 while (true) {
 blockingQueue.put(value);
 System.out.println("Produced " + value);
 value++;
 Thread.sleep(1000);
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 });
 Thread consumerThread = new Thread(() -> {
 try {
 while (true) {
 int value = blockingQueue.take();
 System.out.println("Consume " + value);
 Thread.sleep(1000);
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 });
 producerThread.start();
 consumerThread.start();
 producerThread.join();
 consumerThread.join();
 }
}

雖然runnables看起來跟之前一樣，他們按照之前的方式生產和消費元素。

唯一的區別在於，這裏我們使用blockingQueue代替緩衝區對象。

關於Blocking Queue的更多細節

這兒有很多種類型的BlockingQueue：

無界隊列

有界隊列

一個無界隊列幾乎可以無限地增加元素，任何添加操作將不會被阻止。

你可以以這種方式去創建一個無界隊列：

BlockingQueue blockingQueue = new LinkedBlockingDeque<>();

在這種情況下,由於添加操作不會被阻塞,生產者添加新元素時可以不用等待。每次當生產者想要添加一個新元素時，會有一個隊列先存儲它。但是，這裏面也存在一個異常需要捕獲。如果消費者刪除元素的速度比生產者添加新的元素要慢，那麼內存將被填滿，我們將可能得到一個OutOfMemory異常。

與之相反的則是有界隊列，存在一個固定大小。你可以這樣去創建它：

BlockingQueue blockingQueue = new LinkedBlockingDeque<>(10);

兩者最主要的區別在於，使用有界隊列的情況下，如果隊列內存已滿，而生產者仍然試圖往裏面塞元素，那麼隊列將會被阻塞（具體阻塞方式取決於添加元素的方法）直到有足夠的空間騰出來。

往blocking queue裏面添加元素一共有以下四種方式：

add() - 如果插入成功返回true，否則拋出IllegalStateException

put() - 往隊列中插入元素，並在有必要的情況下等待一個可用的槽(slot)

offer() - 如果插入元素成功返回true，否則返回false

offer(E e, long timeout, TimeUnit unit) – 在隊列沒有滿的情況下，或者爲了一個可用的slot而等待指定的時間後，往隊列中插入一個元素。

所以，如果你使用put()方法插入元素，而隊列內存已滿的情況下，我們的生產者就必須等待，直到有可用的slot出現。

以上就是我們上一個案例的全部，這跟ProducerConsumerExample2的工作原理是一樣的。

使用線程池

還有什麼地方我們可以優化的？那首先來分析一下我們幹了什麼，我們實例化了兩個線程，一個被叫做生產者，專門往隊列裏面塞元素，另一個被叫做消費者，負責從隊列裏面刪元素。

然而，好的軟件技術表明，手動地去創建和銷燬線程是不好的做法。首先創建線程是一項昂貴的任務，每創建一個線程，意味着要經歷一遍下面的步驟：

首先要分配內存給一個線程堆棧

操作系統要創建一個原生線程對應於Java的線程

跟這個線程相關的描述符被添加到JVM內部的數據結構中

首先別誤會我，我們的案例中用了幾個線程是沒有問題的，而那也是併發工作的方式之一。這裏的問題是，我們是手動地去創建線程，這可以說是一次糟糕的實踐。如果我們手動地創建線程，除了創建過程中的消耗外，還有另一個問題，就是我們無法控制同時有多少個線程在運行。舉個例子，如果同時有一百萬次請求線上服務，那麼每一次請求都會相應的創建一個線程，那麼同時會有一百萬個線程在後臺運行，這將會導致thread starvation

所以，我們需要一種全局管理線程的方式，這就用到了線程池。

線程池將基於我們選擇的策略來處理線程的生命週期。它擁有有限數量的空閒線程，並在需要解決任務時啓用它們。通過這種方式,我們不需要爲每一個新的請求創建一個新線程，因此，我們可以避免出現線程飢餓的問題。

Java線程池的實現包括：

一個任務隊列

一個工作線程的集合

一個線程工廠

管理線程池狀態的元數據

爲了同時運行一些任務，你必須把他們先放到任務隊列裏。然後，當一個線程可用的時候，它將接收一個任務並運行它。可用的線程越多，並行執行的任務就越多。

除了管理線程生命週期，使用線程池還有另一個好處，當你計劃如何分割任務，以便同時執行時，你能想到更多種方式。並行性的單位不再是線程了，而是任務。你設計一些任務來併發執行，而不是讓一些線程通過共享公共的內存塊來併發運行。按照功能需求來思考的方式可以幫助我們避免一些常見的多線程問題，如死鎖或數據競爭等。沒有什麼可以阻止我們再次深入這些問題，但是，由於使用了功能範式，我們沒辦法命令式地同步並行計算(鎖)。這比直接使用線程和共享內存所能碰到的機率要少的多。在我們的例子中，共享一個阻塞隊列不是想要的情況，但我就是想強調這個優勢。

在這裏和這裏你可以找到更多有關線程池的內容。

說了那麼多，接下來我們看看在案例中如何使用線程池。

package ProducerConsumer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
public class ProducerConsumerExecutorService {
 public static void main(String[] args) {
 BlockingQueue blockingQueue = new LinkedBlockingDeque<>(2);
 ExecutorService executor = Executors.newFixedThreadPool(2);
 Runnable producerTask = () -> {
 try {
 int value = 0;
 while (true) {
 blockingQueue.put(value);
 System.out.println("Produced " + value);
 value++;
 Thread.sleep(1000);
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 };
 Runnable consumerTask = () -> {
 try {
 while (true) {
 int value = blockingQueue.take();
 System.out.println("Consume " + value);
 Thread.sleep(1000);
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 };
 executor.execute(producerTask);
 executor.execute(consumerTask);
 executor.shutdown();
 }
}

這裏的區別在於，我們不在手動創建或運行消費者和生產者線程。我們建立一個線程池，它將收到兩個任務，生產者和消費者的任務。生產者和消費者的任務，實際上跟之前例子裏面使用的runnable是相同的。現在，執行程序(線程池實現)將接收任務，並安排它的工作線程去執行他們。

在我們簡單的案例下，一切都跟之前一樣運行。就像之前的例子，我們仍然有兩個線程，他們仍然要以同樣的方式生產和消費元素。雖然我們並沒有讓性能得到提升，但是代碼看起來乾淨多了。我們不再手動創建線程，而只是具體說明我們想要什麼：我們想要併發執行某些任務。

所以，當你使用一個線程池時。你不需要考慮線程是併發執行的單位，相反的，你把一些任務看作併發執行的就好。以上就是你需要知道的，剩下的由執行程序去處理。執行程序會收到一些任務，然後，它會分配工作線程去處理它們。

總結

首先，我們看到了一個"傳統"的消費者-生產者問題的解決方案。我們儘量避免了重複造沒有必要的車輪，恰恰相反，我們重用了已經測試過的解決方案，因此，我們不是寫一個通知等待系統，而是嘗試使用Java已經提供的blocking queue，因爲Java爲我們提供了一個非常有效的線程池來管理線程生命週期，讓我們可以擺脫手動創建線程。通過這些改進，消費者-生產者問題的解決方案看起來更可靠和更好理解。