生產者與消費者java

⑴ java實現生產者和消費者問題的幾種方式

生產者消費者問題是多線程的一個經典問題，它描述是有一塊緩沖區作為倉庫，生產者可以將產品放入倉庫，消費者則可以從倉庫中取走產品。
解決生產者/消費者問題的方法可分為兩類：
採用某種機制保護生產者和消費者之間的同步；
在生產者和消費者之間建立一個管道。
第一種方式有較高的效率，並且易於實現，代碼的可控制性較好，屬於常用的模式。第二種管道緩沖區不易控制，被傳輸數據對象不易於封裝等，實用性不強。
在Java中有四種方法支持同步，其中前三個是同步方法，一個是管道方法。
wait()
/
notify()方法
await()
/
signal()方法
BlockingQueue阻塞隊列方法
PipedInputStream
/
PipedOutputStream
通過
wait()
/
notify()方法實現：
wait()
/
nofity()方法是基類Object的兩個方法：
wait()方法：當緩沖區已滿/空時，生產者/消費者線程停止自己的執行，放棄鎖，使自己處於等等狀態，讓其他線程執行。
notify()方法：當生產者/消費者向緩沖區放入/取出一個產品時，向其他等待的線程發出可執行的通知，同時放棄鎖，使自己處於等待狀態。
通過await()
/
signal()方法實現：
await()和signal()的功能基本上和wait()
/
nofity()相同，完全可以取代它們，但是它們和新引入的鎖定機制Lock直接掛鉤，具有更大的靈活性。通過在Lock對象上調用newCondition()方法，將條件變數和一個鎖對象進行綁定，進而控制並發程序訪問競爭資源的安全。
通過BlockingQueue方法實現：
它是一個已經在內部實現了同步的隊列，實現方式採用的是我們第2種await()
/
signal()方法。它可以在生成對象時指定容量大小。它用於阻塞操作的是put()和take()方法：
put()方法：類似於我們上面的生產者線程，容量達到最大時，自動阻塞。
take()方法：類似於我們上面的消費者線程，容量為0時，自動阻塞。

⑵ JAVA模擬生產者與消費者實例

