Java多種方式解決生產者消費者問題（十分詳細）

阿新 • • 發佈：2019-02-09

一、問題描述

生產者消費者問題（Producer-consumer problem），也稱有限緩衝問題（Bounded-buffer problem），是一個多執行緒同步問題的經典案例。生產者生成一定量的資料放到緩衝區中，然後重複此過程；與此同時，消費者也在緩衝區消耗這些資料。生產者和消費者之間必須保持同步，要保證生產者不會在緩衝區滿時放入資料，消費者也不會在緩衝區空時消耗資料。不夠完善的解決方法容易出現死鎖的情況，此時程序都在等待喚醒。

示意圖：
生產者消費者

二、解決方法

思路

採用某種機制保護生產者和消費者之間的同步。有較高的效率，並且易於實現，程式碼的可控制性較好，屬於常用的模式。
在生產者和消費者之間建立一個管道。管道緩衝區不易控制，被傳輸資料物件不易於封裝等，實用性不強。

解決問題的核心

保證同一資源被多個執行緒併發訪問時的完整性。常用的同步方法是採用訊號或加鎖機制，保證資源在任意時刻至多被一個執行緒訪問。

Java能實現的幾種方法

wait() / notify()方法
await() / signal()方法
BlockingQueue阻塞佇列方法
訊號量
管道

三、程式碼實現

1. wait() / notify()方法

當緩衝區已滿時，生產者執行緒停止執行，放棄鎖，使自己處於等狀態，讓其他執行緒執行；
當緩衝區已空時，消費者執行緒停止執行，放棄鎖，使自己處於等狀態，讓其他執行緒執行。

當生產者向緩衝區放入一個產品時，向其他等待的執行緒發出可執行的通知，同時放棄鎖，使自己處於等待狀態；
當消費者從緩衝區取出一個產品時，向其他等待的執行緒發出可執行的通知，同時放棄鎖，使自己處於等待狀態。

倉庫Storage.java

import java.util.LinkedList;

public class Storage {

    // 倉庫容量
    private final int MAX_SIZE = 10;
    // 倉庫儲存的載體
    private LinkedList<Object> list = new LinkedList<>();

