722
JAVA線程中的生產者和消費者問題
生產者消費者問題是研究多線程時繞不開的問題，描述的是有一塊生產者和消費者共享的有界緩衝區，生產者往緩衝區放入產品，消費者從緩衝區取走產品，這個過程可以無休止的執行，不能因緩衝區滿生產者放不進產品而終止，也不能因緩衝區空消費者無產品可取而終止。解決生產者消費者問題的方法有兩種，一種是采用某種機製保持生產者和消費者之間的同步，一種是在生產者和消費者之間建立一個管道。前一種有較高的效率並且可控製性較好，比較常用，後一種由於管道緩衝區不易控製及被傳輸數據對象不易封裝等原因，比較少用。
同步問題的核心在於，CPU是按時間片輪詢的方式執行程序，我們無法知道某一個線程是否被執行、是否被搶占、是否結束等，因此生產者完全可能當緩衝區已滿的時候還在放入產品，消費者也完全可能當緩衝區為空時還在取出產品。現在同步問題的解決方法一般是采用信號或者加鎖機製，即生產者線程當緩衝區已滿時放棄自己的執行權，進入等待狀態，並通知消費者線程執行。消費者線程當緩衝區已空時放棄自己的執行權，進入等待狀態，並通知生產者線程執行。這樣一來就保持了線程的同步，並避免了線程間互相等待而進入死鎖狀態。
JAVA語言提供了獨立於平台的線程機製，保持了write once, run anywhere的特色。同時也提供了對同步機製的良好支持。
JAVA中一共有四種方法支持同步，其中三個是同步方法，一個是管道方法。
①方法wait()/notify()
②方法await()/signal()
③阻塞隊列方法BlockingQueue
④管道方法PipedInputStream/PipedOutputStream
一、方法wait()/notify()
wait()和notify()是根類Object的兩個方法，也就意味著所有的JAVA類都具有這個兩個方法，可以認為所有的對象默認都具有一個鎖，雖然看不到也無法直接操作，但它是存在的。
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!");
}
}
}
}
}
}
二、方法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();
}
}
}
}
}
三、阻塞隊列方法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()之間沒有同步造成的。真正的原因是因為當緩衝區已滿，生產者在put()操作時，put()內部調用了await()方法放棄了線程的執行，然後消費者線程執行，調用take()方法，take()內部調用了signal()方法，通知生產者線程可以執行，致使在消費者的println()還沒運行的情況下生產者的println()先被執行，所以有了上麵的輸出。run()中的synchronized其實並沒有起什麼作用。對於BlockingQueue大家可以放心使用，這可不是它的問題，隻是在和別的對象之間的同步有問題。對於這種多重嵌套同步的問題以後再談。
四、管道方法PipedInputStream/PipedOutputStream
這個類位於java.io包中，是解決同步問題的最簡單的辦法，一個線程將數據寫入管道，另一個線程從管道讀取數據，這樣便構成了一種生產者/消費者的緩衝區編程模式。
在下麵的代碼中沒有使用Object對象，而是簡單的讀寫字節值，這是因為PipedInputStream/PipedOutputStream不允許傳輸對象，這是JAVA本身的一個bug,具體的大家可以看sun的解釋：https://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4131126import java.io.*;
public class Sycn4
{
private PipedOutputStream pos;
private PipedInputStream pis;
public Sycn4()
{
try
{
pos = new PipedOutputStream();
pis = new PipedInputStream(pos);
}
catch (IOException e)
{
System.out.println(e);
}
}
public void start()
{
new Producer().start();
new Consumer().start();
}
public static void main(String[] args) throws Exception
{
Sycn4 s4 = new Sycn4();
s4.start();
}
class Producer extends Thread
{
public void run()
{
try
{
while (true)
{
int b = (int) (Math.random() * 255);
System.out.println("Producer: a byte, the value is " + b);
pos.write(b);
pos.flush();
}
}
catch (Exception e)
{
e.printStackTrace();
}
finally
{
try
{
pos.close();
pis.close();
}
catch (IOException e)
{
System.out.println(e);
}
}
}
}
class Consumer extends Thread
{
public void run()
{
try
{
while (true)
{
int b = pis.read();
System.out.println("Consumer: a byte, the value is " +
String.valueOf(b));
}
}
catch (Exception e)
{
e.printStackTrace();
}
finally
{
try
{
pos.close();
pis.close();
}
catch (IOException e)
{
System.out.println(e);
}
}
}
}
}
原帖地址：https://blog.csdn.net/jhj735412/article/details/6931135
最後更新：2017-04-03 16:49:21