如何在 Java 中使用 PipedInputStream 在两个线程之间建立直接的字节通讯管道

如何在 Ja va 中使用 PipedInputStream 在两个线程之间建立直接的字节通讯管道

在 Ja va 中实现线程间通信，PipedInputStream 和 PipedOutputStream 这对“管道流”是绕不开的经典组合。它们能建立一个直接的字节通道，让数据从一个线程的“笔尖”流向另一个线程的“眼前”。不过，这对组合用起来有不少讲究，稍不注意就会踩坑。今天，我们就来聊聊如何正确地驾驭它们。

为什么 PipedInputStream 单独 new 出来会抛 IOException: Pipe not connected

这个问题困扰过不少开发者。核心原因很简单：PipedInputStream 生来就不是一个“独行侠”，它必须与它的另一半——PipedOutputStream——配对使用，并且这个“牵手”动作必须在读线程开始工作前完成。

最常见的错误场景是这样的：先启动读线程，读线程立刻尝试 read()，此时才去调用 connect() 方法连接输出流。结果呢？管道尚未建立，读端却已经伸手要数据了，自然就会抛出“Pipe not connected”异常。

那么，正确的“牵手”姿势有哪些？

• 推荐方式一：构造时即绑定。 在创建输入流时，直接将已创建好的输出流传入构造函数：new PipedInputStream(pipedOutputStream)。一步到位，最为稳妥。
• 推荐方式二：显式连接。 分别创建输入流和输出流，然后在任何 I/O 操作开始之前，调用 pipedInputStream.connect(pipedOutputStream) 完成连接。
• 务必警惕： 避免在不同线程中交叉执行连接和启动操作，线程调度的不确定性很容易导致连接失败。

简单来说，确保管道在数据传输开始前就已畅通，是避免这个异常的关键。

如何让读线程不卡死、写线程不丢数据

管道建立后，下一个挑战就是协调读写双方的行为。PipedInputStream 内部维护了一个默认大小为 1KB 的环形缓冲区（大小可通过构造函数指定）。这个设计决定了它的工作模式：

• 读端： 调用 read() 时，如果缓冲区没有数据，线程就会阻塞，耐心等待写端写入。
• 写端： 调用 write() 时，如果缓冲区已满（比如读端消费太慢），线程同样会阻塞，直到读端腾出空间。

这不是 bug，而是管道流实现同步通信的机制。要让整个流程顺畅结束，关键在于生命周期的协同，特别是结束信号的传递。

这里有几个必须牢记的要点：

• 读线程的循环： 读线程应该使用一个循环，持续调用 read() 直到它返回 -1（即文件结束符 EOF），然后再优雅退出。不要仅仅依赖捕获 IOException 来判断流是否结束。
• 写线程的职责： 写线程在完成所有数据写入后，务必调用 pipedOutputStream.close()。这个关闭动作会向管道另一端的读线程发送 EOF 信号，告知其数据已全部送达。
• 注意： 只调用 flush() 是无法触发 EOF 的，读线程会一直等待下去。
• 异常终止： 如果需要中途强制终止通信，对任意一端调用 close()，都会导致另一端正在阻塞的 I/O 操作立即抛出 IOException。

一句话总结：写端负责用 close() 说“再见”，读端负责听到“再见”（读到 -1）后离开。

实际代码中怎么组织线程与管道对象

要让代码清晰可控，推荐的结构是：由主线程创建好配对的管道流，然后分别启动读、写线程，并将对应的流对象作为参数传递进去。避免让线程自己去创建流，那样很容易导致作用域混乱和生命周期管理问题。

下面是一个典型的示例代码结构：

PipedOutputStream pos = new PipedOutputStream();
PipedInputStream pis = new PipedInputStream(pos); // 构造即连接

Thread writer = new Thread(() -> {
    try {
        pos.write("hello".getBytes());
        pos.close(); // 关键：发送 EOF 信号
    } catch (IOException e) {
        e.printStackTrace();
    }
});

Thread reader = new Thread(() -> {
    try {
        int b;
        while ((b = pis.read()) != -1) { // 循环读取直到 EOF
            System.out.print((char) b);
        }
        System.out.println(" — read done");
    } catch (IOException e) {
        // 可能是 writer 异常关闭，或管道被中断
    }
});

writer.start();
reader.start();

在实际使用中，还有几个细节值得注意：

• PipedInputStream 和 PipedOutputStream 必须严格配对使用，不能与其他流对象混用。
• 谨慎包装：尽量不要将 PipedInputStream 再包装成 BufferedInputStream。额外的缓冲层可能会掩盖 EOF 信号的及时到达，或者改变 I/O 阻塞的时机，引入难以调试的问题。
• 字符通信：如果需要传输文本（字符），可以在管道流之上使用 InputStreamReader 和 OutputStreamWriter，但底层通信依然是字节流。

替代方案比 PipedInputStream 更合适吗

PipedInputStream 适用于经典的“单生产者-单消费者”线程协作场景，简单直接。然而，它的能力边界也很明显：不支持多个写者或读者、没有超时控制、不具备非阻塞 I/O 能力，也缺乏背压反馈机制（即写端无法直接感知读端的处理能力）。

一旦业务逻辑变得复杂，它的局限性就暴露出来了：

• 多线程安全写？ 不行。PipedOutputStream 本身不是线程安全的，多个线程同时写入需要开发者自行加锁同步。
• 想设置读超时？ 不行。API 没有提供 setReadTimeout() 这样的方法。如果想实现超时，只能依靠中断线程，或者考虑使用 ja va.nio.channels.Pipe（那是另一套基于通道的模型）。

那么，有哪些更灵活的替代方案呢？

• BlockingQueue： 这是更通用、更强大的选择。它天然支持多生产者和多消费者，队列容量可控，能更好地实现生产消费之间的解耦与流量控制。
• Exchanger： 适用于两个线程需要进行双向、单次数据交换的场景。
• 现代异步模式： 在新项目中，更流行的做法是使用 CompletableFuture 配合回调函数，或者直接使用线程安全的队列（如 ConcurrentLinkedQueue）配合自定义的消息协议，来实现更清晰、更可控的线程间通信。

最后，还有一个容易忽略的点：PipedInputStream 的缓冲区大小在构造后就固定了，无法动态调整。如果在异常堆栈中看到 ja va.io.IOException: Write end dead，这通常意味着写端已经关闭了流，但读端还在尝试读取。这并非并发 Bug，而是读写两端的生命周期没有对齐的信号，提醒我们需要检查流程设计。