您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

Java 線程通信之管道流：PipedInputStream 與 PipedOutputStream 實戰詳解

編輯：Java 綜合教學

前言

管道流（PipedInputStream / PipedOutputStream）是 Java 提供的線程間內存數據傳輸機制，實現「生產者-消費者」模式的輕量級方案。本文通過完整代碼示例，講解連接方式、線程協作流程、關鍵注意點及適用場景。

一、核心概念

管道輸出流 (PipedOutputStream)：生產者寫入端，類比水管「入口」；
管道輸入流 (PipedInputStream)：消費者讀取端，類比水管「出口」；
連接：兩端必須配對連接，否則讀寫會拋出 IOException；
線程隔離：必須在不同線程中操作兩端，否則單線程讀寫會導致死鎖（缓冲區滿無人讀取）。

二、完整代碼示例


import java.io.*;

public class PipedStreamDemo {
    public static void main(String[] args) throws IOException {
        // 1. 創建管道兩端
        PipedOutputStream out = new PipedOutputStream();
        PipedInputStream in = new PipedInputStream(out); // 構造器直接連接
        // 等價於：in = new PipedInputStream(); in.connect(out);

        // 2. 啟動讀線程（消費者）
        Thread reader = new Thread(new ReaderTask(in), "Reader-Thread");
        // 3. 啟動寫線程（生產者）
        Thread writer = new Thread(new WriterTask(out), "Writer-Thread");

        reader.start();
        writer.start();
    }
}

class WriterTask implements Runnable {
    private final PipedOutputStream out;
    WriterTask(PipedOutputStream out) { this.out = out; }

    public void run() {
        try {
            System.out.println("[Writer] 3秒後寫入數據...");
            Thread.sleep(3000);
            String data = "wangzhihong";
            out.write(data.getBytes());
            out.flush(); // 強制刷新緩衝區
            System.out.println("[Writer] 寫入完成: " + data);
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            closeQuietly(out);
        }
    }
}

class ReaderTask implements Runnable {
    private final PipedInputStream in;
    ReaderTask(PipedInputStream in) { this.in = in; }

    public void run() {
        try {
            byte[] buf = new byte[1024];
            int len = in.read(buf); // 阻塞直到有數據或流關閉
            if (len > 0) {
                System.out.println("[Reader] 讀取到: " + new String(buf, 0, len));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            closeQuietly(in);
        }
    }
}

// 通用靜默關閉工具
static void closeQuietly(Closeable c) {
    if (c != null) try { c.close(); } catch (IOException ignored) {}
}

三、運行流程時序圖

主線程
  │
  ├─► new PipedOutputStream()
  ├─► new PipedInputStream(out)  // 內部建立共享環形緩衝區 (默認 1024 字節)
  ├─► reader.start() → ReaderTask.run() → in.read() 【阻塞等待】
  ├─► writer.start() → WriterTask.run() → sleep(3s) → out.write() → out.flush()
  │                                                   │
  │                                                   ▼
  │                                            數據寫入共享緩衝區
  │                                                   │
  │                                                   ▼
  └─► (reader 解除阻塞) in.read() 返回長度 → 解析輸出

四、關鍵注意點與最佳實踐

問題	原因	解決方案
單線程讀寫死鎖	緩衝區滿，寫端阻塞；讀端未啟動，無人消費	嚴格使用兩個獨立線程，先啟動讀線程
流未關閉導致資源洩漏	異常路徑未調用 close()	使用 `try-with-resources` (Java 7+) 或 `finally` 保證關閉
數據截斷/亂碼	read() 返回實際讀取長度，需用 `new String(buf, 0, len)`	嚴格使用返回長度截取字節數組
緩衝區大小固定 (1024)	大數據傳輸可能頻繁阻塞	大數據場景改用 `ByteArrayOutputStream` + 隊列、或 Netty 等網絡框架
斷開連接異常	一端關閉，另端讀寫拋 IOException	協議設計明確「結束標記」（如特殊字節、長度前綴），優雅關閉

五、進階：字符管道流

若傳輸文本，推薦使用字符版：


PipedWriter writer = new PipedWriter();
PipedReader reader = new PipedReader(writer);
// 支持 write(String)、read(char[])，自動處理編碼

六、適用場景與替代方案對比

場景	推薦方案	理由
單進程線程間少量數據傳遞	PipedInputStream/OutputStream	輕量、標準庫、無依賴
高吞吐、背壓控制、流式處理	Reactor/RxJava Flow、Disruptor	非阻塞、背壓、豐富操作符
進程間通信 (IPC)	Socket、gRPC、共享內存	跨進程/機器
線程間共享狀態、任務提交	BlockingQueue、CompletableFuture	語義更清晰、支持超時、取消

總結

管道流是 Java 線程通信的「經典教具」，通過共享環形緩衝區實現生產者-消費者解耦。掌握其雙線程強制要求、連接順序、緩衝區阻塞語義，能避開 90% 的坑。實際工程中，根據數據量、延遲要求、擴展性選擇更高級的併發組件（Queue、Reactive Streams、Disruptor）。