thinking in java (二十) ----- IO之管道（PipedOutputStream和PipedInputStream）
介紹
PipedInputStream和PipedOuputStream管道輸入流和管道輸出流
他的作用是讓多執行緒可以通過管道進行執行緒間的通訊，在使用管道通訊時候，必須將兩者配套使用。使用管道的大致流程是：我們在程序A中向PipedOutputStream中寫入資料，然後這些資料會自動傳送到與PipedOutputStream對應的IPipedInputStream中，進而儲存在PipedInputStream的快取中，此時，執行緒B通過讀取PipedInputStream中的資料，就可以實現執行緒A和執行緒B之間的通訊
示例
先看示例，在看原始碼。例子中有3個類，執行緒Sender，執行緒Receiver，還有測試類test
Sender:
import java.io.IOException; import java.io.PipedOutputStream; @SuppressWarnings("all") /** * 傳送者執行緒 */ public class Sender extends Thread { // 管道輸出流物件。 // 它和“管道輸入流(PipedInputStream)”物件繫結， // 從而可以將資料傳送給“管道輸入流”的資料，然後使用者可以從“管道輸入流”讀取資料。 private PipedOutputStream out = new PipedOutputStream(); // 獲得“管道輸出流”物件 public PipedOutputStream getOutputStream(){ 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.getBytes()); 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.getBytes()); out.close(); } catch (IOException e) { e.printStackTrace(); } } }
Recriver:
import java.io.IOException; import java.io.PipedInputStream; @SuppressWarnings("all") /** * 接收者執行緒 */ public class Receiver extends Thread { // 管道輸入流物件。 // 它和“管道輸出流(PipedOutputStream)”物件繫結， // 從而可以接收“管道輸出流”的資料，再讓使用者讀取。 private PipedInputStream in = new PipedInputStream(); // 獲得“管道輸入流”物件 public PipedInputStream getInputStream(){ return in; } @Override public void run(){ readMessageOnce() ; //readMessageContinued() ; } // 從“管道輸入流”中讀取1次資料 public void readMessageOnce(){ // 雖然buf的大小是2048個位元組，但最多隻會從“管道輸入流”中讀取1024個位元組。 // 因為，“管道輸入流”的緩衝區大小預設只有1024個位元組。 byte[] buf = new byte[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) { byte[] buf = new byte[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(); } } }
PipedStreamTest:
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.io.IOException;
@SuppressWarnings("all")
/**
* 管道輸入流和管道輸出流的互動程式
*/
public class PipedStreamTest {
public static void main(String[] args) {
Sender t1 = new Sender();
Receiver t2 = new Receiver();
PipedOutputStream out = t1.getOutputStream();
PipedInputStream in = t2.getInputStream();
try {
//管道連線。下面2句話的本質是一樣。
//out.connect(in);
in.connect(out);
/**
* Thread類的START方法：
* 使該執行緒開始執行；Java 虛擬機器呼叫該執行緒的 run 方法。
* 結果是兩個執行緒併發地執行；當前執行緒（從呼叫返回給 start 方法）和另一個執行緒（執行其 run 方法）。
* 多次啟動一個執行緒是非法的。特別是當執行緒已經結束執行後，不能再重新啟動。
*/
t1.start();
t2.start();
} catch (IOException e) {
e.printStackTrace();
}
}
}執行結果：this is a short message
說明：
1，in.connect(out)
將“管道輸入流”和“管道輸出流”關聯起來，下面檢視PipedInputStream和PipedOutputStream中connect的原始碼，我們知道out.connec他(in)是一樣的效果。
2. t1.start()；//啟動sender執行緒
t2.start(); /啟動receiver執行緒
執行緒啟動後執行run()方法，在sender中的run（）方法，呼叫了writeShortMessage（）；這個方法的作用就是“向管道輸出流中書寫資料“this is a ....””;這條資料會被管道輸入流接收到，
先看write（byte[]）原始碼：
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);
}我們發現了 sink.receive(b, off, len);這個方法，檢視原始碼可以知道，這個方法的作用是“將管道輸出流”中儲存的 資料儲存到“管道輸入流”中，而“管道輸入流”的緩衝區預設大小是1024個位元組。
至此我們瞭解了t1.start()；//啟動sender執行緒,sender執行緒將資料寫入到管道輸出流，然後管道輸出流將資料傳輸給管道輸入流，從而儲存在快取區中。
接下里就是從緩衝區中讀取資料了，實際上就是receiver的動作
start(); /啟動receiver執行緒。呼叫了readMessageOnce，readMessageOnce呼叫了in.read(buf),從管道輸入流中讀取資料儲存在buf中，因此buf就是“this is a....”
為了加深理解，進行下面兩個小實驗
- 修改Sender
修改為
public void run(){
//writeShortMessage();
writeLongMessage();
} 執行程式結果為：
0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567801234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789abcd
我們可以發現，str的長度是1046個位元組，然後執行結果只有1024個位元組！為什麼會這樣呢？
道理很簡單：管道輸入流的緩衝區預設大小是1024個位元組。所以，最多隻能寫入1024個位元組。
觀察PipedInputStream.java的原始碼，我們能瞭解的更透徹。
private static final int DEFAULT_PIPE_SIZE = 1024;
public PipedInputStream() {
initPipe(DEFAULT_PIPE_SIZE);
}預設建構函式呼叫initPipe(DEFAULT_PIPE_SIZE)，它的原始碼如下：
private void initPipe(int pipeSize) {
if (pipeSize <= 0) {
throw new IllegalArgumentException("Pipe Size <= 0");
}
buffer = new byte[pipeSize];
}可以看出緩衝區的記憶體預設就是1024
- 在實驗一的基礎上修改Receiver
修改為
public void run(){
//readMessageOnce() ;
readMessageContinued() ;
}結果為：
012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789abcd
efghijklmnopqrstuvwxyz
這個結果才是writeLongMessage()寫入到“輸入緩衝區”的完整資料。
receiver中程式碼
if (total > 1024)
break;
第一次讀取了1024，第二次讀取了剩下的元素，total超過1024就跳出。
原始碼分析
PipedOutputStream
package java.io;
import java.io.*;
public class PipedOutputStream extends OutputStream {
// 與PipedOutputStream通訊的PipedInputStream物件
private PipedInputStream sink;
// 建構函式，指定配對的PipedInputStream
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;
// 初始化“管道輸入流”的讀寫位置
// int是PipedInputStream中定義的，代表“管道輸入流”的讀寫位置
snk.in = -1;
// 初始化“管道輸出流”的讀寫位置。
// out是PipedInputStream中定義的，代表“管道輸出流”的讀寫位置
snk.out = 0;
// 設定“管道輸入流”和“管道輸出流”為已連線狀態
// connected是PipedInputStream中定義的，用於表示“管道輸入流與管道輸出流”是否已經連線
snk.connected = true;
}
// 將int型別b寫入“管道輸出流”中。
// 將b寫入“管道輸出流”之後，它會將b傳輸給“管道輸入流”
public void write(int b) throws IOException {
if (sink == null) {
throw new IOException("Pipe not connected");
}
sink.receive(b);
}
// 將位元組陣列b寫入“管道輸出流”中。
// 將陣列b寫入“管道輸出流”之後，它會將其傳輸給“管道輸入流”
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);
}
// 清空“管道輸出流”。
// 這裡會呼叫“管道輸入流”的notifyAll()；
// 目的是讓“管道輸入流”放棄對當前資源的佔有，讓其它的等待執行緒(等待讀取管道輸出流的執行緒)讀取“管道輸出流”的值。
public synchronized void flush() throws IOException {
if (sink != null) {
synchronized (sink) {
sink.notifyAll();
}
}
}
// 關閉“管道輸出流”。
// 關閉之後，會呼叫receivedLast()通知“管道輸入流”它已經關閉。
public void close() throws IOException {
if (sink != null) {
sink.receivedLast();
}
}
}PipedInputStream
package java.io;
public class PipedInputStream extends InputStream {
// “管道輸出流”是否關閉的標記
boolean closedByWriter = false;
// “管道輸入流”是否關閉的標記
volatile boolean closedByReader = false;
// “管道輸入流”與“管道輸出流”是否連線的標記
// 它在PipedOutputStream的connect()連線函式中被設定為true
boolean connected = false;
Thread readSide; // 讀取“管道”資料的執行緒
Thread writeSide; // 向“管道”寫入資料的執行緒
// “管道”的預設大小
private static final int DEFAULT_PIPE_SIZE = 1024;
protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE;
// 緩衝區
protected byte buffer[];
//下一個寫入位元組的位置。in==out代表滿，說明“寫入的資料”全部被讀取了。
protected int in = -1;
//下一個讀取位元組的位置。in==out代表滿，說明“寫入的資料”全部被讀取了。
protected int out = 0;
// 建構函式：指定與“管道輸入流”關聯的“管道輸出流”
public PipedInputStream(PipedOutputStream src) throws IOException {
this(src, DEFAULT_PIPE_SIZE);
}
// 建構函式：指定與“管道輸入流”關聯的“管道輸出流”，以及“緩衝區大小”
public PipedInputStream(PipedOutputStream src, int pipeSize)
throws IOException {
initPipe(pipeSize);
connect(src);
}
// 建構函式：預設緩衝區大小是1024位元組
public PipedInputStream() {
initPipe(DEFAULT_PIPE_SIZE);
}
// 建構函式：指定緩衝區大小是pipeSize
public PipedInputStream(int pipeSize) {
initPipe(pipeSize);
}
// 初始化“管道”：新建緩衝區大小
private void initPipe(int pipeSize) {
if (pipeSize <= 0) {
throw new IllegalArgumentException("Pipe Size <= 0");
}
buffer = new byte[pipeSize];
}
// 將“管道輸入流”和“管道輸出流”繫結。
// 實際上，這裡呼叫的是PipedOutputStream的connect()函式
public void connect(PipedOutputStream src) throws IOException {
src.connect(this);
}
// 接收int型別的資料b。
// 它只會在PipedOutputStream的write(int b)中會被呼叫
protected synchronized void receive(int b) throws IOException {
// 檢查管道狀態
checkStateForReceive();
// 獲取“寫入管道”的執行緒
writeSide = Thread.currentThread();
// 若“寫入管道”的資料正好全部被讀取完，則等待。
if (in == out)
awaitSpace();
if (in < 0) {
in = 0;
out = 0;
}
// 將b儲存到緩衝區
buffer[in++] = (byte)(b & 0xFF);
if (in >= buffer.length) {
in = 0;
}
}
// 接收位元組陣列b。
synchronized void receive(byte b[], int off, int len) throws IOException {
// 檢查管道狀態
checkStateForReceive();
// 獲取“寫入管道”的執行緒
writeSide = Thread.currentThread();
int bytesToTransfer = len;
while (bytesToTransfer > 0) {
// 若“寫入管道”的資料正好全部被讀取完，則等待。
if (in == out)
awaitSpace();
int nextTransferAmount = 0;
// 如果“管道中被讀取的資料，少於寫入管道的資料”；
// 則設定nextTransferAmount=“buffer.length - in”
if (out < in) {
nextTransferAmount = buffer.length - in;
} else if (in < out) { // 如果“管道中被讀取的資料，大於/等於寫入管道的資料”，則執行後面的操作
// 若in==-1(即管道的寫入資料等於被讀取資料)，此時nextTransferAmount = buffer.length - in;
// 否則，nextTransferAmount = out - in;
if (in == -1) {
in = out = 0;
nextTransferAmount = buffer.length - in;
} else {
nextTransferAmount = out - in;
}
}
if (nextTransferAmount > bytesToTransfer)
nextTransferAmount = bytesToTransfer;
// assert斷言的作用是，若nextTransferAmount <= 0，則終止程式。
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");
}
}
// 等待。
// 若“寫入管道”的資料正好全部被讀取完(例如，管道緩衝滿)，則執行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();
}
}
}
// 當PipedOutputStream被關閉時，被呼叫
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++] & 0xFF;
if (out >= buffer.length) {
out = 0;
}
if (in == out) {
/* now empty */
in = -1;
}
return ret;
}
// 從管道(的緩衝)中讀取資料，並將其存入到陣列b中
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) {
/* now empty */
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;
}
}
}