IO--PipedInputSteamAndPipedOutputStream

PipedInputStream和PipedOutputStream在两个线程之间建立通信的管道比如线程A持有PipedOutputStream os，线程B持有PipedInputStream is，可以用os绑定is也可以用is绑定os。然后os写入数据，is读取数据。import java.io.IOException;import java.io.PipedInputStream;import java.io.PipedOutputStream;/*** Created by marsares on 15/7/9.*/public class Receiver extends Thread{private PipedInputStream in=new PipedInputStream();public PipedInputStream getPipedInputStream(){return in;}public Receiver(){}public Receiver(Sender s)throws IOException{in=new PipedInputStream(s.getPipedOutputStream());}public void run(){try{int data;while((data=in.read())!=-1){System.out.println(data);}in.close();}catch(IOException e){e.printStackTrace();}}public static void main(String[]args)throws IOException{Receiver r=new Receiver();Sender s=new Sender(r);s.start();r.start();}}class Sender extends Thread{private PipedOutputStream out;public PipedOutputStream getPipedOutputStream(){return out;}public Sender(){}public Sender(Receiver r)throws IOException{out=new PipedOutputStream(r.getPipedInputStream());}public void run(){try{for(int i=0;i<100;i++){out.write(i);//yield();}out.close();}catch (IOException e){e.printStackTrace();}}}从PipedOutputStream和PipedInputStream源码中可以看出这两者的绑定是对称的，只要有一个在初始化的时候绑定另一个就行了。PipedInputStreampublic PipedInputStream(PipedOutputStream src, int pipeSize)throws IOException {initPipe(pipeSize);connect(src);}PipedOutputStreampublic PipedOutputStream(PipedInputStream snk) throws IOException {connect(snk);}PipedOutputStream的write方法调用的PipedInputStream的receive方法，管道流本质上就是在PipedInputStream中的byte[]buffer中读和写。

public void write(int b)  throws IOException {
if (sink == null) {
throw new IOException("Pipe not connected");
}
sink.receive(b);
}

但是这就产生一个疑问，如果只是在PipedInputStream中的buffer进行读和写，为什么需要PipedOutputStream呢？其实管道流还有个重要的特点就是在两个线程之间建立通信通道，如果一个线程关闭了管道能够及时通知另一个线程不再等待。PipedOutputStream关闭管道public void close() throws IOException {if (sink != null) {sink.receivedLast();}}如果buffer中的数据读完并且管道关闭那么PipedInputStream跳出read。如果buffer中数据读完但是管道没有关闭，PipedInputStream将进入阻塞状态等待写进程写入数据。

public synchronized int read()  throws IOException {
if (!connected) {
throw new IOException("Pipe not connected");
} else if (closedByReader) {
throw new IOException("Pipe closed");
} else if (writeSide != null && !writeSide.isAlive()
&& !closedByWriter && (in < 0)) {
throw new IOException("Write end dead");
}

readSide = Thread.currentThread();
int trials = 2;
while (in < 0) {
if (closedByWriter) {
/* closed by writer, return EOF */
return -1;
}
if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) {
throw new IOException("Pipe broken");
}
/* might be a writer waiting */
notifyAll();
try {
wait(1000);
} catch (InterruptedException ex) {
throw new java.io.InterruptedIOException();
}
}
int ret = buffer[out++] & 0xFF;
if (out >= buffer.length) {
out = 0;
}
if (in == out) {
/* now empty */
in = -1;
}

return ret;
}

如果不使用PipedOutputStream就需要用户自己设定closedByWriter和closedByReader参数，造成了紧耦耦合。