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

while (mylist.size() == 0)

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

object o = mylist.removelast();

system.out.println("consumer: " + o);

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

二、方法await()/signal()

在jdk5.0以後，java提供了新的更加健壯的線程處理機制，包括了同步、鎖定、線程池等等，可以實作更小粒度上的控制。await()和signal()就是其中用來同步的兩種方法，功能基本上和wait()/notify()相同，完全可以取代它們，但是它們和新引入的鎖定機制lock直接挂鈎，具有更大的靈活性。

import java.util.concurrent.locks.*;

public class sycn2

private final lock lock = new reentrantlock();

private final condition full = lock.newcondition();

private final condition empty = lock.newcondition();

sycn2 s2 = new sycn2();

s2.start();

lock.lock();

full.await();

empty.signal();

finally

lock.unlock();

empty.await();

full.signal();

三、阻塞隊列方法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);

sycn3 s3 = new sycn3();

s3.start();

if (queue.size() == max)

queue.put(o);

catch (interruptedexception e)

if (queue.size() == 0)

object o = queue.take();

運作一下代碼發現問題：

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的解釋：http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4131126import java.io.*;

public class sycn4

private pipedoutputstream pos;

private pipedinputstream pis;

public sycn4()

pos = new pipedoutputstream();

pis = new pipedinputstream(pos);

catch (ioexception e)

system.out.println(e);

new producer().start();

new consumer().start();

sycn4 s4 = new sycn4();

s4.start();

class producer extends thread

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();

pos.close();

pis.close();

class consumer extends thread

int b = pis.read();

system.out.println("consumer: a byte, the value is " + 

                                        string.valueof(b));