Java IO 管道通信：PipedInputStream 與 PipedOutputStream 多線程數據傳輸詳解

管道流核心概念

PipedOutputStream（寫端）與 PipedInputStream（讀端）構成內存管道，實現同進程多線程間的字節流通信。必須成對使用，通過 connect() 綁定。

源碼核心機制（JDK 1.7）

共享緩衝區模型

PipedInputStream 內部維護環形字節數組 buffer[1024]（默認），雙指針：

• in：下一寫入位置（-1 表示空）
• out：下一讀取位置
• in == out 表示滿（寫端阻塞）或空（讀端阻塞），靠 in == -1 區分

寫端關鍵邏輯

// PipedOutputStream.write() -> sink.receive()
synchronized void receive(int b) throws IOException {
    checkState(); writeSide = Thread.currentThread();
    if (in == out) awaitSpace(); // 滿則等待
    if (in < 0) in = out = 0;
    buffer[in++] = (byte)b;
    if (in >= buffer.length) in = 0;
}

讀端關鍵邏輯

synchronized int read() throws IOException {
    readSide = Thread.currentThread();
    while (in < 0) { // 空則等待
        if (closedByWriter) return -1; // EOF
        notifyAll(); wait(1000);
    }
    int ret = buffer[out++] & 0xFF;
    if (out >= buffer.length) out = 0;
    if (in == out) in = -1; // 讀空標記
    return ret;
}

雙端通過 wait/notifyAll 協作，flush() 觸發 notifyAll() 喚醒讀端。

完整生產者-消費者示例

class Sender extends Thread {
    private PipedOutputStream out = new PipedOutputStream();
    public PipedOutputStream getOutputStream() { return out; }
    public void run() {
        try { out.write("this is a short message".getBytes()); out.close(); }
        catch (IOException e) { e.printStackTrace(); }
    }
}

class Receiver extends Thread {
    private PipedInputStream in = new PipedInputStream();
    public PipedInputStream getInputStream() { return in; }
    public void run() {
        try {
            byte[] buf = new byte[2048];
            int len = in.read(buf); // 讀取直到 EOF
            System.out.println(new String(buf, 0, len));
            in.close();
        } catch (IOException e) { e.printStackTrace(); }
    }
}

public class PipeDemo {
    public static void main(String[] args) throws IOException {
        Sender sender = new Sender();
        Receiver receiver = new Receiver();
        // 綁定：兩種方式等價
        sender.getOutputStream().connect(receiver.getInputStream());
        // receiver.getInputStream().connect(sender.getOutputStream());
        sender.start(); receiver.start();
    }
}

緩衝區滿與擴容實驗

1. 發送 1046 字節，默認緩衝 1024 → 僅讀到 1024，寫端阻塞直到讀端消費。
2. 改用 readMessageContinued() 循環讀取 → 完整接收 1046 字節。
3. 可通過 new PipedInputStream(pipeSize) 自定義緩衝區大小。

總結

管道流基於共享環形緩衝區 + 線程等待/通知實現線程間零拷貝傳輸。適用於單進程內生產者-消費者場景。注意：默認 1KB 緩衝較小，高吞吐場景需擴容；跨進程通信請用 Socket/文件/共享內存。