淺談Java中生產(chǎn)者與消費(fèi)者問題的演變

想要了解更多關(guān)于Java生產(chǎn)者消費(fèi)者問題的演變嗎？那就看看這篇文章吧，我們分別用舊方法和新方法來處理這個(gè)問題。

生產(chǎn)者消費(fèi)者問題是一個(gè)典型的多進(jìn)程同步問題。

對于大多數(shù)人來說，這個(gè)問題可能是我們在學(xué)校，執(zhí)行第一次并行算法所遇到的第一個(gè)同步問題。

雖然它很簡單，但一直是并行計(jì)算中的最大挑戰(zhàn) - 多個(gè)進(jìn)程共享一個(gè)資源。

問題陳述

生產(chǎn)者和消費(fèi)者兩個(gè)程序,共享一個(gè)大小有限的公共緩沖區(qū)。

假設(shè)一個(gè)生產(chǎn)者“生產(chǎn)”一份數(shù)據(jù)并將其存儲在緩沖區(qū)中，而一個(gè)消費(fèi)者“消費(fèi)”這份數(shù)據(jù)，并將這份數(shù)據(jù)從緩沖區(qū)中刪除。

再假設(shè)現(xiàn)在這兩個(gè)程序在并發(fā)地運(yùn)行，我們需要確保當(dāng)緩沖區(qū)的數(shù)據(jù)已滿時(shí)，生產(chǎn)者不會放置新數(shù)據(jù)進(jìn)來，也要確保當(dāng)緩沖區(qū)的數(shù)據(jù)為空時(shí)，消費(fèi)者不會試圖刪除數(shù)據(jù)緩沖區(qū)的數(shù)據(jù)。

解決方案

為了解決上述的并發(fā)問題，生產(chǎn)者和消費(fèi)者將不得不相互通信。

如果緩沖區(qū)已滿，生產(chǎn)者將處于睡眠狀態(tài)，直到有通知信息喚醒。

在消費(fèi)者將一些數(shù)據(jù)從緩沖區(qū)刪除后，消費(fèi)者將通知生產(chǎn)者，隨后生產(chǎn)者將重新開始填充數(shù)據(jù)到緩沖區(qū)中。

如果緩沖區(qū)內(nèi)容為空的化，那么情況是一樣的，只不過，消費(fèi)者會先等待生產(chǎn)者的通知。

但如果這種溝通做得不恰當(dāng)，在進(jìn)程彼此等待的位置可能導(dǎo)致程序死鎖。

經(jīng)典的方法

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

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<Integer> 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);
        }
      }
    }
  }
}

這里我們有生產(chǎn)者和消費(fèi)者兩個(gè)線程，它們共享一個(gè)公共緩沖區(qū)。生產(chǎn)者線程開始產(chǎn)生新的元素并將它們存儲在緩沖區(qū)。如果緩沖區(qū)已滿，那么生產(chǎn)者線程進(jìn)入睡眠狀態(tài)，直到有通知喚醒。否則，生產(chǎn)者線程將會在緩沖區(qū)創(chuàng)建一個(gè)新元素然后通知消費(fèi)者。就像我之前說的，這個(gè)過程也適用于消費(fèi)者。如果緩沖區(qū)為空，那么消費(fèi)者將等待生產(chǎn)者的通知。否則，消費(fèi)者將從緩沖區(qū)刪除一個(gè)元素并通知生產(chǎn)者。

正如你所看到的，在之前的例子中，生產(chǎn)者和消費(fèi)者的工作都是管理緩沖區(qū)的對象。這些線程僅僅調(diào)用了buffer.produce()和buffer.consume()兩個(gè)方法就搞定了一切。

對于緩沖區(qū)是否應(yīng)該負(fù)責(zé)創(chuàng)建或者刪除元素，一直都是一個(gè)有爭議的話題，但在我看來，緩沖區(qū)不應(yīng)該做這種事情。當(dāng)然，這取決于你想要達(dá)到的目的，但在這種情況下，緩沖區(qū)應(yīng)該只是負(fù)責(zé)以線程安全的形式存儲合并元素，而不是生產(chǎn)新的元素。

所以，讓我們把生產(chǎn)和消費(fèi)的邏輯從緩沖對象中進(jìn)行解耦。

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<Integer> 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;
      }
    }
  }
}

這樣好多了，至少現(xiàn)在緩沖區(qū)僅僅負(fù)責(zé)以線程安全的形式來存儲和刪除元素。

隊(duì)列阻塞(BlockingQueue)

不過，我們還可以進(jìn)一步改善。

