thinking in java (二十八) ----- IO之PipedReader和PipedWriter
阿新 • • 發佈：2018-11-16
PipedReader和PipedWriter介紹
PipedWriter是字元管道輸出流，繼承於Writer
PipedReader是字元管道輸入流，繼承於Reader
PipedReader和PipedWriter的作用是可以通過管道之間進行執行緒間的通訊，在使用管道的時候，必須將兩者配套使用
PipedReader和PipedWriter原始碼分析
PipedReader原始碼
package java.io; public class PipedWriter extends Writer { // 與PipedWriter通訊的PipedReader物件 private PipedReader sink; // PipedWriter的關閉標記 private boolean closed = false; // 建構函式，指定配對的PipedReader public PipedWriter(PipedReader snk) throws IOException { connect(snk); } // 建構函式 public PipedWriter() { } // 將“PipedWriter” 和 “PipedReader”連線。 public synchronized void connect(PipedReader snk) throws IOException { if (snk == null) { throw new NullPointerException(); } else if (sink != null || snk.connected) { throw new IOException("Already connected"); } else if (snk.closedByReader || closed) { throw new IOException("Pipe closed"); } sink = snk; snk.in = -1; snk.out = 0; // 設定“PipedReader”和“PipedWriter”為已連線狀態 // connected是PipedReader中定義的，用於表示“PipedReader和PipedWriter”是否已經連線 snk.connected = true; } // 將一個字元c寫入“PipedWriter”中。 // 將c寫入“PipedWriter”之後，它會將c傳輸給“PipedReader” public void write(int c) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } sink.receive(c); } // 將字元陣列b寫入“PipedWriter”中。 // 將陣列b寫入“PipedWriter”之後，它會將其傳輸給“PipedReader” public void write(char cbuf[], int off, int len) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } else if ((off | len | (off + len) | (cbuf.length - (off + len))) < 0) { throw new IndexOutOfBoundsException(); } sink.receive(cbuf, off, len); } // 清空“PipedWriter”。 // 這裡會呼叫“PipedReader”的notifyAll()； // 目的是讓“PipedReader”放棄對當前資源的佔有，讓其它的等待執行緒(等待讀取PipedWriter的執行緒)讀取“PipedWriter”的值。 public synchronized void flush() throws IOException { if (sink != null) { if (sink.closedByReader || closed) { throw new IOException("Pipe closed"); } synchronized (sink) { sink.notifyAll(); } } } // 關閉“PipedWriter”。 // 關閉之後，會呼叫receivedLast()通知“PipedReader”它已經關閉。 public void close() throws IOException { closed = true; if (sink != null) { sink.receivedLast(); } } }
PipedReader原始碼
package java.io; public class PipedReader extends Reader { // “PipedWriter”是否關閉的標記 boolean closedByWriter = false; // “PipedReader”是否關閉的標記 boolean closedByReader = false; // “PipedReader”與“PipedWriter”是否連線的標記 // 它在PipedWriter的connect()連線函式中被設定為true boolean connected = false; Thread readSide; // 讀取“管道”資料的執行緒 Thread writeSide; // 向“管道”寫入資料的執行緒 // “管道”的預設大小 private static final int DEFAULT_PIPE_SIZE = 1024; // 緩衝區 char buffer[]; //下一個寫入字元的位置。in==out代表滿，說明“寫入的資料”全部被讀取了。 int in = -1; //下一個讀取字元的位置。in==out代表滿，說明“寫入的資料”全部被讀取了。 int out = 0; // 建構函式：指定與“PipedReader”關聯的“PipedWriter” public PipedReader(PipedWriter src) throws IOException { this(src, DEFAULT_PIPE_SIZE); } // 建構函式：指定與“PipedReader”關聯的“PipedWriter”，以及“緩衝區大小” public PipedReader(PipedWriter src, int pipeSize) throws IOException { initPipe(pipeSize); connect(src); } // 建構函式：預設緩衝區大小是1024字元 public PipedReader() { initPipe(DEFAULT_PIPE_SIZE); } // 建構函式：指定緩衝區大小是pipeSize public PipedReader(int pipeSize) { initPipe(pipeSize); } // 初始化“管道”：新建緩衝區大小 private void initPipe(int pipeSize) { if (pipeSize <= 0) { throw new IllegalArgumentException("Pipe size <= 0"); } buffer = new char[pipeSize]; } // 將“PipedReader”和“PipedWriter”繫結。 // 實際上，這裡呼叫的是PipedWriter的connect()函式 public void connect(PipedWriter src) throws IOException { src.connect(this); } // 接收int型別的資料b。 // 它只會在PipedWriter的write(int b)中會被呼叫 synchronized void receive(int c) 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"); } // 獲取“寫入管道”的執行緒 writeSide = Thread.currentThread(); // 如果“管道中被讀取的資料，等於寫入管道的資料”時， // 則每隔1000ms檢查“管道狀態”，並喚醒管道操作：若有“讀取管道資料執行緒被阻塞”，則喚醒該執行緒。 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(); } } if (in < 0) { in = 0; out = 0; } buffer[in++] = (char) c; if (in >= buffer.length) { in = 0; } } // 接收字元陣列b。 synchronized void receive(char c[], int off, int len) throws IOException { while (--len >= 0) { receive(c[off++]); } } // 當PipedWriter被關閉時，被呼叫 synchronized void receivedLast() { closedByWriter = true; notifyAll(); } // 從管道(的緩衝)中讀取一個字元，並將其轉換成int型別 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++]; if (out >= buffer.length) { out = 0; } if (in == out) { /* now empty */ in = -1; } return ret; } // 從管道(的緩衝)中讀取資料，並將其存入到陣列b中 public synchronized int read(char cbuf[], int off, int len) 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"); } if ((off < 0) || (off > cbuf.length) || (len < 0) || ((off + len) > cbuf.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return 0; } /* possibly wait on the first character */ int c = read(); if (c < 0) { return -1; } cbuf[off] = (char)c; int rlen = 1; while ((in >= 0) && (--len > 0)) { cbuf[off + rlen] = buffer[out++]; rlen++; if (out >= buffer.length) { out = 0; } if (in == out) { /* now empty */ in = -1; } } return rlen; } // 是否能從管道中讀取下一個資料 public synchronized boolean ready() 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"); } if (in < 0) { return false; } else { return true; } } // 關閉PipedReader public void close() throws IOException { in = -1; closedByReader = true; } }
示例
下面，我們看看多執行緒中通過PipedWriter和PipedReader通訊的例子。例子中包括3個類：Receiver.java, Sender.java 和 PipeTest.java
Receiver.java的程式碼如下：
import java.io.IOException; import java.io.PipedReader; @SuppressWarnings("all") /** * 接收者執行緒 */ public class Receiver extends Thread { // 管道輸入流物件。 // 它和“管道輸出流(PipedWriter)”物件繫結， // 從而可以接收“管道輸出流”的資料，再讓使用者讀取。 private PipedReader in = new PipedReader(); // 獲得“管道輸入流物件” public PipedReader getReader(){ return in; } @Override public void run(){ readMessageOnce() ; //readMessageContinued() ; } // 從“管道輸入流”中讀取1次資料 public void readMessageOnce(){ // 雖然buf的大小是2048個字元，但最多隻會從“管道輸入流”中讀取1024個字元。 // 因為，“管道輸入流”的緩衝區大小預設只有1024個字元。 char[] buf = new char[2048]; try { int len = in.read(buf); System.out.println(new String(buf,0,len)); in.close(); } catch (IOException e) { e.printStackTrace(); } } // 從“管道輸入流”讀取>1024個字元時，就停止讀取 public void readMessageContinued(){ int total=0; while(true) { char[] buf = new char[1024]; try { int len = in.read(buf); total += len; System.out.println(new String(buf,0,len)); // 若讀取的字元總數>1024，則退出迴圈。 if (total > 1024) break; } catch (IOException e) { e.printStackTrace(); } } try { in.close(); } catch (IOException e) { e.printStackTrace(); } } }
Sender.java的程式碼如下：
import java.io.IOException;
import java.io.PipedWriter;
@SuppressWarnings("all")
/**
* 傳送者執行緒
*/
public class Sender extends Thread {
// 管道輸出流物件。
// 它和“管道輸入流(PipedReader)”物件繫結，
// 從而可以將資料傳送給“管道輸入流”的資料，然後使用者可以從“管道輸入流”讀取資料。
private PipedWriter out = new PipedWriter();
// 獲得“管道輸出流”物件
public PipedWriter getWriter(){
return out;
}
@Override
public void run(){
writeShortMessage();
//writeLongMessage();
}
// 向“管道輸出流”中寫入一則較簡短的訊息："this is a short message"
private void writeShortMessage() {
String strInfo = "this is a short message" ;
try {
out.write(strInfo.toCharArray());
out.close();
} catch (IOException e) {
e.printStackTrace();
}
}
// 向“管道輸出流”中寫入一則較長的訊息
private void writeLongMessage() {
StringBuilder sb = new StringBuilder();
// 通過for迴圈寫入1020個字元
for (int i=0; i<102; i++)
sb.append("0123456789");
// 再寫入26個字元。
sb.append("abcdefghijklmnopqrstuvwxyz");
// str的總長度是1020+26=1046個字元
String str = sb.toString();
try {
// 將1046個字元寫入到“管道輸出流”中
out.write(str);
out.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}PipeTest.java的程式碼如下：
import java.io.PipedReader;
import java.io.PipedWriter;
import java.io.IOException;
@SuppressWarnings("all")
/**
* 管道輸入流和管道輸出流的互動程式
*/
public class PipeTest {
public static void main(String[] args) {
Sender t1 = new Sender();
Receiver t2 = new Receiver();
PipedWriter out = t1.getWriter();
PipedReader in = t2.getReader();
try {
//管道連線。下面2句話的本質是一樣。
//out.connect(in);
in.connect(out);
/**
* Thread類的START方法：
* 使該執行緒開始執行；Java 虛擬機器呼叫該執行緒的 run 方法。
* 結果是兩個執行緒併發地執行；當前執行緒（從呼叫返回給 start 方法）和另一個執行緒（執行其 run 方法）。
* 多次啟動一個執行緒是非法的。特別是當執行緒已經結束執行後，不能再重新啟動。
*/
t1.start();
t2.start();
} catch (IOException e) {
e.printStackTrace();
}
}
}結果說明：
(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 [] ）是呼叫的PipedWriter中的write(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();
}執行程式。執行結果如下：
樣呢？
我分析一下程式流程。
(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() ;
}執行結果正常。