使用的生產者和消費者模型具有如下特點：
(1)本實驗的多個緩沖區不是環形循環的，也不要求按順序訪問。生產者可以把產品放到目前某一個空緩沖區中。
(2)消費者只消費指定生產者的產品。
(3)在測試用例文件中指定了所有的生產和消費的需求，只有當共享緩沖區的數據滿足了所有關於它的消費需求後，此共享緩沖區才可以作為空閑空間允許新的生產者使用。
(4)本實驗在為生產者分配緩沖區時各生產者間必須互斥，此後各個生產者的具體生產活動可以並發。而消費者之間只有在對同一產品進行消費時才需要互斥，同時它們在消費過程結束時需要判斷該消費對象是否已經消費完畢並清除該產品。
Windows
用來實現同步和互斥的實體。在Windows
中，常見的同步對象有：信號量(Semaphore)、
互斥量(Mutex)、臨界段(CriticalSection)和事件(Event)等。本程序中用到了前三個。使用這些對象都分
為三個步驟，一是創建或者初始化：接著請求該同步對象，隨即進入臨界區，這一步對應於互斥量的
上鎖；最後釋放該同步對象，這對應於互斥量的解鎖。這些同步對象在一個線程中創建，在其他線程
中都可以使用，從而實現同步互斥。當然，在進程間使用這些同步對象實現同步的方法是類似的。
1．用鎖操作原語實現互斥
為解決進程互斥進人臨界區的問題，可為每類臨界區設置一把鎖，該鎖有打開和關閉兩種狀態，進程執行臨界區程序的操作按下列步驟進行：
①關鎖。先檢查鎖的狀態，如為關閉狀態，則等待其打開；如已打開了，則將其關閉，繼續執行步驟②的操作。
②執行臨界區程序。
③開鎖。將鎖打開，退出臨界區。
2．信號量及WAIT，SIGNAL操作原語
信號量的初值可以由系統根據資源情況和使用需要來確定。在初始條件下信號量的指針項可以置為0，表示隊列為空。信號量在使用過程中它的值是可變的，但只能由WAIT，SIGNAL操作來改變。設信號量為S，對S的WAIT操作記為WAIT（S），對它的SIGNAL操作記為SIGNAL（S）。
WAIT（S）：順序執行以下兩個動作：
①信號量的值減1，即S=S－1；
②如果S≥0，則該進程繼續執行；
如果
S（0，則把該進程的狀態置為阻塞態，把相應的WAITCB連人該信號量隊列的末尾，並放棄處理機，進行等待（直至其它進程在S上執行SIGNAL操作，把它釋放出來為止）。
SIGNAL（S）：順序執行以下兩個動作
①S值加
1，即
S=S＋1；
②如果S）0，則該進程繼續運行；
如果S（0則釋放信號量隊列上的第一個PCB（既信號量指針項所指向的PCB）所對應的進程（把阻塞態改為就緒態），執行SIGNAL操作的進程繼續運行。
在具體實現時注意，WAIT，SIGNAL操作都應作為一個整體實施，不允許分割或相互穿插執行。也就是說，WAIT，SIGNAL操作各自都好像對應一條指令，需要不間斷地做下去，否則會造成混亂。
從物理概念上講，信號量S）時，S值表示可用資源的數量。執行一次WAIT操作意味著請求分配一個單位資源，因此S值減1；當S<0時，表示已無可用資源，請求者必須等待別的進程釋放了該類資源，它才能運行下去。所以它要排隊。而執行一次SIGNAL操作意味著釋放一個單位資源，因此S值加1；若S（0時，表示有某些進程正在等待該資源，因而要把隊列頭上的進程喚醒，釋放資源的進程總是可以運行下去的。
---------------
/**
*
生產者
*
*/
public
class
Procer
implements
Runnable{
private
Semaphore
mutex,full,empty;
private
Buffer
buf;
String
name;
public
Procer(String
name,Semaphore
mutex,Semaphore
full,Semaphore
empty,Buffer
buf){
this.mutex
=
mutex;
this.full
=
full;
this.empty
=
empty;
this.buf
=
buf;
this.name
=
name;
}
public
void
run(){
while(true){
empty.p();
mutex.p();
System.out.println(name+"
inserts
a
new
proct
into
"+buf.nextEmptyIndex);
buf.nextEmptyIndex
=
(buf.nextEmptyIndex+1)%buf.size;
mutex.v();
full.v();
try
{
Thread.sleep(1000);
}
catch
(InterruptedException
e)
{
e.printStackTrace();
}
}
}
}
---------------
/**
*
消費者
*
*/
public
class
Customer
implements
Runnable{
private
Semaphore
mutex,full,empty;
private
Buffer
buf;
String
name;
public
Customer(String
name,Semaphore
mutex,Semaphore
full,Semaphore
empty,Buffer
buf){
this.mutex
=
mutex;
this.full
=
full;
this.empty
=
empty;
this.buf
=
buf;
this.name
=
name;
}
public
void
run(){
while(true){
full.p();
mutex.p();
System.out.println(name+"
gets
a
proct
from
"+buf.nextFullIndex);
buf.nextFullIndex
=
(buf.nextFullIndex+1)%buf.size;
mutex.v();
empty.v();
try
{
Thread.sleep(1000);
}
catch
(InterruptedException
e)
{
e.printStackTrace();
}
}
}
}
-------------------------
/**
*
緩沖區
*
*/
public
class
Buffer{
public
Buffer(int
size,int
nextEmpty,int
nextFull){
this.nextEmptyIndex
=
nextEmpty;
this.nextFullIndex
=
nextFull;
this.size
=
size;
}
public
int
size;
public
int
nextEmptyIndex;
public
int
nextFullIndex;
}
-----------------
/**
*
此類用來模擬信號量
*
*/
public
class
Semaphore{
private
int
semValue;
public
Semaphore(int
semValue){
this.semValue
=
semValue;
}
public
synchronized
void
p(){
semValue--;
if(semValue<0){
try
{
this.wait();
}
catch
(InterruptedException
e)
{
e.printStackTrace();
}
}
}
public
synchronized
void
v(){
semValue++;
if(semValue<=0){
this.notify();
}
}
}
------------------------
public
class
Test
extends
Thread
{
public
static
void
main(String[]
args)
{
Buffer
bf=new
Buffer(10,0,0);
Semaphore
mutex=new
Semaphore(1);
Semaphore
full=new
Semaphore(0);
Semaphore
empty=new
Semaphore(10);
//new
Thread(new
Procer("p001",mutex,full,empty,bf)).start();
Procer
p=new
Procer("p001",mutex,full,empty,bf);
new
Thread(new
Procer("p002",mutex,full,empty,bf)).start();
new
Thread(new
Procer("p003",mutex,full,empty,bf)).start();
new
Thread(new
Procer("p004",mutex,full,empty,bf)).start();
new
Thread(new
Procer("p005",mutex,full,empty,bf)).start();
try{
sleep(3000);
}
catch(Exception
ex)
{
ex.printStackTrace();
}
new
Thread(new
Customer("c001",mutex,full,empty,bf)).start();
new
Thread(new
Customer("c002",mutex,full,empty,bf)).start();
new
Thread(new
Customer("c003",mutex,full,empty,bf)).start();
new
Thread(new
Customer("c004",mutex,full,empty,bf)).start();
new
Thread(new
Customer("c005",mutex,full,empty,bf)).start();
}
}
--------------------------------------------

⑶ java中生產者和消費者問題

好不好，要看情況，單這樣，提不提都問題不大。

類的架構，由設計者決定。

this 是指當前類的當前實例

⑷ JAVA多生產者多消費者問題。希望用wait()和notify().謝謝！

publicclassProceConsumerDemo{

	publicstaticvoidmain(String[]args){
		//1.創建資源
		Resourceresource=newResource();
		//2.創建兩個任務
		Procerprocer=newProcer(resource);
		Consumerconsumer=newConsumer(resource);
		//3.創建線程
		/*
		*多生產多消費產生的問題:重復生產、重復消費
		*/
		Threadthread0=newThread(procer);
		Threadthread1=newThread(procer);

		thread0.setName("生產者（NO0）");
		thread1.setName("生產者（NO1）");

		Threadthread2=newThread(consumer);
		Threadthread3=newThread(consumer);

		thread2.setName("消費者（NO2）");
		thread3.setName("消費者（NO3）");

		thread0.start();
		thread1.start();
		thread2.start();
		thread3.start();
	}
}

classResource{
	privateStringname;
	privateintcount=1;
	//定義標記
	privatebooleanflag;

	//提供給商品賦值的方法
	publicsynchronizedvoidsetName(Stringname){//thread0,thread1在這里運行
		while(flag)//判斷標記為true，執行wait等待，為false則生產
			/*
			*這里使用while，而不使用if的理由如下：
			*
			*thread0有可能第二次也搶到鎖的執行權，判斷為真，則有麵包不生產，所以接下來執行等待，此時thread0在線程池中。
			*接下來活的線程有3個（除了thread0），這三個線程都有可能獲取到執行權.
			*假設thread1獲得了執行權，判斷為真，則有麵包不生產，執行等待。此時thread1又進入到了線程池中。
			*接下來有兩個活的線程thread2和thread3。假設thread2又搶到了執行權，所以程序轉到了消費get處……
			*/
			try{
				this.wait();//這里wait語句必須包含在try/catch塊中，拋出異常。
			}catch(InterruptedExceptione){
				e.printStackTrace();
			}
		this.name=name+count;//第一個麵包
		count++;//2
		System.out.println(Thread.currentThread().getName()+this.name);//thread0線程生產了麵包1

		//生產完畢，將標記改成true.
		flag=true;//thread0第一次生產完麵包以後，將標記改為真，表示有麵包了
		
		//喚醒消費者（這里使用notifyAll而不使用notify的原因在下面）
		
		this.notifyAll();//第一次在這里是空喚醒，沒有意義
	}

	/*
	*通過同步，解決了沒生產就消費的問題
	*生產完以後，生產者釋放了this鎖，此時，生產者和消費者同時去搶鎖，又是生產者搶到了鎖，所以就出現了一直生產的情況。
	*與「生產一個就消費一個的需求不符合」等待喚醒機制wait();該方法可以使線程處於凍結狀態，並將線程臨時存儲到線程池
	*notify();喚醒指定線程池中的任意一個線程。notifyAll();喚醒指定線程池中的所有線程
	*這些方法必須使用在同步函數中，因為他們用來操作同步鎖上的線程上的狀態的。
	*在使用這些方法時候，必須標識他們所屬於的鎖，標識方式就是鎖對象.wait();鎖對象.notify();鎖對象.notifyAll();
	*相同鎖的notify()可以獲取相同鎖的wait();
	*/

	publicsynchronizedvoidgetName(){//thread2,thread3在這里運行
		while(!flag)
			/*
			*……接著上面的程序執行分析thread2拿到鎖獲取執行權之後，判斷！flag為假，則不等待，直接消費麵包1，輸出一次.
			*消費完成之後將flag改為假接下來又喚醒了thread0或者thread1生產者中的一個
			*假設又喚醒了thread0線程，現在活的線程有thread0，thread2，thread3三個線程
			*假設接下來thread2又搶到了執行權，判斷！flag為真，沒麵包了，停止消費，所以thread2執行等待.
			*此時活著的線程有thread0和thread3。
			*假設thread3得到了執行權，拿到鎖之後進來執行等待，此時活著的線程只有thread0.
			*所以thread0隻能搶到執行權之後，生產麵包2，將標記改為true告訴消費者有麵包可以消費了。
			*接下來執行notify喚醒，此時喚醒休眠中的3個線程中的任何一個都有可能。
			*如果喚醒了消費者thread2或者thread3中的任何一個，程序都是正常。如果此時喚醒thread1則不正常。
			*如果喚醒了thread1，此時活著的線程有thread0和thread1兩個線程。
			*假設thread0又獲得了執行權，判讀為真有麵包，則又一次執行等待。
			*接下來只有thread1線程有執行權（此時沒有判斷標記直接生產了，出錯了），所以又生產了麵包3。在這個過程中，麵包2沒有被消費。
			*這就是連續生產和消費容易出現的問題。
			*
			*原因：被喚醒的線程沒有判斷標記就開始執行了，導致了重復的生產和消費發生。
			*
			*解決：被喚醒的線程必須判斷標記，使用while循環標記，而不使用if判斷的理由。
			*
			*但是接下來會出現死鎖，原因在於：
			*上面的程序中thread0在執行notify的時候喚醒了thread1，而此時thread2和thread3兩個消費者線程都處於等待狀態
			*thread1在執行while判斷語句之後判斷為真，則執行等待，此時所有的線程都處於凍結等待狀態了。
			*
			*原因:本方線程在執行喚醒的時候又一次喚醒了本方線程，而本方線程循環判斷標記又繼續等待，而導致所有的線程都等待。
			*
			*解決：本方線程喚醒對方線程，可以使用notifyAll()方法
			*		喚醒之後，既有本方，又有對方，但是本方線程判斷標記之後，會繼續等待，這樣就有對方線程在執行。
			*/
			try{
				this.wait();
			}catch(InterruptedExceptione){
				e.printStackTrace();
			}
		System.out.println(Thread.currentThread().getName()+this.name);
		//將標記改為false
		flag=false;
		//喚醒生產者
		this.notify();
	}
}

//生產者
classProcerimplementsRunnable{
	privateResourceresource;

	publicProcer(Resourceresource){
		this.resource=resource;
	}

	publicvoidrun(){
		while(true){
			resource.setName("麵包");
		}
	}
}

//消費者
{
	privateResourceresource;

	publicConsumer(Resourceresource){
		this.resource=resource;
	}

	@Override
	publicvoidrun(){
		while(true){
			resource.getName();
		}
	}
}

⑸ java生產者消費者問題

這是因為你的日誌並不同步所導致的。生產確實在消費之前，但是生產日誌卻在消費之後皮派純。你生產的push方法和消費的pop方法是加了synchronized同步的，但是列印卻不在同步的塊中。所以很有可能是在生產線程執行完push之後cpu就被消費者搶佔了，燃咐直羨州到列印出了消費信息之後才還給生產者列印生產信息。

⑹ 由生產者/消費者問題看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 Sycn {

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

private int MAX = ;

public Sycn (){

}

public void start(){

new Procer() start();

new Consumer() start();

}

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

Sycn s = new Sycn ();

s start();

}

class Procer 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( Procer: + o);

myList notify();

}

}catch(InterruptedException ie){

System out println( procer is interrupted! );

}

}

}

}

}

