java io系列20之 PipedReader和PipedWriter

简介: 本章,我们学习PipedReader和PipedWriter。它们和“PipedInputStream和PipedOutputStream”一样,都可以用于管道通信。 PipedWriter 是字符管道输出流,它继承于Writer。

本章,我们学习PipedReader和PipedWriter。它们和“PipedInputStreamPipedOutputStream”一样,都可以用于管道通信。

PipedWriter 是字符管道输出流,它继承于Writer。
PipedReader 是字符管道输入流,它继承于Writer。
PipedWriter和PipedReader的作用是可以通过管道进行线程间的通讯。在使用管道通信时,必须将PipedWriter和PipedReader配套使用。

转载请注明出处:http://www.cnblogs.com/skywang12345/p/io_20.html

更多内容请参考:java io系列01之 "目录"

PipedWriter和PipedReader源码分析

1. PipedWriter 源码(基于jdk1.7.40)

复制代码
 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 }
复制代码

2. PipedReader 源码(基于jdk1.7.40)

复制代码
  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 }
复制代码

 

示例

下面,我们看看多线程中通过PipedWriter和PipedReader通信的例子。例子中包括3个类:Receiver.java, Sender.java 和 PipeTest.java

Receiver.java的代码如下:

复制代码
 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 }
复制代码

Sender.java的代码如下:

复制代码
 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 }
复制代码

PipeTest.java的代码如下:

复制代码
 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 }
复制代码

 

运行结果

this is a short message
结果说明
(01) in.connect(out);

        它的作用是将“管道输入流”和“管道输出流”关联起来。查看PipedWriter.java和PipedReader.java中connect()的源码;我们知道 out.connect(in); 等价于 in.connect(out);
(02)
t1.start(); // 启动“Sender”线程
t2.start(); // 启动“Receiver”线程
先查看Sender.java的源码,线程启动后执行run()函数;在Sender.java的run()中,调用writeShortMessage();
writeShortMessage();的作用就是向“管道输出流”中写入数据"this is a short message" ;这条数据会被“管道输入流”接收到。下面看看这是如何实现的。
先看write(char char的源码。PipedWriter.java继承于Writer.java;Writer.java中write(char c[])的源码如下:

public void write(char cbuf[]) throws IOException {
    write(cbuf, 0, cbuf.length); }

实际上write(char c[])是调用的PipedWriter.java中的write(char c[], int off, int len)函数。查看write(char c[], int off, int len)的源码,我们发现:它会调用 sink.receive(cbuf, off, len); 进一步查看receive(char c[], int off, int len)的定义,我们知道sink.receive(cbuf, off, len)的作用就是:将“管道输出流”中的数据保存到“管道输入流”的缓冲中。而“管道输入流”的缓冲区buffer的默认大小是1024个字符。

至此,我们知道:t1.start()启动Sender线程,而Sender线程会将数据"this is a short message"写入到“管道输出流”;而“管道输出流”又会将该数据传输给“管道输入流”,即而保存在“管道输入流”的缓冲中。


接下来,我们看看“用户如何从‘管道输入流’的缓冲中读取数据”。这实际上就是Receiver线程的动作。
t2.start() 会启动Receiver线程,从而执行Receiver.java的run()函数。查看Receiver.java的源码,我们知道run()调用了readMessageOnce()。
而readMessageOnce()就是调用in.read(buf)从“管道输入流in”中读取数据,并保存到buf中。
通过上面的分析,我们已经知道“管道输入流in”的缓冲中的数据是"this is a short message";因此,buf的数据就是"this is a short message"。


为了加深对管道的理解。我们接着进行下面两个小试验。
试验一:修改Sender.java

public void run(){   
    writeShortMessage();
    //writeLongMessage();
}

修改为

public void run(){   
    //writeShortMessage();
 writeLongMessage(); }

运行程序。运行结果如下:

从中,我们看出,程序运行出错!抛出异常 java.io.IOException: Pipe closed

为什么会这样呢?
我分析一下程序流程。
(01) 在PipeTest中,通过in.connect(out)将输入和输出管道连接起来;然后,启动两个线程。t1.start()启动了线程Sender,t2.start()启动了线程Receiver。
(02) Sender线程启动后,通过writeLongMessage()写入数据到“输出管道”,out.write(str.toCharArray()) 共写入了1046个字符。而根据PipedWriter的源码,PipedWriter的write()函数会调用PipedReader的 receive()函数。而观察PipedReader的receive()函数,我们知道,PipedReader会将接受的数据存储缓冲区。仔细观察 receive()函数,有如下代码:

复制代码
while (in == out) {
    if ((readSide != null) && !readSide.isAlive()) {
        throw new IOException("Pipe broken"); } /* full: kick any waiting readers */ notifyAll(); try { wait(1000); } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } }
复制代码

