轉(zhuǎn)載自：http://blog.csdn.net/jhj735412/article/details/6931135

生產(chǎn)者消費(fèi)者問題是研究多線程程序時繞不開的問題，它的描述是有一塊生產(chǎn)者和消費(fèi)者共享的有界緩沖區(qū)，生產(chǎn)者往緩沖區(qū)放入產(chǎn)品，消費(fèi)者從緩沖區(qū)取走產(chǎn)品，這個過程可以無休止的執(zhí)行，不能因緩沖區(qū)滿生產(chǎn)者放不進(jìn)產(chǎn)品而終止，也不能因緩沖區(qū)空消費(fèi)者無產(chǎn)品可取而終止。

解決生產(chǎn)者消費(fèi)者問題的方法有兩種，一種是采用某種機(jī)制保持生產(chǎn)者和消費(fèi)者之間的同步，一種是在生產(chǎn)者和消費(fèi)者之間建立一個管道。前一種有較高的效率并且可控制性較好，比較常用，后一種由于管道緩沖區(qū)不易控制及被傳輸數(shù)據(jù)對象不易封裝等原因，比較少用。 同步問題的核心在于，CPU是按時間片輪詢的方式執(zhí)行程序，我們無法知道某一個線程是否被執(zhí)行、是否被搶占、是否結(jié)束等，因此生產(chǎn)者完全可能當(dāng)緩沖區(qū)已滿的時候還在放入產(chǎn)品，消費(fèi)者也完全可能當(dāng)緩沖區(qū)為空時還在取出產(chǎn)品。 現(xiàn)在同步問題的解決方法一般是采用信號或者加鎖機(jī)制，即生產(chǎn)者線程當(dāng)緩沖區(qū)已滿時放棄自己的執(zhí)行權(quán)，進(jìn)入等待狀態(tài)，并通知消費(fèi)者線程執(zhí)行。消費(fèi)者線程當(dāng)緩沖區(qū)已空時放棄自己的執(zhí)行權(quán)，進(jìn)入等待狀態(tài)，并通知生產(chǎn)者線程執(zhí)行。這樣一來就保持了線程的同步，并避免了線程間互相等待而進(jìn)入死鎖狀態(tài)。 JAVA語言提供了獨(dú)立于平臺的線程機(jī)制，保持了”write once, run anywhere”的特色。同時也提供了對同步機(jī)制的良好支持。 在JAVA中，一共有四種方法支持同步，其中三個是同步方法，一個是管道方法。 1.?????? 方法wait()/notify() 2.?????? 方法await()/signal() 3.?????? 阻塞隊(duì)列方法BlockingQueue 4.?????? 管道方法PipedInputStream/PipedOutputStream 下面我們看各個方法的實(shí)現(xiàn)： 1.?????? 方法wait()/notify() wait()和notify()是根類Object的兩個方法，也就意味著所有的JAVA類都會具有這個兩個方法，為什么會被這樣設(shè)計(jì)呢？我們可以認(rèn)為所有的對象默認(rèn)都具有一個鎖，雖然我們看不到，也沒有辦法直接操作，但它是存在的。 wait()方法表示：當(dāng)緩沖區(qū)已滿或空時，生產(chǎn)者或消費(fèi)者線程停止自己的執(zhí)行，放棄鎖，使自己處于等待狀態(tài)，讓另一個線程開始執(zhí)行； notify()方法表示：當(dāng)生產(chǎn)者或消費(fèi)者對緩沖區(qū)放入或取出一個產(chǎn)品時，向另一個線程發(fā)出可執(zhí)行通知，同時放棄鎖，使自己處于等待狀態(tài)。 下面是一個例子代碼：

?

