Java IO 體系中，管道流 提供了一種線程間直接傳輸數據的機制：一個線程寫入 PipedOutputStream，另一個線程從 PipedInputStream 讀取，無需臨時文件或共享內存緩衝區，實現生產者-消費者模式的流式數據交換。本文詳細講解管道創建、連接、讀寫協作、完整實戰案例及注意事項。
PipedOutputStream：管道寫入端（生產者）。PipedInputStream：管道讀取端（消費者）。new PipedInputStream(pipedOutputStream)pipedInputStream.connect(pipedOutputStream)一旦連接，雙方形成綁定關係，數據從 OutputStream 流向 InputStream，內部維護一個默認 1024 字節的循環緩衝區（JDK 实现细节，不可配置）。
// 生產者線程
new Thread(() -> {
try (PrintWriter out = new PrintWriter(pipedOutputStream)) {
for (int i = 0; i < 100; i++) {
out.println("Data-" + i);
out.flush(); // 關鍵：確保數據及時推送到管道緩衝區
}
} catch (IOException e) { /* 處理 */ }
}, "Producer").start();
// 消費者線程（或主線程）
try (BufferedReader in = new BufferedReader(new InputStreamReader(pipedInputStream))) {
String line;
while ((line = in.readLine()) != null) {
System.out.println("消費: " + line);
}
} catch (IOException e) { /* 處理 */ }
read() 會永久阻塞等待 EOF。write() 阻塞，直到消費者讀取騰出空間（背壓機制）。IOException: Pipe broken。模擬：主線程讀文件 → 線程 A 轉換 'x'→'y' → 線程 B 轉換 'y'→'z' → 主線程輸出結果。
public class PipeChainDemo {
public static void main(String[] args) throws IOException {
// 1. 源文件輸入流
FileInputStream fileIn = new FileInputStream("input.txt");
DataInputStream dataIn = new DataInputStream(fileIn);
// 2. 第一級管道：x -> y
PipedOutputStream out1 = new PipedOutputStream();
PipedInputStream in1 = new PipedInputStream(out1);
new Thread(new Transformer(dataIn, new PrintStream(out1), 'x', 'y'), "X-to-Y").start();
// 3. 第二級管道：y -> z
PipedOutputStream out2 = new PipedOutputStream();
PipedInputStream in2 = new PipedInputStream(out2);
new Thread(new Transformer(new DataInputStream(in1), new PrintStream(out2), 'y', 'z'), "Y-to-Z").start();
// 4. 主線程消費最終結果
try (BufferedReader finalIn = new BufferedReader(new InputStreamReader(in2))) {
String line;
while ((line = finalIn.readLine()) != null) {
System.out.println(line);
}
}
}
static class Transformer implements Runnable {
private final DataInputStream in;
private final PrintStream out;
private final char from, to;
Transformer(DataInputStream in, PrintStream out, char from, char to) {
this.in = in; this.out = out; this.from = from; this.to = to;
}
@Override public void run() {
try {
String line;
while ((line = in.readLine()) != null) {
out.println(line.replace(from, to));
out.flush();
}
} catch (IOException ignored) {
} finally {
out.close(); // 關鍵：關閉輸出端，傳遞 EOF 給下游
}
}
}
}
| 局限 | 說明 | 替代/改進 |
|---|---|---|
| 緩衝區固定小 (1KB) | 高吞吐場景易阻塞，不可配置 | 使用 BlockingQueue、Disruptor、共享內存環形緩衝區 |
| 單向、點對點 | 一對一，不支持多生產/多消費、廣播 | JUC BlockingQueue、消息隊列 (Kafka/RocketMQ) |
| 無超時機制 | read/write 只有阻塞/異常，無超時控制 | NIO Channel + Selector、Netty、JUC Lock 超時 |
| 序列化對象傳輸麻煩 | 需手動 ObjectOutputStream/InputStream 包裝 | 直接傳遞對象引用（共享內存）或成熟 RPC 框架 |
| 異常處理複雜 | Pipe broken、流關閉順序敏感 | 結構化併發（Structured Concurrency）、Try-with-resources 規範管理 |
finally 中關閉輸出流。IOException: Pipe broken 正常退出。PrintWriter/BufferedReader 等包裝流提升效率與易用性。flush() 確保及時推送。管道流是 Java 早期為線程間流式通信設計的輕量級機制，理解其原理有助於掌握流式處理模型，但現代高併發系統中多被更高性能、功能豐富的組件取代。