而in和out的初始值分别是in=-1, out=0;结合上面的while(in==out)。我们知道,它的含义就是,每往管道中写入一个字符,就达到了in==out这个条件。然后,就调用notifyAll(),唤醒“读取管道的线程”。
也就是,每往管道中写入一个字符,都会阻塞式的等待其它线程读取。
然而,PipedReader的缓冲区的默认大小是1024!但是,此时要写入的数据却有1046!所以,一次性最多只能写入1024个字符。
(03) Receiver线程启动后,会调用readMessageOnce()读取管道输入流。读取1024个字符会,会调用close()关闭,管道。

由(02)和(03)的分析可知,Sender要往管道写入1046个字符。其 中,前1024个字符(缓冲区容量是1024)能正常写入,并且每写入一个就读取一个。当写入1025个字符时,依然是依次的调用 PipedWriter.java中的write();然后,write()中调用PipedReader.java中的receive();在 PipedReader.java中,最终又会调用到receive(int c)函数。 而此时,管道输入流已经被关闭,也就是closedByReader为true,所以抛出throw new IOException("Pipe closed")。

我们对“试验一”继续进行修改,解决该问题。


试验二: 在“试验一”的基础上继续修改Receiver.java

public void run(){   
    readMessageOnce() ;
    //readMessageContinued() ;
}

修改为

public void run(){   
    //readMessageOnce() ;
 readMessageContinued() ; }

此时,程序能正常运行。运行结果为:
01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789

012345678901234567890123456789abcd

efghijklmnopqrstuvwxyz

相关文章
|
2月前
|
存储 Java
【IO面试题 四】、介绍一下Java的序列化与反序列化
Java的序列化与反序列化允许对象通过实现Serializable接口转换成字节序列并存储或传输,之后可以通过ObjectInputStream和ObjectOutputStream的方法将这些字节序列恢复成对象。
|
6天前
|
Java 数据处理 开发者
揭秘Java IO流:字节流与字符流的神秘面纱!
揭秘Java IO流:字节流与字符流的神秘面纱!
14 1
|
6天前
|
自然语言处理 Java 数据处理
Java IO流全解析:字节流和字符流的区别与联系!
Java IO流全解析:字节流和字符流的区别与联系!
23 1
|
1月前
|
安全 Java API
【Java面试题汇总】Java基础篇——String+集合+泛型+IO+异常+反射(2023版)
String常量池、String、StringBuffer、Stringbuilder有什么区别、List与Set的区别、ArrayList和LinkedList的区别、HashMap底层原理、ConcurrentHashMap、HashMap和Hashtable的区别、泛型擦除、ABA问题、IO多路复用、BIO、NIO、O、异常处理机制、反射
【Java面试题汇总】Java基础篇——String+集合+泛型+IO+异常+反射(2023版)
|
25天前
|
Java 大数据 API
Java 流(Stream)、文件(File)和IO的区别
Java中的流(Stream)、文件(File)和输入/输出(I/O)是处理数据的关键概念。`File`类用于基本文件操作,如创建、删除和检查文件;流则提供了数据读写的抽象机制,适用于文件、内存和网络等多种数据源;I/O涵盖更广泛的输入输出操作,包括文件I/O、网络通信等,并支持异常处理和缓冲等功能。实际开发中,这三者常结合使用,以实现高效的数据处理。例如,`File`用于管理文件路径,`Stream`用于读写数据,I/O则处理复杂的输入输出需求。
|
5天前
|
存储 Java 程序员
【Java】文件IO
【Java】文件IO
21 0
|
1月前
|
数据采集 Java 数据挖掘
Java IO异常处理:在Web爬虫开发中的实践
Java IO异常处理:在Web爬虫开发中的实践
|
2月前
|
Java 数据处理
Java IO 接口(Input)究竟隐藏着怎样的神秘用法?快来一探究竟,解锁高效编程新境界!
【8月更文挑战第22天】Java的输入输出(IO)操作至关重要,它支持从多种来源读取数据,如文件、网络等。常用输入流包括`FileInputStream`,适用于按字节读取文件;结合`BufferedInputStream`可提升读取效率。此外,通过`Socket`和相关输入流,还能实现网络数据读取。合理选用这些流能有效支持程序的数据处理需求。
31 2
|
2月前
|
XML 存储 JSON
【IO面试题 六】、 除了Java自带的序列化之外,你还了解哪些序列化工具?
除了Java自带的序列化,常见的序列化工具还包括JSON(如jackson、gson、fastjson)、Protobuf、Thrift和Avro,各具特点,适用于不同的应用场景和性能需求。
|
2月前
|
缓存 Java
【IO面试题 一】、介绍一下Java中的IO流
Java中的IO流是对数据输入输出操作的抽象,分为输入流和输出流,字节流和字符流,节点流和处理流,提供了多种类支持不同数据源和操作,如文件流、数组流、管道流、字符串流、缓冲流、转换流、对象流、打印流、推回输入流和数据流等。
【IO面试题 一】、介绍一下Java中的IO流