java io系列04之 管道(PipedOutputStream和PipedInputStream)的简介,源码分析和示例

 1 package java.io;
 2 
 3 import java.io.*;  4  5 public class PipedOutputStream extends OutputStream {  6  7 // 与PipedOutputStream通信的PipedInputStream对象  8 private PipedInputStream sink;  9 10 // 构造函数，指定配对的PipedInputStream 11 public PipedOutputStream(PipedInputStream snk) throws IOException { 12  connect(snk); 13  } 14 15 // 构造函数 16 public PipedOutputStream() { 17  } 18 19 // 将“管道输出流” 和 “管道输入流”连接。 20 public synchronized void connect(PipedInputStream snk) throws IOException { 21 if (snk == null) { 22 throw new NullPointerException(); 23 } else if (sink != null || snk.connected) { 24 throw new IOException("Already connected"); 25  } 26 // 设置“管道输入流” 27 sink = snk; 28 // 初始化“管道输入流”的读写位置 29 // int是PipedInputStream中定义的，代表“管道输入流”的读写位置 30 snk.in = -1; 31 // 初始化“管道输出流”的读写位置。 32 // out是PipedInputStream中定义的，代表“管道输出流”的读写位置 33 snk.out = 0; 34 // 设置“管道输入流”和“管道输出流”为已连接状态 35 // connected是PipedInputStream中定义的，用于表示“管道输入流与管道输出流”是否已经连接 36 snk.connected = true; 37  } 38 39 // 将int类型b写入“管道输出流”中。 40 // 将b写入“管道输出流”之后，它会将b传输给“管道输入流” 41 public void write(int b) throws IOException { 42 if (sink == null) { 43 throw new IOException("Pipe not connected"); 44  } 45  sink.receive(b); 46  } 47 48 // 将字节数组b写入“管道输出流”中。 49 // 将数组b写入“管道输出流”之后，它会将其传输给“管道输入流” 50 public void write(byte b[], int off, int len) throws IOException { 51 if (sink == null) { 52 throw new IOException("Pipe not connected"); 53 } else if (b == null) { 54 throw new NullPointerException(); 55 } else if ((off < 0) || (off > b.length) || (len < 0) || 56 ((off + len) > b.length) || ((off + len) < 0)) { 57 throw new IndexOutOfBoundsException(); 58 } else if (len == 0) { 59 return; 60  } 61 // “管道输入流”接收数据 62  sink.receive(b, off, len); 63  } 64 65 // 清空“管道输出流”。 66 // 这里会调用“管道输入流”的notifyAll()； 67 // 目的是让“管道输入流”放弃对当前资源的占有，让其它的等待线程(等待读取管道输出流的线程)读取“管道输出流”的值。 68 public synchronized void flush() throws IOException { 69 if (sink != null) { 70 synchronized (sink) { 71 sink.notifyAll(); 72 } 73 } 74 } 75 76 // 关闭“管道输出流”。 77 // 关闭之后，会调用receivedLast()通知“管道输入流”它已经关闭。 78 public void close() throws IOException { 79 if (sink != null) { 80 sink.receivedLast(); 81 } 82 } 83 }

  1 package java.io;
  2 
  3 public class PipedInputStream extends InputStream {  4 // “管道输出流”是否关闭的标记  5 boolean closedByWriter = false;  6 // “管道输入流”是否关闭的标记  7 volatile boolean closedByReader = false;  8 // “管道输入流”与“管道输出流”是否连接的标记  9 // 它在PipedOutputStream的connect()连接函数中被设置为true  10 boolean connected = false;  11  12 Thread readSide; // 读取“管道”数据的线程  13 Thread writeSide; // 向“管道”写入数据的线程  14  15 // “管道”的默认大小  16 private static final int DEFAULT_PIPE_SIZE = 1024;  17  18 protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE;  19  20 // 缓冲区  21 protected byte buffer[];  22  23 //下一个写入字节的位置。in==out代表满，说明“写入的数据”全部被读取了。  24 protected int in = -1;  25 //下一个读取字节的位置。in==out代表满，说明“写入的数据”全部被读取了。  26 protected int out = 0;  27  28 // 构造函数：指定与“管道输入流”关联的“管道输出流”  29 public PipedInputStream(PipedOutputStream src) throws IOException {  30 this(src, DEFAULT_PIPE_SIZE);  31  }  32  33 // 构造函数：指定与“管道输入流”关联的“管道输出流”，以及“缓冲区大小”  34 public PipedInputStream(PipedOutputStream src, int pipeSize)  35 throws IOException {  36  initPipe(pipeSize);  37  connect(src);  38  }  39  40 // 构造函数：默认缓冲区大小是1024字节  41 public PipedInputStream() {  42  initPipe(DEFAULT_PIPE_SIZE);  43  }  44  45 // 构造函数：指定缓冲区大小是pipeSize  46 public PipedInputStream(int pipeSize) {  47  initPipe(pipeSize);  48  }  49  50 // 初始化“管道”：新建缓冲区大小  51 private void initPipe(int pipeSize) {  52 if (pipeSize <= 0) {  53 throw new IllegalArgumentException("Pipe Size <= 0");  54  }  55 buffer = new byte[pipeSize];  56  }  57  58 // 将“管道输入流”和“管道输出流”绑定。  59 // 实际上，这里调用的是PipedOutputStream的connect()函数  60 public void connect(PipedOutputStream src) throws IOException {  61 src.connect(this);  62  }  63  64 // 接收int类型的数据b。  65 // 它只会在PipedOutputStream的write(int b)中会被调用  66 protected synchronized void receive(int b) throws IOException {  67 // 检查管道状态  68  checkStateForReceive();  69 // 获取“写入管道”的线程 70 writeSide = Thread.currentThread(); 71 // 若“写入管道”的数据正好全部被读取完，则等待。 72 if (in == out) 73 awaitSpace(); 74 if (in < 0) { 75 in = 0; 76 out = 0; 77 } 78 // 将b保存到缓冲区 79 buffer[in++] = (byte)(b & 0xFF); 80 if (in >= buffer.length) { 81 in = 0; 82 } 83 } 84 85 // 接收字节数组b。 86 synchronized void receive(byte b[], int off, int len) throws IOException { 87 // 检查管道状态 88 checkStateForReceive(); 89 // 获取“写入管道”的线程 90 writeSide = Thread.currentThread(); 91 int bytesToTransfer = len; 92 while (bytesToTransfer > 0) { 93 // 若“写入管道”的数据正好全部被读取完，则等待。 94 if (in == out) 95 awaitSpace(); 96 int nextTransferAmount = 0; 97 // 如果“管道中被读取的数据，少于写入管道的数据”； 98 // 则设置nextTransferAmount=“buffer.length - in” 99 if (out < in) { 100 nextTransferAmount = buffer.length - in; 101 } else if (in < out) { // 如果“管道中被读取的数据，大于/等于写入管道的数据”，则执行后面的操作 102 // 若in==-1(即管道的写入数据等于被读取数据)，此时nextTransferAmount = buffer.length - in; 103 // 否则，nextTransferAmount = out - in; 104 if (in == -1) { 105 in = out = 0; 106 nextTransferAmount = buffer.length - in; 107 } else { 108 nextTransferAmount = out - in; 109 } 110 } 111 if (nextTransferAmount > bytesToTransfer) 112 nextTransferAmount = bytesToTransfer; 113 // assert断言的作用是，若nextTransferAmount <= 0，则终止程序。 114 assert(nextTransferAmount > 0); 115 // 将数据写入到缓冲中 116 System.arraycopy(b, off, buffer, in, nextTransferAmount); 117 bytesToTransfer -= nextTransferAmount; 118 off += nextTransferAmount; 119 in += nextTransferAmount; 120 if (in >= buffer.length) { 121 in = 0; 122 } 123 } 124 } 125 126 // 检查管道状态 127 private void checkStateForReceive() throws IOException { 128 if (!connected) { 129 throw new IOException("Pipe not connected"); 130 } else if (closedByWriter || closedByReader) { 131 throw new IOException("Pipe closed"); 132 } else if (readSide != null && !readSide.isAlive()) { 133 throw new IOException("Read end dead"); 134 } 135 } 136 137 // 等待。 138 // 若“写入管道”的数据正好全部被读取完(例如，管道缓冲满)，则执行awaitSpace()操作； 139 // 它的目的是让“读取管道的线程”管道产生读取数据请求，从而才能继续的向“管道”中写入数据。 140 private void awaitSpace() throws IOException { 141 142 // 如果“管道中被读取的数据，等于写入管道的数据”时， 143 // 则每隔1000ms检查“管道状态”，并唤醒管道操作：若有“读取管道数据线程被阻塞”，则唤醒该线程。 144 while (in == out) { 145 checkStateForReceive(); 146 147 /* full: kick any waiting readers */ 148 notifyAll(); 149 try { 150 wait(1000); 151 } catch (InterruptedException ex) { 152 throw new java.io.InterruptedIOException(); 153 } 154 } 155 } 156 157 // 当PipedOutputStream被关闭时，被调用 158 synchronized void receivedLast() { 159 closedByWriter = true; 160 notifyAll(); 161 } 162 163 // 从管道(的缓冲)中读取一个字节，并将其转换成int类型 164 public synchronized int read() throws IOException { 165 if (!connected) { 166 throw new IOException("Pipe not connected"); 167 } else if (closedByReader) { 168 throw new IOException("Pipe closed"); 169 } else if (writeSide != null && !writeSide.isAlive() 170 && !closedByWriter && (in < 0)) { 171 throw new IOException("Write end dead"); 172 } 173 174 readSide = Thread.currentThread(); 175 int trials = 2; 176 while (in < 0) { 177 if (closedByWriter) { 178 /* closed by writer, return EOF */ 179 return -1; 180 } 181 if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) { 182 throw new IOException("Pipe broken"); 183 } 184 /* might be a writer waiting */ 185 notifyAll(); 186 try { 187 wait(1000); 188 } catch (InterruptedException ex) { 189 throw new java.io.InterruptedIOException(); 190 } 191 } 192 int ret = buffer[out++] & 0xFF; 193 if (out >= buffer.length) { 194 out = 0; 195 } 196 if (in == out) { 197 /* now empty */ 198 in = -1; 199 } 200 201 return ret; 202 } 203 204 // 从管道(的缓冲)中读取数据，并将其存入到数组b中 205 public synchronized int read(byte b[], int off, int len) throws IOException { 206 if (b == null) { 207 throw new NullPointerException(); 208 } else if (off < 0 || len < 0 || len > b.length - off) { 209 throw new IndexOutOfBoundsException(); 210 } else if (len == 0) { 211 return 0; 212 } 213 214 /* possibly wait on the first character */ 215 int c = read(); 216 if (c < 0) { 217 return -1; 218 } 219 b[off] = (byte) c; 220 int rlen = 1; 221 while ((in >= 0) && (len > 1)) { 222 223 int available; 224 225 if (in > out) { 226 available = Math.min((buffer.length - out), (in - out)); 227 } else { 228 available = buffer.length - out; 229 } 230 231 // A byte is read beforehand outside the loop 232 if (available > (len - 1)) { 233 available = len - 1; 234 } 235 System.arraycopy(buffer, out, b, off + rlen, available); 236 out += available; 237 rlen += available; 238 len -= available; 239 240 if (out >= buffer.length) { 241 out = 0; 242 } 243 if (in == out) { 244 /* now empty */ 245 in = -1; 246 } 247 } 248 return rlen; 249 } 250 251 // 返回不受阻塞地从此输入流中读取的字节数。 252 public synchronized int available() throws IOException { 253 if(in < 0) 254 return 0; 255 else if(in == out) 256 return buffer.length; 257 else if (in > out) 258 return in - out; 259 else 260 return in + buffer.length - out; 261 } 262 263 // 关闭管道输入流 264 public void close() throws IOException { 265 closedByReader = true; 266 synchronized (this) { 267 in = -1; 268 } 269 } 270 }

 1 import java.io.IOException;   
 2    
 3 import java.io.PipedInputStream;  4  5 @SuppressWarnings("all")  6 /**  7  * 接收者线程  8 */  9 public class Receiver extends Thread { 10 11 // 管道输入流对象。 12 // 它和“管道输出流(PipedOutputStream)”对象绑定， 13 // 从而可以接收“管道输出流”的数据，再让用户读取。 14 private PipedInputStream in = new PipedInputStream(); 15 16 // 获得“管道输入流”对象 17 public PipedInputStream getInputStream(){ 18 return in; 19  } 20 21  @Override 22 public void run(){ 23  readMessageOnce() ; 24 //readMessageContinued() ; 25  } 26 27 // 从“管道输入流”中读取1次数据 28 public void readMessageOnce(){ 29 // 虽然buf的大小是2048个字节，但最多只会从“管道输入流”中读取1024个字节。 30 // 因为，“管道输入流”的缓冲区大小默认只有1024个字节。 31 byte[] buf = new byte[2048]; 32 try { 33 int len = in.read(buf); 34 System.out.println(new String(buf,0,len)); 35  in.close(); 36 } catch (IOException e) { 37  e.printStackTrace(); 38  } 39  } 40 // 从“管道输入流”读取>1024个字节时，就停止读取 41 public void readMessageContinued() { 42 int total=0; 43 while(true) { 44 byte[] buf = new byte[1024]; 45 try { 46 int len = in.read(buf); 47 total += len; 48 System.out.println(new String(buf,0,len)); 49 // 若读取的字节总数>1024，则退出循环。 50 if (total > 1024) 51 break; 52 } catch (IOException e) { 53  e.printStackTrace(); 54  } 55  } 56 57 try { 58  in.close(); 59 } catch (IOException e) { 60  e.printStackTrace(); 61  } 62  } 63 }

 1 import java.io.IOException;   
 2    
 3 import java.io.PipedOutputStream;  4 @SuppressWarnings("all")  5 /**  6  * 发送者线程  7 */  8 public class Sender extends Thread {  9 10 // 管道输出流对象。 11 // 它和“管道输入流(PipedInputStream)”对象绑定， 12 // 从而可以将数据发送给“管道输入流”的数据，然后用户可以从“管道输入流”读取数据。 13 private PipedOutputStream out = new PipedOutputStream(); 14 15 // 获得“管道输出流”对象 16 public PipedOutputStream getOutputStream(){ 17 return out; 18  } 19 20  @Override 21 public void run(){ 22  writeShortMessage(); 23 //writeLongMessage(); 24  } 25 26 // 向“管道输出流”中写入一则较简短的消息："this is a short message" 27 private void writeShortMessage() { 28 String strInfo = "this is a short message" ; 29 try { 30  out.write(strInfo.getBytes()); 31  out.close(); 32 } catch (IOException e) { 33  e.printStackTrace(); 34  } 35  } 36 // 向“管道输出流”中写入一则较长的消息 37 private void writeLongMessage() { 38 StringBuilder sb = new StringBuilder(); 39 // 通过for循环写入1020个字节 40 for (int i=0; i<102; i++) 41 sb.append("0123456789"); 42 // 再写入26个字节。 43 sb.append("abcdefghijklmnopqrstuvwxyz"); 44 // str的总长度是1020+26=1046个字节 45 String str = sb.toString(); 46 try { 47 // 将1046个字节写入到“管道输出流”中 48  out.write(str.getBytes()); 49  out.close(); 50 } catch (IOException e) { 51  e.printStackTrace(); 52  } 53  } 54 }

 1 import java.io.PipedInputStream;
 2 import java.io.PipedOutputStream;  3 import java.io.IOException;  4  5 @SuppressWarnings("all")  6 /**  7  * 管道输入流和管道输出流的交互程序  8 */  9 public class PipedStreamTest { 10 11 public static void main(String[] args) { 12 Sender t1 = new Sender(); 13 14 Receiver t2 = new Receiver(); 15 16 PipedOutputStream out = t1.getOutputStream(); 17 18 PipedInputStream in = t2.getInputStream(); 19 20 try { 21 //管道连接。下面2句话的本质是一样。 22 //out.connect(in); 23  in.connect(out); 24 25 /** 26  * Thread类的START方法： 27  * 使该线程开始执行；Java 虚拟机调用该线程的 run 方法。 28  * 结果是两个线程并发地运行；当前线程（从调用返回给 start 方法）和另一个线程（执行其 run 方法）。 29  * 多次启动一个线程是非法的。特别是当线程已经结束执行后，不能再重新启动。 30 */ 31  t1.start(); 32  t2.start(); 33 } catch (IOException e) { 34  e.printStackTrace(); 35  } 36  } 37 }