java io系列20之 PipedReader和PipedWriter

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

  1 package java.io;
  2 
  3 public class PipedReader extends Reader {  4 // “PipedWriter”是否关闭的标记  5 boolean closedByWriter = false;  6 // “PipedReader”是否关闭的标记  7 boolean closedByReader = false;  8 // “PipedReader”与“PipedWriter”是否连接的标记  9 // 它在PipedWriter的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 // 缓冲区  19 char buffer[];  20  21 //下一个写入字符的位置。in==out代表满，说明“写入的数据”全部被读取了。  22 int in = -1;  23 //下一个读取字符的位置。in==out代表满，说明“写入的数据”全部被读取了。  24 int out = 0;  25  26 // 构造函数：指定与“PipedReader”关联的“PipedWriter”  27 public PipedReader(PipedWriter src) throws IOException {  28 this(src, DEFAULT_PIPE_SIZE);  29  }  30  31 // 构造函数：指定与“PipedReader”关联的“PipedWriter”，以及“缓冲区大小”  32 public PipedReader(PipedWriter src, int pipeSize) throws IOException {  33  initPipe(pipeSize);  34  connect(src);  35  }  36  37 // 构造函数：默认缓冲区大小是1024字符  38 public PipedReader() {  39  initPipe(DEFAULT_PIPE_SIZE);  40  }  41  42 // 构造函数：指定缓冲区大小是pipeSize  43 public PipedReader(int pipeSize) {  44  initPipe(pipeSize);  45  }  46  47 // 初始化“管道”：新建缓冲区大小  48 private void initPipe(int pipeSize) {  49 if (pipeSize <= 0) {  50 throw new IllegalArgumentException("Pipe size <= 0");  51  }  52 buffer = new char[pipeSize];  53  }  54  55 // 将“PipedReader”和“PipedWriter”绑定。  56 // 实际上，这里调用的是PipedWriter的connect()函数  57 public void connect(PipedWriter src) throws IOException {  58 src.connect(this);  59  }  60  61 // 接收int类型的数据b。  62 // 它只会在PipedWriter的write(int b)中会被调用  63 synchronized void receive(int c) throws IOException {  64 // 检查管道状态  65 if (!connected) {  66 throw new IOException("Pipe not connected");  67 } else if (closedByWriter || closedByReader) {  68 throw new IOException("Pipe closed");  69 } else if (readSide != null && !readSide.isAlive()) { 70 throw new IOException("Read end dead"); 71 } 72 73 // 获取“写入管道”的线程 74 writeSide = Thread.currentThread(); 75 // 如果“管道中被读取的数据，等于写入管道的数据”时， 76 // 则每隔1000ms检查“管道状态”，并唤醒管道操作：若有“读取管道数据线程被阻塞”，则唤醒该线程。 77 while (in == out) { 78 if ((readSide != null) && !readSide.isAlive()) { 79 throw new IOException("Pipe broken"); 80 } 81 /* full: kick any waiting readers */ 82 notifyAll(); 83 try { 84 wait(1000); 85 } catch (InterruptedException ex) { 86 throw new java.io.InterruptedIOException(); 87 } 88 } 89 if (in < 0) { 90 in = 0; 91 out = 0; 92 } 93 buffer[in++] = (char) c; 94 if (in >= buffer.length) { 95 in = 0; 96 } 97 } 98 99 // 接收字符数组b。 100 synchronized void receive(char c[], int off, int len) throws IOException { 101 while (--len >= 0) { 102 receive(c[off++]); 103 } 104 } 105 106 // 当PipedWriter被关闭时，被调用 107 synchronized void receivedLast() { 108 closedByWriter = true; 109 notifyAll(); 110 } 111 112 // 从管道(的缓冲)中读取一个字符，并将其转换成int类型 113 public synchronized int read() throws IOException { 114 if (!connected) { 115 throw new IOException("Pipe not connected"); 116 } else if (closedByReader) { 117 throw new IOException("Pipe closed"); 118 } else if (writeSide != null && !writeSide.isAlive() 119 && !closedByWriter && (in < 0)) { 120 throw new IOException("Write end dead"); 121 } 122 123 readSide = Thread.currentThread(); 124 int trials = 2; 125 while (in < 0) { 126 if (closedByWriter) { 127 /* closed by writer, return EOF */ 128 return -1; 129 } 130 if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) { 131 throw new IOException("Pipe broken"); 132 } 133 /* might be a writer waiting */ 134 notifyAll(); 135 try { 136 wait(1000); 137 } catch (InterruptedException ex) { 138 throw new java.io.InterruptedIOException(); 139 } 140 } 141 int ret = buffer[out++]; 142 if (out >= buffer.length) { 143 out = 0; 144 } 145 if (in == out) { 146 /* now empty */ 147 in = -1; 148 } 149 return ret; 150 } 151 152 // 从管道(的缓冲)中读取数据，并将其存入到数组b中 153 public synchronized int read(char cbuf[], int off, int len) throws IOException { 154 if (!connected) { 155 throw new IOException("Pipe not connected"); 156 } else if (closedByReader) { 157 throw new IOException("Pipe closed"); 158 } else if (writeSide != null && !writeSide.isAlive() 159 && !closedByWriter && (in < 0)) { 160 throw new IOException("Write end dead"); 161 } 162 163 if ((off < 0) || (off > cbuf.length) || (len < 0) || 164 ((off + len) > cbuf.length) || ((off + len) < 0)) { 165 throw new IndexOutOfBoundsException(); 166 } else if (len == 0) { 167 return 0; 168 } 169 170 /* possibly wait on the first character */ 171 int c = read(); 172 if (c < 0) { 173 return -1; 174 } 175 cbuf[off] = (char)c; 176 int rlen = 1; 177 while ((in >= 0) && (--len > 0)) { 178 cbuf[off + rlen] = buffer[out++]; 179 rlen++; 180 if (out >= buffer.length) { 181 out = 0; 182 } 183 if (in == out) { 184 /* now empty */ 185 in = -1; 186 } 187 } 188 return rlen; 189 } 190 191 // 是否能从管道中读取下一个数据 192 public synchronized boolean ready() throws IOException { 193 if (!connected) { 194 throw new IOException("Pipe not connected"); 195 } else if (closedByReader) { 196 throw new IOException("Pipe closed"); 197 } else if (writeSide != null && !writeSide.isAlive() 198 && !closedByWriter && (in < 0)) { 199 throw new IOException("Write end dead"); 200 } 201 if (in < 0) { 202 return false; 203 } else { 204 return true; 205 } 206 } 207 208 // 关闭PipedReader 209 public void close() throws IOException { 210 in = -1; 211 closedByReader = true; 212 } 213 }

 1 import java.io.IOException;   
 2    
 3 import java.io.PipedReader;  4  5 @SuppressWarnings("all")  6 /**  7  * 接收者线程  8 */  9 public class Receiver extends Thread { 10 11 // 管道输入流对象。 12 // 它和“管道输出流(PipedWriter)”对象绑定， 13 // 从而可以接收“管道输出流”的数据，再让用户读取。 14 private PipedReader in = new PipedReader(); 15 16 // 获得“管道输入流对象” 17 public PipedReader getReader(){ 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 char[] buf = new char[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 41 // 从“管道输入流”读取>1024个字符时，就停止读取 42 public void readMessageContinued(){ 43 int total=0; 44 while(true) { 45 char[] buf = new char[1024]; 46 try { 47 int len = in.read(buf); 48 total += len; 49 System.out.println(new String(buf,0,len)); 50 // 若读取的字符总数>1024，则退出循环。 51 if (total > 1024) 52 break; 53 } catch (IOException e) { 54  e.printStackTrace(); 55  } 56  } 57 58 try { 59  in.close(); 60 } catch (IOException e) { 61  e.printStackTrace(); 62  } 63  } 64 }

 1 import java.io.IOException;   
 2    
 3 import java.io.PipedWriter;  4 @SuppressWarnings("all")  5 /**  6  * 发送者线程  7 */  8 public class Sender extends Thread {  9 10 // 管道输出流对象。 11 // 它和“管道输入流(PipedReader)”对象绑定， 12 // 从而可以将数据发送给“管道输入流”的数据，然后用户可以从“管道输入流”读取数据。 13 private PipedWriter out = new PipedWriter(); 14 15 // 获得“管道输出流”对象 16 public PipedWriter getWriter(){ 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.toCharArray()); 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); 49  out.close(); 50 } catch (IOException e) { 51  e.printStackTrace(); 52  } 53  } 54 }

 1 import java.io.PipedReader;
 2 import java.io.PipedWriter;  3 import java.io.IOException;  4  5 @SuppressWarnings("all")  6 /**  7  * 管道输入流和管道输出流的交互程序  8 */  9 public class PipeTest { 10 11 public static void main(String[] args) { 12 Sender t1 = new Sender(); 13 14 Receiver t2 = new Receiver(); 15 16 PipedWriter out = t1.getWriter(); 17 18 PipedReader in = t2.getReader(); 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 }