    public 
 void produce() {
        synchronized (list) {
            while (list.size() + 1 > MAX_SIZE) {
                System.out.println("【生產者" + Thread.currentThread().getName()
                        + "】倉庫已滿");
                try {
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.add(new Object());
            System.out.println("【生產者" + Thread.currentThread().getName()
                    + "】生產一個產品，現庫存" + list.size());
            list.notifyAll();
        }
    }

    public void consume() {
        synchronized (list) {
            while (list.size() == 0) {
                System.out.println("【消費者" + Thread.currentThread().getName() 
                        + "】倉庫為空");
                try {
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.remove();
            System.out.println("【消費者" + Thread.currentThread().getName()
                    + "】消費一個產品，現庫存" + list.size());
            list.notifyAll();
        }
    }
}

生產者

public class Producer implements Runnable{
    private Storage storage;

    public Producer(){}

    public Producer(Storage storage , String name){
        this.storage = storage;
    }

    @Override
    public void run(){
        while(true){
            try{
                Thread.sleep(1000);
                storage.produce();
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }
}

消費者

public class Consumer implements Runnable{
    private Storage storage;

    public Consumer(){}

    public Consumer(Storage storage , String name){
        this.storage = storage;
    }

    @Override
    public void run(){
        while(true){
            try{
                Thread.sleep(3000);
                storage.consume();
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }
}

主函式

public class Main {

    public static void main(String[] args) {
        Storage storage = new Storage();
        Thread p1 = new Thread(new Producer(storage));
        Thread p2 = new Thread(new Producer(storage));
        Thread p3 = new Thread(new Producer(storage));

        Thread c1 = new Thread(new Consumer(storage));
        Thread c2 = new Thread(new Consumer(storage));
        Thread c3 = new Thread(new Consumer(storage));

        p1.start();
        p2.start();
        p3.start();
        c1.start();
        c2.start();
        c3.start();
    }
}

執行結果

【生產者p1】生產一個產品，現庫存1
【生產者p2】生產一個產品，現庫存2
【生產者p3】生產一個產品，現庫存3
【生產者p1】生產一個產品，現庫存4
【生產者p2】生產一個產品，現庫存5
【生產者p3】生產一個產品，現庫存6
【生產者p1】生產一個產品，現庫存7
【生產者p2】生產一個產品，現庫存8
【消費者c1】消費一個產品，現庫存7
【生產者p3】生產一個產品，現庫存8
【消費者c2】消費一個產品，現庫存7
【消費者c3】消費一個產品，現庫存6
【生產者p1】生產一個產品，現庫存7
【生產者p2】生產一個產品，現庫存8
【生產者p3】生產一個產品，現庫存9
【生產者p1】生產一個產品，現庫存10
【生產者p2】倉庫已滿
【生產者p3】倉庫已滿
【生產者p1】倉庫已滿
【消費者c1】消費一個產品，現庫存9
【生產者p1】生產一個產品，現庫存10
【生產者p3】倉庫已滿
。。。。。。以下省略

一個生產者執行緒執行produce方法，睡眠1s；一個消費者執行一次consume方法，睡眠3s。此次實驗過程中，有3個生產者和3個消費者，也就是我們說的多對多的情況。倉庫的容量為10，可以看出消費的速度明顯慢於生產的速度，符合設定。

注意：

notifyAll()方法可使所有正在等待佇列中等待同一共享資源的“全部”執行緒從等待狀態退出，進入可執行狀態。此時，優先順序最高的哪個執行緒最先執行，但也有可能是隨機執行的，這要取決於JVM虛擬機器的實現。即最終也只有一個執行緒能被執行，上述執行緒優先順序都相同，每次執行的執行緒都不確定是哪個，後來給執行緒設定優先順序後也跟預期不一樣，還是要看JVM的具體實現吧。

2. await() / signal()方法

在JDK5中，用ReentrantLock和Condition可以實現等待/通知模型，具有更大的靈活性。通過在Lock物件上呼叫newCondition()方法，將條件變數和一個鎖物件進行繫結，進而控制併發程式訪問競爭資源的安全。

在這裡只需改動Storage類

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Storage {

    // 倉庫最大儲存量
    private final int MAX_SIZE = 10;
    // 倉庫儲存的載體
    private LinkedList<Object> list = new LinkedList<Object>();
    // 鎖
    private final Lock lock = new ReentrantLock();
    // 倉庫滿的條件變數
    private final Condition full = lock.newCondition();
    // 倉庫空的條件變數
    private final Condition empty = lock.newCondition();

    public void produce()
    {
        // 獲得鎖
        lock.lock();
        while (list.size() + 1 > MAX_SIZE) {
            System.out.println("【生產者" + Thread.currentThread().getName()
                     + "】倉庫已滿");
            try {
                full.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        list.add(new Object());
        System.out.println("【生產者" + Thread.currentThread().getName() 
                 + "】生產一個產品，現庫存" + list.size());

        // 喚醒其他所有執行緒、釋放鎖
        full.signalAll();
        empty.signalAll();
        lock.unlock();
    }

    public void consume()
    {
        // 獲得鎖
        lock.lock();
        while (list.size() == 0) {
            System.out.println("【消費者" + Thread.currentThread().getName()
                     + "】倉庫為空");
            try {
                empty.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        list.remove();
        System.out.println("【消費者" + Thread.currentThread().getName()
                 + "】消費一個產品，現庫存" + list.size());

        // 喚醒其他所有執行緒、釋放鎖
        full.signalAll();
        empty.signalAll();
        lock.unlock();
    }
}

執行結果與wait()/notify()類似

3. BlockingQueue阻塞佇列方法

BlockingQueue是JDK5.0的新增內容，它是一個已經在內部實現了同步的佇列，實現方式採用的是我們第2種await() / signal()方法。它可以在生成物件時指定容量大小，用於阻塞操作的是put()和take()方法。

put()方法：類似於我們上面的生產者執行緒，容量達到最大時，自動阻塞。
take()方法：類似於我們上面的消費者執行緒，容量為0時，自動阻塞。

import java.util.concurrent.LinkedBlockingQueue;

public class Storage {

    // 倉庫儲存的載體
    private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<>(10);

    public void produce() {
        try{
            list.put(new Object());
            System.out.println("【生產者" + Thread.currentThread().getName()
                    + "】生產一個產品，現庫存" + list.size());
        } catch (InterruptedException e){
            e.printStackTrace();
        }
    }

    public void consume() {
        try{
            list.take();
            System.out.println("【消費者" + Thread.currentThread().getName()
                    + "】消費了一個產品，現庫存" + list.size());
        } catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

可能會出現put()或take()和System.out.println()輸出不匹配的情況，是由於它們之間沒有同步造成的。BlockingQueue可以放心使用，這可不是它的問題，只是在它和別的物件之間的同步有問題。

4. 訊號量

Semaphore是一種基於計數的訊號量。它可以設定一個閾值，基於此，多個執行緒競爭獲取許可訊號，做完自己的申請後歸還，超過閾值後，執行緒申請許可訊號將會被阻塞。Semaphore可以用來構建一些物件池，資源池之類的，比如資料庫連線池，我們也可以建立計數為1的Semaphore，將其作為一種類似互斥鎖的機制，這也叫二元訊號量，表示兩種互斥狀態。計數為0的Semaphore是可以release的，然後就可以acquire（即一開始使執行緒阻塞從而完成其他執行。）。

import java.util.LinkedList;
import java.util.concurrent.Semaphore;

public class Storage {

    // 倉庫儲存的載體
    private LinkedList<Object> list = new LinkedList<Object>();
    // 倉庫的最大容量
    final Semaphore notFull = new Semaphore(10);
    // 將執行緒掛起，等待其他來觸發
    final Semaphore notEmpty = new Semaphore(0);
    // 互斥鎖
    final Semaphore mutex = new Semaphore(1);

    public void produce()
    {
        try {
            notFull.acquire();
            mutex.acquire();
            list.add(new Object());
            System.out.println("【生產者" + Thread.currentThread().getName()
                    + "】生產一個產品，現庫存" + list.size());
        }
        catch (Exception e) {
            e.printStackTrace();
        } finally {
            mutex.release();
            notEmpty.release();
        }
    }

    public void consume()
    {
        try {
            notEmpty.acquire();
            mutex.acquire();
            list.remove();
            System.out.println("【消費者" + Thread.currentThread().getName()
                    + "】消費一個產品，現庫存" + list.size());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            mutex.release();
            notFull.release();
        }
    }
}

5. 管道

一種特殊的流，用於不同執行緒間直接傳送資料，一個執行緒傳送資料到輸出管道，另一個執行緒從輸入管道中讀資料。

inputStream.connect(outputStream)或outputStream.connect(inputStream)作用是使兩個Stream之間產生通訊連結，這樣才可以將資料進行輸出與輸入。

這種方式只適用於兩個執行緒之間通訊，不適合多個執行緒之間通訊。

1. PipedInputStream / PipedOutputStream （操作位元組流）

Producer

import java.io.IOException;
import java.io.PipedOutputStream;

public class Producer implements Runnable {

    private PipedOutputStream pipedOutputStream;

    public Producer() {
        pipedOutputStream = new PipedOutputStream();
    }

    public PipedOutputStream getPipedOutputStream() {
        return pipedOutputStream;
    }

    @Override
    public void run() {
        try {
            for (int i = 1; i <= 5; i++) {
                pipedOutputStream.write(("This is a test, Id=" + i + "!\n").getBytes());
            }
            pipedOutputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Consumer

import java.io.IOException;
import java.io.PipedInputStream;

public class Consumer implements Runnable {
    private PipedInputStream pipedInputStream;

    public Consumer() {
        pipedInputStream = new PipedInputStream();
    }

    public PipedInputStream getPipedInputStream() {
        return pipedInputStream;
    }

    @Override
    public void run() {
        int len = -1;
        byte[] buffer = new byte[1024];
        try {
            while ((len = pipedInputStream.read(buffer)) != -1) {
                System.out.println(new String(buffer, 0, len));
            }
            pipedInputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

主函式

import java.io.IOException;

public class Main {

    public static void main(String[] args) {
        Producer p = new Producer();
        Consumer c = new Consumer();
        Thread t1 = new Thread(p);
        Thread t2 = new Thread(c);
        try {
            p.getPipedOutputStream().connect(c.getPipedInputStream());
            t2.start();
            t1.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

2. PipedReader / PipedWriter （操作字元流）

Producer

import java.io.IOException;
import java.io.PipedWriter;

public class Producer implements Runnable {

    private PipedWriter pipedWriter;

    public Producer() {
        pipedWriter = new PipedWriter();
    }

    public PipedWriter getPipedWriter() {
        return pipedWriter;
    }

    @Override
    public void run() {
        try {
            for (int i = 1; i <= 5; i++) {
                pipedWriter.write("This is a test, Id=" + i + "!\n");
            }
            pipedWriter.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Consumer

import java.io.IOException;
import java.io.PipedReader;

public class Consumer implements Runnable {
    private PipedReader pipedReader;

    public Consumer() {
        pipedReader = new PipedReader();
    }

    public PipedReader getPipedReader() {
        return pipedReader;
    }

    @Override
    public void run() {
        int len = -1;
        char[] buffer = new char[1024];
        try {
            while ((len = pipedReader.read(buffer)) != -1) {
                System.out.println(new String(buffer, 0, len));
            }
            pipedReader.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

主函式

import java.io.IOException;

public class Main {

    public static void main(String[] args) {
        Producer p = new Producer();
        Consumer c = new Consumer();
        Thread t1 = new Thread(p);
        Thread t2 = new Thread(c);
        try {
            p.getPipedWriter().connect(c.getPipedReader());
            t2.start();
            t1.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

想檢視上面幾種方式的完整程式碼，請點選這裡：生產者消費者問題的一些實驗

Java多種方式解決生產者消費者問題（十分詳細）

一、問題描述

二、解決方法

思路

解決問題的核心

Java能實現的幾種方法

三、程式碼實現

1. wait() / notify()方法

2. await() / signal()方法

3. BlockingQueue阻塞佇列方法

4. 訊號量

5. 管道

1. PipedInputStream / PipedOutputStream （操作位元組流）

2. PipedReader / PipedWriter （操作字元流）

參考

Java多種方式解決生產者消費者問題（十分詳細）

java每日一練------生產者消費者（執行緒）

兩種方式解決生產者消費者問題

多線程-生產者消費者（BlockingQueue實現）

java搭建阿里雲伺服器環境（java環境+mysql+tomcat）和部署 JavaWeb 專案到雲伺服器（十分詳細）

（原創）android PowerManager分析（十分詳細）

Hadoop單機、偽分散式叢集搭建（十分詳細）

Java Executor併發框架（十四）Executor框架執行緒池使用原始方式實現生產者消費者模式

java併發之----實現生產者/消費者模式（操作值&一對一交替列印）

java第18天----生產者消費者，Lock鎖，守護執行緒，join（）方法

Java多執行緒之生產者消費者問題：使用synchronized 關鍵字解決生產者消費者問題

Java多執行緒之生產者消費者問題：使用重入鎖、條件變數優雅地解決生產者消費者問題

Java多執行緒-生產者消費者例子-使用阻塞佇列（BlockingQueue）實現

Thinking in Java---執行緒通訊+三種方式實現生產者消費者問題

java並發之生產者消費者模型

Linux 使用rpm方式安裝最新mysql（5.7）步驟以及常見問題解決

通訊機制解決生產者消費者問題

管程機制--解決生產者消費者問題

java斐波那契數列（Fibonacci sequence）的三種方式：遞迴，備忘錄，動態規劃

Springcloud電子商城系統 java B2B2C-服務消費者（rest+ribbon）

Java多種方式解決生產者消費者問題（十分詳細）

一、問題描述

二、解決方法

思路

解決問題的核心

Java能實現的幾種方法

三、程式碼實現

1. wait() / notify()方法

2. await() / signal()方法

3. BlockingQueue阻塞佇列方法

4. 訊號量

5. 管道

1. PipedInputStream / PipedOutputStream （操作位元組流）

2. PipedReader / PipedWriter （操作字元流）

參考

相關推薦