在前面的例子中，我們已經(jīng)創(chuàng)建了一個(gè)緩沖區(qū)，每當(dāng)存儲一個(gè)元素之前，緩沖區(qū)將等待是否有可用的一個(gè)槽以防止沒有足夠的存儲空間，并且，在合并之前，緩沖區(qū)也會等待一個(gè)新的元素出現(xiàn)，以確保存儲和刪除的操作是線程安全的。

但是，Java本身的庫已經(jīng)整合了這些操作。它被稱之為BlockingQueue，在這里可以查看它的詳細(xì)文檔。

BlockingQueue是一個(gè)以線程安全的形式存入和取出實(shí)例的隊(duì)列。而這就是我們所需要的。

所以,如果我們在示例中使用BlockingQueue，我們就不需要再去實(shí)現(xiàn)等待和通知的機(jī)制。

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

package ProducerConsumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

public class ProducerConsumerWithBlockingQueue {

  public static void main(String[] args) throws InterruptedException {
    BlockingQueue<Integer> 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看起來跟之前一樣，他們按照之前的方式生產(chǎn)和消費(fèi)元素。

唯一的區(qū)別在于，這里我們使用blockingQueue代替緩沖區(qū)對象。

關(guān)于Blocking Queue的更多細(xì)節(jié)

這兒有很多種類型的BlockingQueue：

• ×××隊(duì)列
• 有界隊(duì)列

一個(gè)×××隊(duì)列幾乎可以無限地增加元素，任何添加操作將不會被阻止。

你可以以這種方式去創(chuàng)建一個(gè)×××隊(duì)列：

BlockingQueue<String> blockingQueue = new LinkedBlockingDeque<>();

在這種情況下,由于添加操作不會被阻塞,生產(chǎn)者添加新元素時(shí)可以不用等待。每次當(dāng)生產(chǎn)者想要添加一個(gè)新元素時(shí)，會有一個(gè)隊(duì)列先存儲它。但是，這里面也存在一個(gè)異常需要捕獲。如果消費(fèi)者刪除元素的速度比生產(chǎn)者添加新的元素要慢，那么內(nèi)存將被填滿，我們將可能得到一個(gè)OutOfMemory異常。

與之相反的則是有界隊(duì)列，存在一個(gè)固定大小。你可以這樣去創(chuàng)建它：

BlockingQueue<String> blockingQueue = new LinkedBlockingDeque<>(10);

兩者最主要的區(qū)別在于，使用有界隊(duì)列的情況下，如果隊(duì)列內(nèi)存已滿，而生產(chǎn)者仍然試圖往里面塞元素，那么隊(duì)列將會被阻塞（具體阻塞方式取決于添加元素的方法）直到有足夠的空間騰出來。

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

• add() - 如果插入成功返回true，否則拋出IllegalStateException
• put() - 往隊(duì)列中插入元素，并在有必要的情況下等待一個(gè)可用的槽(slot)
• offer() - 如果插入元素成功返回true，否則返回false
• offer(E e, long timeout, TimeUnit unit) – 在隊(duì)列沒有滿的情況下，或者為了一個(gè)可用的slot而等待指定的時(shí)間后，往隊(duì)列中插入一個(gè)元素。

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

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

使用線程池

還有什么地方我們可以優(yōu)化的？那首先來分析一下我們干了什么，我們實(shí)例化了兩個(gè)線程，一個(gè)被叫做生產(chǎn)者，專門往隊(duì)列里面塞元素，另一個(gè)被叫做消費(fèi)者，負(fù)責(zé)從隊(duì)列里面刪元素。

然而，好的軟件技術(shù)表明，手動(dòng)地去創(chuàng)建和銷毀線程是不好的做法。首先創(chuàng)建線程是一項(xiàng)昂貴的任務(wù)，每創(chuàng)建一個(gè)線程，意味著要經(jīng)歷一遍下面的步驟：

• 首先要分配內(nèi)存給一個(gè)線程堆棧
• 操作系統(tǒng)要?jiǎng)?chuàng)建一個(gè)原生線程對應(yīng)于Java的線程
• 跟這個(gè)線程相關(guān)的描述符被添加到JVM內(nèi)部的數(shù)據(jù)結(jié)構(gòu)中

