Java 管道流實現線程間通訊

前言

Java I/O 體系中，管道流（PipedInputStream/PipedOutputStream、PipedReader/PipedWriter）提供了一種線程間直接傳遞資料的機制，無需臨時檔案或共享記憶體。本文透過一個完整範例 pipeapp，示範如何利用管道串聯多個工作線程，完成資料的逐級轉換。

管道流基本概念

• PipedOutputStream / PipedWriter：資料寫入端（生產者）。
• PipedInputStream / PipedReader：資料讀取端（消費者）。
• 建立管道時，必須將輸出端傳給輸入端建構子，或反之，完成「連接」。

// 位元組管道
PipedOutputStream out = new PipedOutputStream();
PipedInputStream in = new PipedInputStream(out);

// 字元管道
PipedWriter writer = new PipedWriter();
PipedReader reader = new PipedReader(writer);

範例：三階段資料轉換管線

流程：input.txt（每行 "x"） → 執行緒 Y 轉 "y" → 執行緒 Z 轉 "z" → 主線程列印。

主類 PipeApp

import java.io.*;

public class PipeApp {
    public static void main(String[] args) {
        PipeApp app = new PipeApp();
        try (FileInputStream fis = new FileInputStream("input.txt")) {
            InputStream yPipe = app.changeToY(fis);
            InputStream zPipe = app.changeToZ(yPipe);

            try (DataInputStream dis = new DataInputStream(zPipe)) {
                String line;
                System.out.println("=== 轉換結果 ===");
                while ((line = dis.readLine()) != null) {
                    System.out.println(line);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 階段 1：x → y
    public InputStream changeToY(InputStream in) throws IOException {
        DataInputStream dis = new DataInputStream(in);
        PipedOutputStream pos = new PipedOutputStream();
        PipedInputStream pis = new PipedInputStream(pos);
        PrintStream ps = new PrintStream(pos);

        new YThread(dis, ps).start();
        return pis; // 回傳讀取端給下一階段
    }

    // 階段 2：y → z
    public InputStream changeToZ(InputStream in) throws IOException {
        DataInputStream dis = new DataInputStream(in);
        PipedOutputStream pos = new PipedOutputStream();
        PipedInputStream pis = new PipedInputStream(pos);
        PrintStream ps = new PrintStream(pos);

        new ZThread(dis, ps).start();
        return pis;
    }
}

工作線程 YThread / ZThread

import java.io.*;

class YThread extends Thread {
    private final DataInputStream in;
    private final PrintStream out;

    public YThread(DataInputStream in, PrintStream out) {
        this.in = in;
        this.out = out;
    }

    @Override
    public void run() {
        try {
            String line;
            while ((line = in.readLine()) != null) {
                out.println(line.replace('x', 'y'));
                out.flush(); // 確保資料及時進入管道
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try { out.close(); } catch (Exception ignored) {}
        }
    }
}

// ZThread 結構相同，僅將 replace('x','y') 改為 replace('y','z')

關鍵注意事項

1. 緩衝區大小：管道預設緩衝區較小（1KB），大量資料時生產者可能阻塞，需考慮 PipedOutputStream 的 write 行為。
2. 關閉順序：生產者必須在寫入完成後呼叫 close()，消費者才能讀到 -1/null 結束標記。
3. 執行緒安全：管道流內部已同步，可直接多線程讀寫。
4. 死鎖風險：若在同一線程中先讀後寫同一管道，會導致死鎖，務必分離生產/消費線程。

應用場景

• 編譯器/解析器的前端管線（詞法分析 → 語法分析 → 語義分析）。
• 日誌處理管線：讀取 → 過濾 → 格式化 → 輸出。
• 網路協議解碼：接收 → 解碼 → 業務處理 → 回應編碼 → 發送。

總結

管道流是 Java 線程間通訊的輕量級方案，適合「串流式、單向、生產-消費」模型。配合 DataInputStream/PrintStream 等裝飾器，能快速構建靈活的資料處理管線。