Java IO 管道通訊：PipedReader 與 PipedWriter 線程間字元傳輸詳解

核心定位：線程間字元管道

PipedReader 與 PipedWriter 是字元導向的管道通訊類，分別繼承自 Reader 與 Writer。它們成對使用，實現生產者-消費者模式的線程間字元資料傳輸：寫入執行緒向 PipedWriter 寫入字元，讀取執行緒從 PipedReader 讀取。內部維護一個循環字元緩衝區（預設 1024 字元），透過 wait/notifyAll 實現同步阻塞機制。

關鍵建構與連接方式

// 方式 1：建構時指定配對物件（自動連接）
PipedWriter writer = new PipedWriter();
PipedReader reader = new PipedReader(writer);

// 方式 2：先建構再連接（順序不分先後）
PipedWriter writer = new PipedWriter();
PipedReader reader = new PipedReader();
writer.connect(reader);  // 或 reader.connect(writer)

// 方式 3：指定緩衝區大小
PipedReader reader = new PipedReader(writer, 4096);  // 4KB 緩衝區

核心同步機制深度剖析

循環緩衝區與指標設計

// PipedReader 內部核心欄位
char[] buffer;        // 循環緩衝區
int in = -1;          // 下一個寫入位置
int out = 0;          // 下一個讀取位置
// in == out 表示：緩衝區空（初始）或滿（寫滿後）

// 同步狀態旗標
boolean closedByWriter = false;
boolean closedByReader = false;
boolean connected = false;
Thread readSide;      // 讀取執行緒
Thread writeSide;     // 寫入執行緒

寫入流程：receive() 阻塞等待空間

// PipedReader.receive(int c) - 由 PipedWriter.write() 呼叫
synchronized void receive(int c) throws IOException {
    // 狀態檢查
    if (!connected) throw new IOException("Pipe not connected");
    if (closedByWriter || closedByReader) throw new IOException("Pipe closed");
    if (readSide != null && !readSide.isAlive()) throw new IOException("Read end dead");
    
    writeSide = Thread.currentThread();
    
    // 緩衝區滿：阻塞等待讀取端消費
    while (in == out) {
        if (readSide != null && !readSide.isAlive()) 
            throw new IOException("Pipe broken");
        notifyAll();           // 喚醒可能等待的讀取執行緒
        wait(1000);            // 釋放鎖，最多等待 1 秒
    }
    
    if (in < 0) { in = 0; out = 0; }  // 初始化
    buffer[in++] = (char) c;
    if (in >= buffer.length) in = 0;  // 循環
}

讀取流程：read() 阻塞等待資料

// PipedReader.read() 核心邏輯
public int read() throws IOException {
    synchronized (this) {
        if (!connected) throw new IOException("Pipe not connected");
        if (closedByReader) throw new IOException("Pipe closed");
        
        readSide = Thread.currentThread();
        
        // 緩衝區空：阻塞等待寫入端產出
        while (in == out) {
            if (closedByWriter) return -1;  // 寫入端正常關閉，EOF
            if (writeSide != null && !writeSide.isAlive()) 
                throw new IOException("Pipe broken");
            notifyAll();
            wait(1000);
        }
        
        int c = buffer[out++];
        if (out >= buffer.length) out = 0;
        return c;
    }
}

完整生產者-消費者範例

import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;

// 生產者：寫入執行緒
class DataProducer extends Thread {
    private final PipedWriter writer;
    
    public DataProducer(PipedWriter writer) {
        this.writer = writer;
    }
    
    @Override
    public void run() {
        try {
            String[] messages = {
                "訂單編號: 2024001",
                "客戶: 張三",
                "商品: 筆記型電腦 x1",
                "金額: NT$ 35,000",
                "狀態: 已出貨"
            };
            
            for (String msg : messages) {
                writer.write(msg);
                writer.write('\n');
                writer.flush();  // 確保資料推送到管道
                Thread.sleep(200);  // 模擬處理間隔
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            try { writer.close(); } catch (IOException ignored) {}
        }
    }
}

// 消費者：讀取執行緒
class DataConsumer extends Thread {
    private final PipedReader reader;
    
    public DataConsumer(PipedReader reader) {
        this.reader = reader;
    }
    
    @Override
    public void run() {
        try (BufferedReader br = new BufferedReader(reader)) {
            String line;
            while ((line = br.readLine()) != null) {
                System.out.println("[消費者] 收到: " + line);
                // 這裡可進行業務處理：解析、存儲、推送等
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

// 主程式啟動管道通訊
public class PipedCommunicationDemo {
    public static void main(String[] args) throws IOException {
        PipedWriter writer = new PipedWriter();
        PipedReader reader = new PipedReader(writer);  // 自動連接
        
        new DataProducer(writer).start();
        new DataConsumer(reader).start();
        
        // 主執行緒可繼續做其他事，或等待子執行緒結束
    }
}

緩衝區大小與死鎖陷阱

預設 1024 字元限制

若生產者單次寫入超過緩衝區容量（如 1046 字元），會導致寫入執行緒在 wait() 中阻塞。若消費者讀取後直接 close()，生產者喚醒時會收到 "Pipe closed" 異常。

解決策略

1. 增大緩衝區：new PipedReader(writer, 8192) 或更大。
2. 消費者持續讀取：不要一次性讀完就關閉，改用循環讀取直到 EOF（read() == -1）。
3. 生產者分批寫入：大資料切片寫入，配合 flush()。

PipedReader/Writer vs PipedInputStream/OutputStream

現代替代方案：BlockingQueue 與 Reactive Streams

// 更靈活、功能更豐富的線程間傳輸
BlockingQueue queue = new LinkedBlockingQueue<>(100);

// 生產者
queue.put("訊息內容");  // 滿時阻塞

// 消費者
String msg = queue.take();  // 空時阻塞

// 或使用 Reactor / RxJava 進行背壓控制、操作符組合
Flux flux = Flux.create(sink -> {
    // 生產者邏輯
    sink.next("訊息");
}).subscribeOn(Schedulers.boundedElastic());

flux.doOnNext(msg -> {
    // 消費者邏輯
}).subscribe();

最佳實踐總結

1. 明確生命週期：生產者結束務必 writer.close()，消費者讀到 -1 才算正常結束。
2. 避免單向關閉：一方異常結束會導致另一方收到 IOException，需妥善捕獲處理。
3. 緩衝區評估：根據訊息大小與產消速度差預估緩衝區，必要時擴大。
4. 考慮替代方案：新專案優先評估 BlockingQueue、Disruptor、Reactive Streams 等更現代的併發傳輸組件。

PipedReader/PipedWriter 提供了線程間字元流傳輸的基礎機制，理解其循環緩衝區、阻塞同步與關閉協議，有助於正確使用或遷移至更現代的併發組件。