Java中实现流管道传输：Piped流线程使用问题咨询

嘿，你找对工具了！JDK提供的PipedInputStream和PipedOutputStream正是为了实现这种实时将OutputStream的输出直接导向InputStream的场景，不过它们的线程限制确实是关键，稍不注意就会踩坑。

为什么必须分线程使用？ #

官方文档的警告不是空穴来风：如果在同一个线程里同时进行写（PipedOutputStream）和读（PipedInputStream）操作，极大概率会触发死锁。原因很简单：

• 当管道缓冲区被写满时，写操作会阻塞，等待读操作取走数据；
• 当管道缓冲区为空时，读操作会阻塞，等待写操作写入数据。
  如果这两个操作在同一个线程里，阻塞发生后就没有其他线程能打破这个僵局了。

完整的线程安全示例 #

下面是一个简单的实现，用两个独立线程分别处理写和读：

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

public class PipeExample {
    public static void main(String[] args) throws IOException {
        // 构建管道流对
        PipedOutputStream out = new PipedOutputStream();
        PipedInputStream in = new PipedInputStream(out);

        // 写线程：负责向管道写入数据
        Thread writerThread = new Thread(() -> {
            try {
                String message = "这是通过管道传输的实时数据！\n再来一行测试内容～";
                out.write(message.getBytes());
                out.flush();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    out.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });

        // 读线程：负责从管道读取数据
        Thread readerThread = new Thread(() -> {
            byte[] buffer = new byte[1024];
            int bytesRead;
            try {
                while ((bytesRead = in.read(buffer)) != -1) {
                    System.out.println("读取到的数据：" + new String(buffer, 0, bytesRead));
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });

        // 启动线程
        writerThread.start();
        readerThread.start();
    }
}

额外注意事项 #

• 缓冲区大小：默认管道缓冲区是1024字节，如果需要传输大量数据，可以在构造PipedInputStream时指定更大的缓冲区，比如new PipedInputStream(out, 4096)；
• 流的关闭：务必在finally块中关闭流，避免资源泄漏；
• 异常处理：管道流操作可能抛出IOException，比如管道被意外关闭，需要合理捕获处理；
• 替代方案：如果不需要实时传输，只是想把写入的字节一次性读取，可以用ByteArrayOutputStream，写完后通过toByteArray()获取字节数组再转成ByteArrayInputStream，但这种方式无法做到实时同步。

内容的提问来源于stack exchange，提问作者Edwin Dalorzo