class Consumer extends Thread{

public void run(){

while(true){

synchronized(myList){

try{

while(myList size() == ){

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

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

下面是一個例子代碼

import java util LinkedList;

import ncurrent locks *;

public class Sycn {

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

private int MAX = ;

private final Lock lock = new ReentrantLock();

private final Condition full = lock newCondition();

private final Condition empty = lock newCondition();

public Sycn (){

}

public void start(){

new Procer() start();

new Consumer() start();

}

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

Sycn s = new Sycn ();

s start();

}

class Procer 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( Procer: + o);

empty signal();

}

}catch(InterruptedException ie){

System out println( procer is interrupted! );

}finally{

lock unlock();

}

}

}

}

class Consumer extends Thread{

public void run(){

while(true){

lock lock();

try{

while(myList size() == ){

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也是JDK 的一部分 它是一個已經在內部實現了同步的隊列 實現方式採用的是我們的第 種await()/signal()方法 它可以在生成對象時指定容量大小

它用於阻塞操作的是put()和take()方法

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

take()方法類似於我們上面的消費者線程 容量為 時 自動阻塞

下面是一個例子代碼

import ncurrent *;

public class Sycn {

private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<Object>( );

private int MAX = ;

public Sycn (){

}

public void start(){

new Procer() start();

new Consumer() start();

}

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

Sycn s = new Sycn ();

s start();

}

class Procer 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( Procer: + o);

}catch(InterruptedException e){

System out println( procer is interrupted! );

}

//}

}

}

}

class Consumer extends Thread{

public void run(){

while(true){

//synchronized(this){

try{

if(queue size() == )

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

Object o = queue take();

System out println( Consumer: + o);

}catch(InterruptedException e){

System out println( procer is interrupted! );

}

//}

}

}

}

}

你發現這個例子中的問題了嗎？

如果沒有 我建議你運行一下這段代碼 仔細觀察它的輸出 是不是有下面這個樣子的？為什麼會這樣呢？

…

warning: it s full!

Procer: java lang object@ e a

…

你可能會說這是因為put()和System out println()之間沒有同步造成的 我也這樣認為 我也這樣認為 但是你把run()中的synchronized前面的注釋去掉 重新編譯運行 有改觀嗎？沒有 為什麼？

這是因為 當緩沖區已滿 生產者在put()操作時 put()內部調用了await()方法 放棄了線程的執行 然後消費者線程執行 調用take()方法 take()內部調用了signal()方法 通知生產者線程可以執行 致使在消費者的println()還沒運行的情況下生產者的println()先被執行 所以有了上面的輸出 run()中的synchronized其實並沒有起什麼作用

對於BlockingQueue大家可以放心使用 這可不是它的問題 只是在它和別的對象之間的同步有問題

對於這種多重嵌套同步的問題 以後再談吧 歡迎大家討論啊！

管道方法PipedInputStream/PipedOutputStream

這個類位於java io包中 是解決同步問題的最簡單的辦法 一個線程將數據寫入管道 另一個線程從管道讀取數據 這樣便構成了一種生產者/消費者的緩沖區編程模式

下面是一個例子代碼 在這個代碼我沒有使用Object對象 而是簡單的讀寫位元組值 這是因為PipedInputStream/PipedOutputStream不允許傳輸對象 這是JAVA本身的一個bug 具體的大家可以看sun的解釋 _bug do?bug_id=

import java io *;

public class Sycn {

private PipedOutputStream pos;

private PipedInputStream pis;

//private ObjectOutputStream oos;

//private ObjectInputStream ois;

public Sycn (){

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

new Consumer() start();

}

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

Sycn s = new Sycn ();

s start();

}

class Procer extends Thread{

public void run() {

try{

while(true){

int b = (int) (Math random() * );

System out println( Procer: a byte the value is + b);

pos write(b);

pos flush();

//Object o = new MyObject();

//oos writeObject(o);

//oos flush();

//System out println( Procer: + 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 {

//}

lishixin/Article/program/Java/gj/201311/27617