import java.util.LinkedList;public class Sycn1...{private LinkedList<Object> myList =new LinkedList<Object>();private int MAX = 10;public Sycn1()...{}public void start()...{new Producer().start();new Consumer().start();}public static void main(String[] args) throws Exception...{Sycn1 s1 = new Sycn1();s1.start();}class Producer extends Thread...{ 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 extends Thread...{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提供了新的更加健壯的線程處理機(jī)制，包括了同步、鎖定、線程池等等，它們可以實(shí)現(xiàn)更小粒度上的控制。await()和signal()就是其中用來做同步的兩種方法，它們的功能基本上和wait()/notify()相同，完全可以取代它們，但是它們和新引入的鎖定機(jī)制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 Sycn2()...{}public void start()...{new Producer().start();new Consumer().start();}public static void main(String[] args) throws Exception...{Sycn2 s2 = new Sycn2();s2.start();}class Producer extends Thread...{ 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 extends Thread...{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.?????? 阻塞隊(duì)列方法BlockingQueue BlockingQueue也是JDK5.0的一部分，它是一個已經(jīng)在內(nèi)部實(shí)現(xiàn)了同步的隊(duì)列，實(shí)現(xiàn)方式采用的是我們的第2種await()/signal()方法。它可以在生成對象時指定容量大小。 它用于阻塞操作的是put()和take()方法。 put()方法類似于我們上面的生產(chǎn)者線程，容量最大時，自動阻塞。 take()方法類似于我們上面的消費(fèi)者線程，容量為0時，自動阻塞。 下面是一個例子代碼： import java.util.concurrent.*;public class Sycn3...{private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<Object>(10);private int MAX = 10;public Sycn3()...{}public void start()...{new Producer().start();new Consumer().start();}public static void main(String[] args) throws Exception...{Sycn3 s3 = new Sycn3();s3.start();}class Producer extends Thread...{ public void run()...{while(true)...{//synchronized(this){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 extends Thread...{public void run()...{while(true)...{//synchronized(this){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!");}//}}}}}

你發(fā)現(xiàn)這個例子中的問題了嗎？ 如果沒有，我建議你運(yùn)行一下這段代碼，仔細(xì)觀察它的輸出，是不是有下面這個樣子的？為什么會這樣呢？ … warning: it's full! Producer: java.lang.object@4526e2a … 你可能會說這是因?yàn)閜ut()和System.out.println()之間沒有同步造成的，我也這樣認(rèn)為，我也這樣認(rèn)為，但是你把run()中的synchronized前面的注釋去掉，重新編譯運(yùn)行，有改觀嗎？沒有。為什么？ 這是因?yàn)?#xff0c;當(dāng)緩沖區(qū)已滿，生產(chǎn)者在put()操作時，put()內(nèi)部調(diào)用了await()方法，放棄了線程的執(zhí)行，然后消費(fèi)者線程執(zhí)行，調(diào)用take()方法，take()內(nèi)部調(diào)用了signal()方法，通知生產(chǎn)者線程可以執(zhí)行，致使在消費(fèi)者的println()還沒運(yùn)行的情況下生產(chǎn)者的println()先被執(zhí)行，所以有了上面的輸出。run()中的synchronized其實(shí)并沒有起什么作用。 對于BlockingQueue大家可以放心使用，這可不是它的問題，只是在它和別的對象之間的同步有問題。 對于這種多重嵌套同步的問題，以后再談吧，歡迎大家討論啊！ 4.?????? 管道方法PipedInputStream/PipedOutputStream 這個類位于java.io包中，是解決同步問題的最簡單的辦法，一個線程將數(shù)據(jù)寫入管道，另一個線程從管道讀取數(shù)據(jù)，這樣便構(gòu)成了一種生產(chǎn)者/消費(fèi)者的緩沖區(qū)編程模式。 下面是一個例子代碼，在這個代碼我沒有使用Object對象，而是簡單的讀寫字節(jié)值，這是因?yàn)镻ipedInputStream/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;//private ObjectOutputStream oos;//private ObjectInputStream ois;public Sycn4()...{try...{pos = new PipedOutputStream();pis = new PipedInputStream(pos);//oos = new ObjectOutputStream(pos);//ois = new ObjectInputStream(pis);}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();//Object o = new MyObject();//oos.writeObject(o);//oos.flush();//System.out.println("Producer: " + o);}}catch(Exception e)...{//System.out.println(e);e.printStackTrace();}finally...{try...{pos.close();pis.close();//oos.close();//ois.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));//Object o = ois.readObject();//if(o != null)//System.out.println("Consumer: " + o);}}catch(Exception e)...{//System.out.println(e);e.printStackTrace();}finally...{try...{pos.close();pis.close();//oos.close();//ois.close();}catch(IOException e)...{System.out.println(e);}}}}//class MyObject implements Serializable {//} }

?

總結(jié)

以上是生活随笔為你收集整理的(Java生产者消费者问题)http://blog.csdn.net/jhj735412/article/details/6931135的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。