标签：
PipedInputStream类与PipedOutputStream类用于在应用程序中创建管道通信.一个PipedInputStream实例对象必须和一个PipedOutputStream实例对象进行连接而产生一个通信管道.PipedOutputStream可以向管道中写入数据,PipedIntputStream可以读取PipedOutputStream向管道中写入的数据.这两个类主要用来完成线程之间的通信.一个线程的PipedInputStream对象能够从另外一个线程的PipedOutputStream对象中读取数据.如下图所示。
PipedInputStream和PipedOutputStream的实现原理类似于"生产者-消费者"原理,PipedOutputStream是生产者,PipedInputStream是消费者,在PipedInputStream中有一个buffer字节数组,默认大小为1024,作为缓冲区,存放"生产者"生产出来的东西.还有两个变量in和out。in是用来记录"生产者"生产了多少,out是用来记录"消费者"消费了多少,in为-1表示消费完了,in==out表示生产满了.当消费者没东西可消费的时候,也就是当in为-1的时候,消费者会一直等待,直到有东西可消费.
1.创建连接与初始化
在两者的构造函数中，都相互提供了连接的构造方法，分别用于接收对方的管道实例，然后调用各自的connect()方法进行连接，如PipedInputStream：
private static final int DEFAULT_PIPE_SIZE = 1024; protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE; // 省略部分构造函数 public PipedInputStream(PipedOutputStream src, int pipeSize) throws IOException { initPipe(pipeSize); connect(src); } private void initPipe(int pipeSize) { // 初始化buffer的大小,pipeSize可以通过构造方法指定，也可以使用默认的PIPE_SIZE的大小 if (pipeSize <= 0) { throw new IllegalArgumentException("Pipe Size <= 0"); } buffer = new byte[pipeSize]; // 初始化缓冲区的大小 } public void connect(PipedOutputStream src) throws IOException { src.connect(this); // 连接输入管道 }
看一下PipedOutputStream的构造函数，如下：
private PipedInputStream sink; public PipedOutputStream(PipedInputStream snk) throws IOException { connect(snk); // 连接输出管理 } public PipedOutputStream() { } public synchronized void connect(PipedInputStream snk) throws IOException { if (snk == null) { throw new NullPointerException(); } else if (sink != null || snk.connected) { throw new IOException("Already connected"); } sink = snk; snk.in = -1; // buffer数组中无数据 snk.out = 0; // 取出的数据为0 snk.connected = true;// 表示连接成功 }
可以看到，相互之间会调用connect()方法来连接，其效果是一样的。
2.数据的写入与读取
连接成功就，就可以进行数据的写入与读出操作了，在PipedOutputStream中的write()写法如下：
public void write(int b) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } sink.receive(b); // 调用receive方法进行数据的输出 } public void write(byte b[], int off, int len) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } else if (b == null) { throw new NullPointerException(); } else if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) > b.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return; } sink.receive(b, off, len); }
方法在写入到byte[]数组缓存区数据后，就会调用PipedInputStream中的receive方法。输出管道中的receive()方法如下：
protected synchronized void receive(int b) throws IOException { // 只会在PipedOutputStream类的write()中调用 checkStateForReceive(); writeSide = Thread.currentThread(); if (in == out) // in==out表示buffer数组已满 awaitSpace(); // 等待空闲空间 if (in < 0) { // 输入管道无数据 in = 0; out = 0; } buffer[in++] = (byte)(b & 0xFF); if (in >= buffer.length) { in = 0; // 缓冲区已经满了，等待下一次从头写入 } } synchronized void receive(byte b[], int off, int len) throws IOException { // 将下标off开始的len个数组数据写入到输出管道中 checkStateForReceive(); // 检查管道的状态 writeSide = Thread.currentThread(); // 获取写入线程 int bytesToTransfer = len; while (bytesToTransfer > 0) { if (in == out) // 缓冲数组已满，只能等待 awaitSpace(); int nextTransferAmount = 0; if (out < in) { nextTransferAmount = buffer.length - in; } else if (in < out) { if (in == -1) { in = out = 0; nextTransferAmount = buffer.length - in; } else { nextTransferAmount = out - in; } } if (nextTransferAmount > bytesToTransfer) nextTransferAmount = bytesToTransfer; assert(nextTransferAmount > 0); // nextTransferAmount<=0，则终止程序的执行 System.arraycopy(b, off, buffer, in, nextTransferAmount);// 拷贝数组中数据到缓冲区 bytesToTransfer -= nextTransferAmount; off += nextTransferAmount; in += nextTransferAmount; if (in >= buffer.length) { in = 0; } } }
输入管理通过如上的对应方法接收到数据并保存到输入缓冲区后，下面就可以使用read()方法读出这些数据了。看一下awaitSpace()方法的实现源代码：
/* * 若写入管道的数据正好全部被读取完(例如,管道缓冲满),则执行awaitSpace()操作; * 让读取管道的线程管道产生读取数据请求,从而才能继续的向“管道”中写入数据 */ private void awaitSpace() throws IOException { /* * 如果管道中被读取的数据,等于写入管道的数据时, * 则每隔1000ms检查“管道状态”,并唤醒管道操作：若有读取管道数据线程被阻塞,则唤醒该线程 */ while (in == out) { checkStateForReceive(); /* full: kick any waiting readers */ notifyAll(); try { wait(1000); } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } } }
PipedInputStream类中的read()方法源代码如下：
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) { // 如果in<0(表示管道中无数据)且closedByWriter为true（表示输入管道已经关闭）则直接返回-1 return -1; } if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) { throw new IOException("Pipe broken"); } 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; } public synchronized int read(byte b[], int off, int len) throws IOException { if (b == null) { throw new NullPointerException(); } else if (off < 0 || len < 0 || len > b.length - off) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return 0; } /* possibly wait on the first character */ int c = read(); if (c < 0) { return -1; } b[off] = (byte) c; int rlen = 1; while ((in >= 0) && (len > 1)) { int available; if (in > out) { available = Math.min((buffer.length - out), (in - out)); } else { available = buffer.length - out; } // A byte is read beforehand outside the loop if (available > (len - 1)) { available = len - 1; } System.arraycopy(buffer, out, b, off + rlen, available); out += available; rlen += available; len -= available; if (out >= buffer.length) { out = 0; } if (in == out) { in = -1; } } return rlen; }
3.刷新与关闭管道
来看一下管道输出流PipedOutputStream中的刷新和关闭方法，源代码如下：
// 刷回管道输出流 public synchronized void flush() throws IOException { if (sink != null) { synchronized (sink) { /* * 调用管道输入流的notifyAll(),通知管道输入流放弃对当前资源的占有， * 让其它的等待线程(等待读取管道输出流的线程)读取管道输出流的值。 */ sink.notifyAll(); } } } // 关闭管道输出流 public void close() throws IOException { if (sink != null) { sink.receivedLast();// 通知管道输入流,输出管理已经关闭 } }
看一下receivedLast()方法PipedInputStream，如下：
synchronized void receivedLast() { closedByWriter = true; // 输出管道标志为true，表示关闭 notifyAll(); // 唤醒所有的等待线程 }
通知所有的等待线程，最后的数据已经全部到达。
public void close() throws IOException { // 关闭管道输出流 closedByReader = true; synchronized (this) { in = -1; // 清空缓冲区数据 } }
PipedInputStream源码完整：
package IO; import java.io.IOException; import java.io.InputStream; import java.io.PipedOutputStream; public class PipedInputStream extends InputStream { boolean closedByWriter = false; // 标识有读取方或写入方关闭 volatile boolean closedByReader = false; boolean connected = false; // 是否建立连接 Thread readSide; // 标识哪个线程 Thread writeSide; protected static final int PIPE_SIZE = 1024; // 缓冲区的默认大小 protected byte buffer[] = new byte[PIPE_SIZE]; // 缓冲区 protected int in = -1; // 下一个写入字节的位置。0代表空，in==out代表满 protected int out = 0; // 下一个读取字节的位置 public PipedInputStream(PipedOutputStream src) throws IOException { // 给定源的输入流 connect(src); } public PipedInputStream() { } // 默认构造器，下部一定要connect源 public void connect(PipedOutputStream src) throws IOException { // 连接输入源 src.connect(this); // 调用源的connect方法连接当前对象 } protected synchronized void receive(int b) throws IOException { // 只被PipedOuputStream调用 checkStateForReceive(); // 检查状态，写入 writeSide = Thread.currentThread(); // 永远是PipedOuputStream if (in == out) awaitSpace(); // 输入和输出相等，等待空间 if (in < 0) { in = 0; out = 0; } buffer[in++] = (byte) (b & 0xFF); // 放入buffer相应的位置 if (in >= buffer.length) { in = 0; } // in为0表示buffer已空 } synchronized void receive(byte b[], int off, int len) throws IOException { checkStateForReceive(); writeSide = Thread.currentThread(); // 从PipedOutputStream可以看出 int bytesToTransfer = len; while (bytesToTransfer > 0) { if (in == out) awaitSpace(); // 满了，会通知读取的；空会通知写入 int nextTransferAmount = 0; if (out < in) { nextTransferAmount = buffer.length - in; } else if (in < out) { if (in == -1) { in = out = 0; nextTransferAmount = buffer.length - in; } else { nextTransferAmount = out - in; } } if (nextTransferAmount > bytesToTransfer) nextTransferAmount = bytesToTransfer; assert (nextTransferAmount > 0); System.arraycopy(b, off, buffer, in, nextTransferAmount); bytesToTransfer -= nextTransferAmount; off += nextTransferAmount; in += nextTransferAmount; if (in >= buffer.length) { in = 0; } } } private void checkStateForReceive() throws IOException { // 检查当前状态，等待输入 if (!connected) { throw new IOException("Pipe not connected"); } else if (closedByWriter || closedByReader) { throw new IOException("Pipe closed"); } else if (readSide != null && !readSide.isAlive()) { throw new IOException("Read end dead"); } } private void awaitSpace() throws IOException { // Buffer已满，等待一段时间 while (in == out) { // in==out表示满了，没有空间 checkStateForReceive(); // 检查接受端的状态 notifyAll(); // 通知读取端 try { wait(1000); } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } } } synchronized void receivedLast() { // 通知所有等待的线程（）已经接受到最后的字节 closedByWriter = true; // notifyAll(); } 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) { return -1; } // 输入断关闭返回-1 if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) { // 状态错误 throw new IOException("Pipe broken"); } 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) { in = -1; } // 没有任何字节 return ret; } public synchronized int read(byte b[], int off, int len) throws IOException { if (b == null) { // 检查输入参数的正确性 throw new NullPointerException(); } else if (off < 0 || len < 0 || len > b.length - off) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return 0; } int c = read(); // 读取下一个 if (c < 0) { return -1; } // 已经到达末尾了，返回-1 b[off] = (byte) c; // 放入外部buffer中 int rlen = 1; // return-len while ((in >= 0) && (--len > 0)) { // 下一个in存在，且没有到达len b[off + rlen] = buffer[out++]; // 依次放入外部buffer rlen++; if (out >= buffer.length) { out = 0; } // 读到buffer的末尾，返回头部 if (in == out) { in = -1; } // 读、写位置一致时，表示没有数据 } return rlen; // 返回填充的长度 } public synchronized int available() throws IOException { // 返回还有多少字节可以读取 if (in < 0) return 0; // 到达末端，没有字节 else if (in == out) return buffer.length; // 写入的和读出的一致，表示满 else if (in > out) return in - out; // 写入的大于读出 else return in + buffer.length - out; // 写入的小于读出的 } public void close() throws IOException { // 关闭当前流，同时释放与其相关的资源 closedByReader = true; // 表示由输入流关闭 synchronized (this) { in = -1; } // 同步化当前对象，in为-1 } }
PipedOutputStream源码完整：
package IO; import java.io.OutputStream; public class PipedOutputStream extends OutputStream { private PipedInputStream sink; // 包含一个PipedInputStream public PipedOutputStream(PipedInputStream snk) throws IOException { // 带有目的地的构造器 connect(snk); } public PipedOutputStream() { } // 默认构造器，必须使用下面的connect方法连接 public synchronized void connect(PipedInputStream snk) throws IOException { if (snk == null) { // 检查输入参数的正确性 throw new NullPointerException(); } else if (sink != null || snk.connected) { throw new IOException("Already connected"); } sink = snk; // 一系列初始化工作 snk.in = -1; snk.out = 0; snk.connected = true; } public void write(int b) throws IOException { // 向流中写入数据 if (sink == null) { throw new IOException("Pipe not connected"); } sink.receive(b); // 本质上是，调用PipedInputStream的receive方法接受此字节 } public void write(byte b[], int off, int len) throws IOException { if (sink == null) { // 首先检查输入参数的正确性 throw new IOException("Pipe not connected"); } else if (b == null) { throw new NullPointerException(); } else if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) > b.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return; } sink.receive(b, off, len); // 调用PipedInputStream的receive方法接受 } public synchronized void flush() throws IOException { // flush输出流 if (sink != null) { synchronized (sink) { sink.notifyAll(); } // 本质是通知输入流，可以读取 } } public void close() throws IOException { // 关闭流同时释放相关资源 if (sink != null) { sink.receivedLast(); } } }
下面来具体举一个例子，如下：
public class test04 { public static void main(String [] args) { Sender sender = new Sender(); Receiver receiver = new Receiver(); PipedOutputStream outStream = sender.getOutStream(); PipedInputStream inStream = receiver.getInStream(); try { //inStream.connect(outStream); // 与下一句一样 outStream.connect(inStream); } catch (Exception e) { e.printStackTrace(); } sender.start(); receiver.start(); } } class Sender extends Thread { private PipedOutputStream outStream = new PipedOutputStream(); public PipedOutputStream getOutStream() { return outStream; } public void run() { String info = "hello, receiver"; try { outStream.write(info.getBytes()); outStream.close(); } catch (Exception e) { e.printStackTrace(); } } } class Receiver extends Thread { private PipedInputStream inStream = new PipedInputStream(); public PipedInputStream getInStream() { return inStream; } public void run() { byte[] buf = new byte[1024]; try { int len = inStream.read(buf); System.out.println("receive message from sender : " + new String(buf, 0, len)); inStream.close(); } catch (Exception e) { e.printStackTrace(); } } }
最后运行后输出的结果如下：receive message from sender : hello, receiver
PipedInputStream和PipedOutputStream详解
标签：
原文地址：http://www.cnblogs.com/wxgblogs/p/5659604.html
