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

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

對(duì)于大多數(shù)人來(lái)說(shuō)，這個(gè)問(wèn)題可能是我們?cè)趯W(xué)校，執(zhí)行第一次并行算法所遇到的第一個(gè)同步問(wèn)題。

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

問(wèn)題陳述

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

假設(shè)一個(gè)生產(chǎn)者“生產(chǎn)”一份數(shù)據(jù)并將其存儲(chǔ)在緩沖區(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)者不會(huì)放置新數(shù)據(jù)進(jìn)來(lái)，也要確保當(dāng)緩沖區(qū)的數(shù)據(jù)為空時(shí)，消費(fèi)者不會(huì)試圖刪除數(shù)據(jù)緩沖區(qū)的數(shù)據(jù)。

解決方案

為了解決上述的并發(fā)問(wèn)題，生產(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)容為空的化，那么情況是一樣的，只不過(guò)，消費(fèi)者會(huì)先等待生產(chǎn)者的通知。

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

經(jīng)典的方法

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

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)生新的元素并將它們存儲(chǔ)在緩沖區(qū)。如果緩沖區(qū)已滿，那么生產(chǎn)者線程進(jìn)入睡眠狀態(tài)，直到有通知喚醒。否則，生產(chǎn)者線程將會(huì)在緩沖區(qū)創(chuàng)建一個(gè)新元素然后通知消費(fèi)者。就像我之前說(shuō)的，這個(gè)過(guò)程也適用于消費(fèi)者。如果緩沖區(qū)為空，那么消費(fèi)者將等待生產(chǎn)者的通知。否則，消費(fèi)者將從緩沖區(qū)刪除一個(gè)元素并通知生產(chǎn)者。

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

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

所以，讓我們把生產(chǎn)和消費(fèi)的邏輯從緩沖對(duì)象中進(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é)以線程安全的形式來(lái)存儲(chǔ)和刪除元素。

隊(duì)列阻塞(BlockingQueue)

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

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

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

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

所以,如果我們?cè)谑纠惺褂肂lockingQueue，我們就不需要再去實(shí)現(xiàn)等待和通知的機(jī)制。

接下來(lái)，我們來(lái)看看具體的代碼。

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看起來(lái)跟之前一樣，他們按照之前的方式生產(chǎn)和消費(fèi)元素。

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

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

這兒有很多種類型的BlockingQueue：

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

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

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

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

在這種情況下,由于添加操作不會(huì)被阻塞,生產(chǎn)者添加新元素時(shí)可以不用等待。每次當(dāng)生產(chǎn)者想要添加一個(gè)新元素時(shí)，會(huì)有一個(gè)隊(duì)列先存儲(chǔ)它。但是，這里面也存在一個(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ì)列將會(huì)被阻塞（具體阻塞方式取決于添加元素的方法）直到有足夠的空間騰出來(lái)。

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

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

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

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

使用線程池

還有什么地方我們可以優(yōu)化的？那首先來(lái)分析一下我們干了什么，我們實(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è)原生線程對(duì)應(yīng)于Java的線程
• 跟這個(gè)線程相關(guān)的描述符被添加到JVM內(nèi)部的數(shù)據(jù)結(jié)構(gòu)中

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

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

線程池將基于我們選擇的策略來(lái)處理線程的生命周期。它擁有有限數(shù)量的空閑線程，并在需要解決任務(wù)時(shí)啟用它們。通過(guò)這種方式,我們不需要為每一個(gè)新的請(qǐng)求創(chuàng)建一個(gè)新線程，因此，我們可以避免出現(xiàn)線程饑餓的問(wè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ù)來(lái)并發(fā)執(zhí)行，而不是讓一些線程通過(guò)共享公共的內(nèi)存塊來(lái)并發(fā)運(yùn)行。按照功能需求來(lái)思考的方式可以幫助我們避免一些常見的多線程問(wèn)題，如死鎖或數(shù)據(jù)競(jìng)爭(zhēng)等。沒(méi)有什么可以阻止我們?cè)俅紊钊脒@些問(wèn)題，但是，由于使用了功能范式，我們沒(méi)辦法命令式地同步并行計(jì)算(鎖)。這比直接使用線程和共享內(nèi)存所能碰到的幾率要少的多。在我們的例子中，共享一個(gè)阻塞隊(duì)列不是想要的情況，但我就是想強(qiáng)調(diào)這個(gè)優(yōu)勢(shì)。

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

說(shuō)了那么多，接下來(lá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是相同的。現(xiàn)在，執(zhí)行程序(線程池實(shí)現(xiàn))將接收任務(wù)，并安排它的工作線程去執(zhí)行他們。

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

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

總結(jié)

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

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。