JAVA線程中的生產者和消費者問題

時間 2020-01-14

原文原文鏈接

生產者消費者問題是研究多線程時繞不開的問題，描述的是有一塊生產者和消費者共享的有界緩衝區，生產者往緩衝區放入產品，消費者從緩衝區取走產品，這個過程能夠無休止的執行，不能因緩衝區滿生產者放不進產品而終止，也不能因緩衝區空消費者無產品可取而終止。java

解決生產者消費者問題的方法有兩種，一種是採用某種機制保持生產者和消費者之間的同步，一種是在生產者和消費者之間創建一個管道。前一種有較高的效率而且可控制性較好，比較經常使用，後一種因爲管道緩衝區不易控制及被傳輸數據對象不易封裝等緣由，比較少用。編程

同步問題的核心在於，CPU是按時間片輪詢的方式執行程序，咱們沒法知道某一個線程是否被執行、是否被搶佔、是否結束等，所以生產者徹底可能當緩衝區已滿的時候還在放入產品，消費者也徹底可能當緩衝區爲空時還在取出產品。如今同步問題的解決方法通常是採用信號或者加鎖機制，即生產者線程當緩衝區已滿時放棄本身的執行權，進入等待狀態，並通知消費者線程執行。消費者線程當緩衝區已空時放棄本身的執行權，進入等待狀態，並通知生產者線程執行。這樣一來就保持了線程的同步，並避免了線程間互相等待而進入死鎖狀態。多線程

JAVA語言提供了獨立於平臺的線程機制，保持了write once, run anywhere的特點。同時也提供了對同步機制的良好支持。dom

JAVA中一共有四種方法支持同步，其中三個是同步方法，一個是管道方法。ide

①方法wait()/notify()spa

②方法await()/signal().net

③阻塞隊列方法BlockingQueue線程

④管道方法PipedInputStream/PipedOutputStream對象

1、方法wait()/notify()blog

wait()和notify()是根類Object的兩個方法，也就意味着全部的AVA類都具備這個兩個方法，能夠認爲全部的對象默認都具備一個鎖，雖然看不到也沒法直接操做，但它是存在的。

wait()方法表示：當緩衝區已滿或空時，生產者或消費者線程中止本身的執行，放棄鎖，使本身處於等待狀態，讓另外一個線程開始執行。

notify()方法表示：當生產者或消費者對緩衝區放入或取出一個產品時，向另外一個線程發出可執行通知，同時放棄鎖，使本身處於等待狀態。

import java.util.LinkedList;

public class Sycn1

