在 Java 多線程開發中，線程間通訊是核心課題之一。除了常見的 wait()/notify()、BlockingQueue 等機制，Java IO 包還提供了一種獨特的通訊方式——管道流。本文將通過完整代碼示例，深入解析 PipedInputStream 與 PipedOutputStream 以及字符管道流 PipedReader/PipedWriter 的工作原理、使用細節與常見坑點。
connect() 方法建立連接，形成「寫入端 → 管道緩衝區 → 讀取端」的通道。PipedOutputStream 寫入數據。PipedInputStream 內部緩衝區（預設 1024 位元組）。PipedInputStream 讀取數據，完成通訊。包含三個類：Sender（發送者）、Receiver（接收者）、PipedStreamTest（測試主類）。
import java.io.IOException;
import java.io.PipedInputStream;
@SuppressWarnings("all")
public class Receiver extends Thread {
private PipedInputStream in = new PipedInputStream();
public PipedInputStream getInputStream() {
return in;
}
@Override
public void run() {
readMessageOnce(); // 或 readMessageContinued()
}
// 單次讀取（最多 1024 位元組）
public void readMessageOnce() {
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));
if (total > 1024) break;
} catch (IOException e) {
e.printStackTrace();
}
}
try { in.close(); } catch (IOException e) { e.printStackTrace(); }
}
}import java.io.IOException;
import java.io.PipedOutputStream;
@SuppressWarnings("all")
public class Sender extends Thread {
private PipedOutputStream out = new PipedOutputStream();
public PipedOutputStream getOutputStream() {
return out;
}
@Override
public void run() {
writeShortMessage(); // 或 writeLongMessage()
}
// 寫入短訊息
private void writeShortMessage() {
String strInfo = "this is a short message";
try {
out.write(strInfo.getBytes());
out.close();
} catch (IOException e) {
e.printStackTrace();
}
}
// 寫入長訊息（1046 位元組）
private void writeLongMessage() {
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 102; i++) sb.append("0123456789"); // 1020 位元組
sb.append("abcdefghijklmnopqrstuvwxyz"); // 26 位元組
String str = sb.toString(); // 總長 1046
try {
out.write(str.getBytes());
out.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
@SuppressWarnings("all")
public class PipedStreamTest {
public static void main(String[] args) {
Sender sender = new Sender();
Receiver receiver = new Receiver();
PipedOutputStream out = sender.getOutputStream();
PipedInputStream in = receiver.getInputStream();
try {
// 管道連接：out.connect(in) 等價於 in.connect(out)
in.connect(out);
sender.start();
receiver.start();
} catch (IOException e) {
e.printStackTrace();
}
}
}PipedInputStream 內部緩衝區預設大小為 1024 位元組（DEFAULT_PIPE_SIZE）。寫入端嘗試寫入 1046 位元組時，前 1024 位元組填滿緩衝區，後續寫入會阻塞，直到讀取端騰出空間。但接收端調用 read(buf) 後直接關閉流，導致剩餘數據無法被讀取。Receiver.run() 改為調用 readMessageContinued()。字符管道流用法與位元組管道流幾乎一致，差別在於處理單位為 char 而非 byte，適合文本數據傳輸。
PipedWriter：字符管道輸出流，繼承 Writer。PipedReader：字符管道輸入流，繼承 Reader。實驗發現：PipedWriter.write() 內部會逐字符調用 PipedReader.receive()，每寫入一個字符就會阻塞等待讀取端消費（源碼中 while (in == out) 邏輯）。這導致：
IOException: Pipe closed。readMessageContinued()），保持管道開放直到所有數據讀完。// PipedInputStream.connect(PipedOutputStream src)
public void connect(PipedOutputStream src) throws IOException {
src.connect(this); // 實際委託給 PipedOutputStream
}
// PipedOutputStream.connect(PipedInputStream snk)
public void connect(PipedInputStream snk) throws IOException {
this.sink = snk;
snk.in = -1; snk.out = 0; // 初始化緩衝區指針
}雙向綁定：out.sink = in 且 in.source = out，形成雙向引用。
// PipedOutputStream.write(byte[] b, int off, int len)
sink.receive(b, off, len); // 委託給輸入流的 receive()
// PipedInputStream.receive(byte[] b, int off, int len)
// 將數據拷貝到內部 buffer，處理環狀緩衝區邏輯
// buffer 預設大小 1024 (DEFAULT_PIPE_SIZE)// PipedInputStream.read(byte[] b, int off, int len)
// 從 buffer 讀取數據，更新 in/out 指針，喚醒等待的寫入線程read()/write() 受限於緩衝區大小（預設 1KB）。close()，接收端讀到 -1 後關閉，否則可能導致線程永久阻塞。PipedWriter 效率較低，大文本傳輸建議用位元組流 + 指定編碼。new PipedInputStream(pipeSize) 可調大緩衝區，減少阻塞切換開銷。管道流是 Java 提供的輕量級線程間通訊機制，適合生產者-消費者模式、流式數據傳遞場景。掌握其緩衝區限制、阻塞協作機制、位元組與字符流差異是正確使用的關鍵。實際項目中，若需高性能或複雜協作，仍建議優先考慮 BlockingQueue、Disruptor 或 Reactive Streams 等更高級組件。
