java并发：管道流（Piped Streams）的应用场景

管道流（Piped Streams）是Java中用于线程间通信的特殊I/O流，它们通过创建"生产-消费"通道实现数据传输。

核心组件

 

连接方式

方式1：通过构造函数连接（推荐）

// 创建时直接连接
PipedOutputStream pos = new PipedOutputStream();
PipedInputStream pis = new PipedInputStream(pos); // 建立连接// 或者反向连接
PipedInputStream pis2 = new PipedInputStream();
PipedOutputStream pos2 = new PipedOutputStream(pis2);

 

方式2：通过 connect() 方法连接

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"

 

连接机制底层原理

管道流的连接在JVM层创建了一个循环缓冲区

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

 

典型应用场景

（1）线程间数据流传递

当两个线程需要顺序处理数据流时，管道流可替代共享变量：

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();}
}

特点：

避免显式同步，通过流阻塞机制自然协调生产消费节奏

 

（2）自定义过滤器链

public class FilterChain {public static void main(String[] args) throws IOException {PipedWriter source = new PipedWriter();PipedReader sink = new PipedReader();// 创建处理链：A → B → CFilter 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;}@Overridepublic 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());

备注：实际应用中需考虑线程安全，建议配合BroadcastStream自定义实现

注意事项

必须显式连接：通过构造函数或connect()方法建立管道

缓冲区大小：默认1024字节，可通过PipedInputStream(int)构造函数调整

资源管理：使用try-with-resources确保关闭

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

 

总结

Java管道流的显式连接机制（构造函数或connect()）是线程间流式通信的基础。

在多线程场景中，务必在启动线程前完成连接，并通过资源管理确保可靠关闭。