淺談Java中生產者與消費者問題的演變

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

屯溪網站制作公司哪家好，找成都創(chuàng)新互聯！從網頁設計、網站建設、微信開發(fā)、APP開發(fā)、自適應網站建設等網站項目制作，到程序開發(fā)，運營維護。成都創(chuàng)新互聯成立于2013年到現在10年的時間，我們擁有了豐富的建站經驗和運維經驗，來保證我們的工作的順利進行。專注于網站建設就選成都創(chuàng)新互聯。

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

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

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

問題陳述

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

假設一個生產者“生產”一份數據并將其存儲在緩沖區(qū)中，而一個消費者“消費”這份數據，并將這份數據從緩沖區(qū)中刪除。

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

解決方案

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

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

在消費者將一些數據從緩沖區(qū)刪除后，消費者將通知生產者，隨后生產者將重新開始填充數據到緩沖區(qū)中。

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

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

經典的方法

首先來看一個典型的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<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);
        }
      }
    }
  }
}

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

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

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

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

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

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

隊列阻塞(BlockingQueue)

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

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

但是，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<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看起來跟之前一樣，他們按照之前的方式生產和消費元素。

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

關于Blocking Queue的更多細節(jié)

這兒有很多種類型的BlockingQueue：

×××隊列
有界隊列

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

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

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

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

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

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

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

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

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

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

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

使用線程池

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

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

首先要分配內存給一個線程堆棧
操作系統(tǒng)要創(chuàng)建一個原生線程對應于Java的線程
跟這個線程相關的描述符被添加到JVM內部的數據結構中

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

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

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

Java線程池的實現包括：

一個任務隊列
一個工作線程的集合
一個線程工廠
管理線程池狀態(tài)的元數據

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

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

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

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

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ū)別在于，我們不在手動創(chuàng)建或運行消費者和生產者線程。我們建立一個線程池，它將收到兩個任務，生產者和消費者的任務。生產者和消費者的任務，實際上跟之前例子里面使用的runnable是相同的?，F在，執(zhí)行程序(線程池實現)將接收任務，并安排它的工作線程去執(zhí)行他們。

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

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

總結

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

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持創(chuàng)新互聯。

標題名稱：淺談Java中生產者與消費者問題的演變
分享地址：http://jinyejixie.com/article36/jpdjpg.html

成都網站建設公司_創(chuàng)新互聯，為您提供企業(yè)建站、App設計、面包屑導航、定制網站、云服務器、關鍵詞優(yōu)化

聲明：本網站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源：創(chuàng)新互聯

猜你還喜歡下面的內容

成人午夜视频全免费观看高清-秋霞福利视频一区二区三区-国产精品久久久久电影小说-亚洲不卡区三一区三区一区

淺談Java中生產者與消費者問題的演變