首先別誤會我，我們的案例中用了幾個(gè)線程是沒有問題的，而那也是并發(fā)工作的方式之一。這里的問題是，我們是手動(dòng)地去創(chuàng)建線程，這可以說是一次糟糕的實(shí)踐。如果我們手動(dòng)地創(chuàng)建線程，除了創(chuàng)建過程中的消耗外，還有另一個(gè)問題，就是我們無法控制同時(shí)有多少個(gè)線程在運(yùn)行。舉個(gè)例子，如果同時(shí)有一百萬次請求線上服務(wù)，那么每一次請求都會相應(yīng)的創(chuàng)建一個(gè)線程，那么同時(shí)會有一百萬個(gè)線程在后臺運(yùn)行，這將會導(dǎo)致[thread starvation](https://en.wikipedia.org/wiki/Starvation_(computer_science))

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

線程池將基于我們選擇的策略來處理線程的生命周期。它擁有有限數(shù)量的空閑線程，并在需要解決任務(wù)時(shí)啟用它們。通過這種方式,我們不需要為每一個(gè)新的請求創(chuàng)建一個(gè)新線程，因此，我們可以避免出現(xiàn)線程饑餓的問題。

Java線程池的實(shí)現(xiàn)包括：

• 一個(gè)任務(wù)隊(duì)列
• 一個(gè)工作線程的集合
• 一個(gè)線程工廠
• 管理線程池狀態(tài)的元數(shù)據(jù)

為了同時(shí)運(yùn)行一些任務(wù)，你必須把他們先放到任務(wù)隊(duì)列里。然后，當(dāng)一個(gè)線程可用的時(shí)候，它將接收一個(gè)任務(wù)并運(yùn)行它?？捎玫木€程越多，并行執(zhí)行的任務(wù)就越多。

除了管理線程生命周期，使用線程池還有另一個(gè)好處，當(dāng)你計(jì)劃如何分割任務(wù)，以便同時(shí)執(zhí)行時(shí)，你能想到更多種方式。并行性的單位不再是線程了，而是任務(wù)。你設(shè)計(jì)一些任務(wù)來并發(fā)執(zhí)行，而不是讓一些線程通過共享公共的內(nèi)存塊來并發(fā)運(yùn)行。按照功能需求來思考的方式可以幫助我們避免一些常見的多線程問題，如死鎖或數(shù)據(jù)競爭等。沒有什么可以阻止我們再次深入這些問題，但是，由于使用了功能范式，我們沒辦法命令式地同步并行計(jì)算(鎖)。這比直接使用線程和共享內(nèi)存所能碰到的幾率要少的多。在我們的例子中，共享一個(gè)阻塞隊(duì)列不是想要的情況，但我就是想強(qiáng)調(diào)這個(gè)優(yōu)勢。

在這里和這里你可以找到更多有關(guān)線程池的內(nèi)容。

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

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<Integer> 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();
  }
}

這里的區(qū)別在于，我們不在手動(dòng)創(chuàng)建或運(yùn)行消費(fèi)者和生產(chǎn)者線程。我們建立一個(gè)線程池，它將收到兩個(gè)任務(wù)，生產(chǎn)者和消費(fèi)者的任務(wù)。生產(chǎn)者和消費(fèi)者的任務(wù)，實(shí)際上跟之前例子里面使用的runnable是相同的?，F(xiàn)在，執(zhí)行程序(線程池實(shí)現(xiàn))將接收任務(wù)，并安排它的工作線程去執(zhí)行他們。

在我們簡單的案例下，一切都跟之前一樣運(yùn)行。就像之前的例子，我們?nèi)匀挥袃蓚€(gè)線程，他們?nèi)匀灰酝瑯拥姆绞缴a(chǎn)和消費(fèi)元素。雖然我們并沒有讓性能得到提升，但是代碼看起來干凈多了。我們不再手動(dòng)創(chuàng)建線程，而只是具體說明我們想要什么：我們想要并發(fā)執(zhí)行某些任務(wù)。

所以，當(dāng)你使用一個(gè)線程池時(shí)。你不需要考慮線程是并發(fā)執(zhí)行的單位，相反的，你把一些任務(wù)看作并發(fā)執(zhí)行的就好。以上就是你需要知道的，剩下的由執(zhí)行程序去處理。執(zhí)行程序會收到一些任務(wù)，然后，它會分配工作線程去處理它們。

總結(jié)

首先，我們看到了一個(gè)“傳統(tǒng)”的消費(fèi)者-生產(chǎn)者問題的解決方案。我們盡量避免了重復(fù)造沒有必要的車輪，恰恰相反，我們重用了已經(jīng)測試過的解決方案，因此，我們不是寫一個(gè)通知等待系統(tǒng)，而是嘗試使用Java已經(jīng)提供的blocking queue，因?yàn)镴ava為我們提供了一個(gè)非常有效的線程池來管理線程生命周期，讓我們可以擺脫手動(dòng)創(chuàng)建線程。通過這些改進(jìn)，消費(fèi)者-生產(chǎn)者問題的解決方案看起來更可靠和更好理解。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持億速云。