{

private LinkedList<Object> myList = new LinkedList<Object>();

private int MAX = 10;

public void start()

{

new Thread(new Producer()).start();

new Thread(new Consumer()).start();

}

public static void main(String[] args) throws Exception

{

Sycn1 s1 = new Sycn1();

s1.start();

}

class Producer implements Runnable

{

public void run()

{

while (true)

{

synchronized (myList)

{

try

{

while (myList.size() == MAX)

{

System.out.println("warning: it's full!");

myList.wait();

}

Object o = new Object();

if (myList.add(o))

{

System.out.println("Producer: " + o);

myList.notify();

}

catch (InterruptedException ie)

{

System.out.println("producer is interrupted!");

}

class Consumer implements Runnable

{

public void run()

{

while (true)

{

synchronized (myList)

{

try

{

while (myList.size() == 0)

{

System.out.println("warning: it's empty!");

myList.wait();

}

Object o = myList.removeLast();

System.out.println("Consumer: " + o);

myList.notify();

}

catch (InterruptedException ie)

{

System.out.println("consumer is interrupted!");

}

2、方法await()/signal()

在JDK5.0之後，JAVA提供了新的更加健壯的線程處理機制，包括了同步、鎖定、線程池等等，能夠實現更小粒度上的控制。await()和signal()就是其中用來同步的兩種方法，功能基本上和wait()/notify()相同，徹底能夠取代它們，可是它們和新引入的鎖定機制Lock直接掛鉤，具備更大的靈活性。

import java.util.LinkedList;

import java.util.concurrent.locks.*;

public class Sycn2

{

private LinkedList<Object> myList = new LinkedList<Object>();

private int MAX = 10;

private final Lock lock = new ReentrantLock();

private final Condition full = lock.newCondition();

private final Condition empty = lock.newCondition();

public void start()

{

new Thread(new Producer()).start();

new Thread(new Consumer()).start();

}

public static void main(String[] args) throws Exception

{

Sycn2 s2 = new Sycn2();

s2.start();

}

class Producer implements Runnable

{

public void run()

{

while (true)

{

lock.lock();

try

{

while (myList.size() == MAX)

{

System.out.println("warning: it's full!");

full.await();

}

Object o = new Object();

if (myList.add(o))

{

System.out.println("Producer: " + o);

empty.signal();

}

catch (InterruptedException ie)

{

System.out.println("producer is interrupted!");

}

finally

{

lock.unlock();

}

class Consumer implements Runnable

{

public void run()

{

while (true)

{

lock.lock();

try

{

while (myList.size() == 0)

{

System.out.println("warning: it's empty!");

empty.await();

}

Object o = myList.removeLast();

System.out.println("Consumer: " + o);

full.signal();

}

catch (InterruptedException ie)

{

System.out.println("consumer is interrupted!");

}

finally

{

lock.unlock();

}

3、阻塞隊列方法BlockingQueue

BlockingQueue也是JDK5.0的一部分，是一個已經在內部實現了同步的隊列，實現方式採用的是第2種await()/signal()方法。它能夠在生成對象時指定容量大小。它用於阻塞操做的是put()和take()方法。

put()方法相似於上面的生產者線程，容量最大時自動阻塞。

take()方法相似於上面的消費者線程，容量爲0時自動阻塞。

import java.util.concurrent.*;

public class Sycn3

{

private LinkedBlockingQueue<Object> queue

= new LinkedBlockingQueue<Object>(10);

private int MAX = 10;

public void start()

{

new Thread(new Producer()).start();

new Thread(new Consumer()).start();

}

public static void main(String[] args) throws Exception

{

Sycn3 s3 = new Sycn3();

s3.start();

}

class Producer implements Runnable

{

public void run()

{

while (true)

{

try

{

if (queue.size() == MAX)

System.out.println("warning: it's full!");

Object o = new Object();

queue.put(o);

System.out.println("Producer: " + o);

}

catch (InterruptedException e)

{

System.out.println("producer is interrupted!");

}

class Consumer implements Runnable

{

public void run()

{

while (true)

{

try

{

if (queue.size() == 0)

System.out.println("warning: it's empty!");

Object o = queue.take();

System.out.println("Consumer: " + o);

}

catch (InterruptedException e)

{

System.out.println("producer is interrupted!");

}

運行一下代碼發現問題：

warning: it's full!

Producer: java.lang.object@4526e2a

可能這是由於put()和System.out.println()之間沒有同步形成的，但改寫成以下代碼仍然沒有改觀

class Producer implements Runnable

{

public void run()

{

while (true)

{

try

{

if (queue.size() == MAX)

System.out.println("warning: it's full!");

Object o = new Object();

queue.put(o);

System.out.println("Producer: " + o);

}

catch (InterruptedException e)

{

System.out.println("producer is interrupted!");

}

真正的緣由是由於當緩衝區已滿，生產者在put()操做時，put()內部調用了await()方法放棄了線程的執行，而後消費者線程執行，調用take()方法，take()內部調用了signal()方法，通知生產者線程能夠執行，導致在消費者的println()還沒運行的狀況下生產者的println()先被執行，因此有了上面的輸出。run()中的synchronized其實並無起什麼做用。對於BlockingQueue你們能夠放心使用，這可不是它的問題，只是在和別的對象之間的同步有問題。對於這種多重嵌套同步的問題之後再談。

4、管道方法PipedInputStream/PipedOutputStream

這個類位於java.io包中，是解決同步問題的最簡單的辦法，一個線程將數據寫入管道，另外一個線程從管道讀取數據，這樣便構成了一種生產者/消費者的緩衝區編程模式。在下面的代碼沒有使用Object對象，而是簡單的讀寫字節值，這是由於PipedInputStream/PipedOutputStream不容許傳輸對象，這是JAVA自己的一個bug,具體的你們能夠看sun的解釋：http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4131126

import java.io.*;

public class Sycn4

{

private PipedOutputStream pos;

private PipedInputStream pis;

public Sycn4()

{

try

{

pos = new PipedOutputStream();

pis = new PipedInputStream(pos);

}