// 创建时直接连接
PipedOutputStream pos = new PipedOutputStream();
PipedInputStream pis = new PipedInputStream(pos); // 建立连接

// 或者反向连接
PipedInputStream pis2 = new PipedInputStream();
PipedOutputStream pos2 = new PipedOutputStream(pis2);

PipedWriter writer = new PipedWriter();
PipedReader reader = new PipedReader();

// 显式建立连接（必须在读写前调用）
reader.connect(writer); 
// 或 writer.connect(reader);

// 现在可以安全使用
writer.write("Data");
char[] buf = new char[4];
reader.read(buf); // 读取 "Data"

graph LR
    Producer[生产者线程] -->|写入数据| Buffer[循环缓冲区]
    Buffer -->|读取数据| Consumer[消费者线程]

public class PipeConnectionDemo {
    public static void main(String[] args) throws IOException {
        // 创建未连接的流
        PipedWriter writer = new PipedWriter();
        PipedReader reader = new PipedReader();
        
        // 显式建立连接（关键步骤）
        writer.connect(reader); 
        
        Thread producer = new Thread(() -> {
            try {
                writer.write("Hello Pipes!");
                writer.close(); // 关闭流通知结束
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        
        Thread consumer = new Thread(() -> {
            try {
                int charRead;
                while ((charRead = reader.read()) != -1) {
                    System.out.print((char) charRead);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        
        producer.start();
        consumer.start();
    }
}

public class FilterChain {
    public static void main(String[] args) throws IOException {
        PipedWriter source = new PipedWriter();
        PipedReader sink = new PipedReader();
        
        // 创建处理链：A → B → C
        Filter filterA = new Filter(source, new PipedWriter());
        Filter filterB = new Filter(filterA.getOutputReader(), new PipedWriter());
        Filter filterC = new Filter(filterB.getOutputReader(), sink);
        
        // 启动过滤器线程
        filterA.start();
        filterB.start();
        filterC.start();
        
        // 写入原始数据
        source.write("Original data");
        source.close();
        
        // 从sink读取最终结果
        System.out.println(sink.read());
    }
}

class Filter extends Thread {
    private final PipedReader in;
    private final PipedWriter out;
    
    public Filter(PipedReader in, PipedWriter out) {
        this.in = in;
        this.out = out;
    }
    
    @Override
    public void run() {
        try {
            int c;
            while ((c = in.read()) != -1) {
                out.write(transform((char) c)); // 自定义转换逻辑
            }
            out.close();
        } catch (IOException e) { e.printStackTrace(); }
    }
    
    private char transform(char c) { ... } // 过滤逻辑
}

// 主生产者
PipedOutputStream mainOutput = new PipedOutputStream();

// 创建多个消费者管道
PipedInputStream consumer1 = new PipedInputStream(mainOutput);
PipedInputStream consumer2 = new PipedInputStream(mainOutput);

// 生产者写入的数据会被所有消费者读取
mainOutput.write("Broadcast".getBytes());

try (PipedOutputStream pos = new PipedOutputStream();
     PipedInputStream pis = new PipedInputStream(pos)) {
    // 自动关闭双向流
}