Java 标准库在 java.io 包中提供了 PipedOutputStream 和 PipedInputStream，它们可以组合成一条 JVM 内部的字节流管道。写入端把数据推进输出流，读取端从输入流中按顺序获取数据。与文件、Socket 等 IO 通道不同，管道流不依赖外部资源，所有传递都发生在堆内存的缓冲区中，因此非常适合多线程任务之间低延迟的数据流通信。只要生产线程和消费线程跑在同一个进程内，并且数据可以边生成边处理，这种管道模型就能避免等待完整结果落盘或跨网络传输带来的额外耗时。
管道流的核心是一个固定大小的内存缓冲区和内置的 wait 与 notifyAll 机制，它既具备流式接口，又能提供背压控制。生产快、消费慢时，写线程会在缓冲区满时阻塞；消费快、生产慢时，读线程会在缓冲区空时阻塞。下面从连接模型、缓冲区配置、多线程协作和帧协议设计几个角度具体展开。
一、PipedStream 的连接模型与基础用法
Java 的 PipedOutputStream 与 PipedInputStream 必须成对出现，通过连接后才能读写。常用两种连接方式：一是直接在 PipedInputStream 构造函数中传入输出流，二是在代码里调用 connect 方法。无论哪种方式，底层都会把输出流绑定到输入流的内部缓冲区上，并检查重复连接和未连接状态。连接完成后，输出流的写入操作会直接作用于输入流的缓冲区，读取端可以立即感知数据到达。
下面是一个基础的多线程示例，生产者向管道写入消息，消费者从管道读取消息。示例中显式指定了 4KB 的内部缓冲区，并借助线程池让读写两端运行在不同线程中。
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
public class PipeBasicDemo {
public static void main(String[] args) throws Exception {
PipedOutputStream out = new PipedOutputStream();
PipedInputStream in = new PipedInputStream(out, 4096); // 指定 4KB 缓冲区
ExecutorService pool = Executors.newFixedThreadPool(2);
Future<?> producer = pool.submit(() -> {
try (out) {
for (int i = 0; i < 20; i++) {
String msg = "message-" + i + "\n";
out.write(msg.getBytes(StandardCharsets.UTF_8));
}
// 关闭输出端，读端才会收到 -1
} catch (IOException e) {
e.printStackTrace();
}
return null;
});
Future<?> consumer = pool.submit(() -> {
try (in) {
byte[] buf = new byte[64];
int n;
while ((n = in.read(buf)) != -1) {
String text = new String(buf, 0, n, StandardCharsets.UTF_8);
System.out.print(text);
}
} catch (IOException e) {
e.printStackTrace();
}
return null;
});
producer.get();
consumer.get();
pool.shutdown();
}
}
运行这个例子时，生产者每写完一条消息，消费线程就能在 read 中拿到数据。即使消费者先启动并阻塞在 read 上，只要生产者随后写入，消费线程会被立即唤醒。这种及时性得益于 PipedInputStream 内部的 wait 与 notifyAll 协调，而不是轮询。因此，只要通道两端的线程都处于活动状态，管道流比较适合低延迟数据流通信。需要特别注意的是，不要在主线程中先写后读同一个管道。如果写缓冲被填满而没有任何其他线程去读取，程序会永久阻塞。
二、缓冲区大小如何影响延迟
PipedInputStream 默认内部缓冲为 1024 字节。这个值对于一些较小的消息流可能不够用，写线程会频繁进入阻塞等待，直到读线程取走部分数据。阻塞虽然能保护内存不被无限堆积，但每次阻塞和唤醒都会引入线程调度开销。反过来，如果缓冲区配置过大，例如 1MB，而读取逻辑偶发变慢，大量数据会在管道中排队，单条消息从写入到读取的平均延迟会被拉高。低延迟场景中通常需要结合单帧大小和消费速度，把缓冲区控制在 4KB 到 64KB 之间。对于固定几十字节的小帧，4KB 通常足够；对于批量日志或图像数据块，可以适当提高。
管道流本身是面向字节的，写入端如果没有帧边界，读取端很难还原消息。为了兼顾低延迟和消息完整性，可以在应用层加入很轻的长度前缀。下面是一个写入帧的辅助方法，使用大端字节序保存负载长度，这样读取端可以精确地按帧取数据。
private static void writeFrame(PipedOutputStream out, byte[] payload) throws IOException {
int length = payload.length;
out.write((length >>> 24) & 0xFF);
out.write((length >>> 16) & 0xFF);
out.write((length >>> 8) & 0xFF);
out.write(length & 0xFF);
out.write(payload);
out.flush();
}
读取端对应需要循环读取长度字段，再按长度读取整个负载。下面的 readFrame 会处理管道关闭和半帧情况。读取时尽量复用同一个 byte 数组，避免每条消息都分配新数组，减少 GC 压力对延迟的影响。
private static byte[] readFrame(PipedInputStream in) throws IOException {
int b0 = in.read();
if (b0 == -1) {
return null;
}
int b1 = in.read();
int b2 = in.read();
int b3 = in.read();
if (b1 == -1 || b2 == -1 || b3 == -1) {
throw new EOFException("pipe closed inside frame header");
}
int len = (b0 << 24) | (b1 << 16) | (b2 << 8) | b3;
byte[] data = new byte[len];
int offset = 0;
while (offset < len) {
int n = in.read(data, offset, len - offset);
if (n == -1) {
throw new EOFException("pipe closed during frame body");
}
offset += n;
}
return data;
}
对于 PipedOutputStream 的 flush 方法，在管道流实现中数据写入内部缓冲区后即对读取端可见，flush 并不会带来额外持久化保障，但调用 flush 可以让代码意图更明确。真正影响延迟的是避免一次只写一个字节并频繁唤醒读取端。可以先把小字段组装成完整帧，再一次 write 出去。也不建议在管道外再包 BufferedOutputStream，因为额外缓冲会让数据滞留，反而抵消管道流即时可见的优势。
三、多线程协作中的阻塞与关闭顺序
PipedStream 本质上是一对一的同步字节通道，生产者和消费者必须在不同线程中运行。写线程在缓冲区剩余空间不足时阻塞，读线程在缓冲区为空时阻塞。这个模型天然提供背压，有助于避免内存被瞬时数据撑爆。但低延迟应用需要特别关注两个线程的调度关系。如果读取线程中执行复杂解析、写数据库或远程调用，消费速度会下降，进而让写线程被阻塞，整体吞吐和单条消息延迟都会恶化。建议读取线程只做快速接收和转发，例如把完整帧交给另一个处理器或队列。
关闭顺序错误是 PipedStream 常见的坑。生产者完成写入后必须关闭输出流，读取端的 read 才会返回 -1 结束循环。如果消费者提前关闭输入流，生产者继续 write 会收到 IOException。反之，如果生产者异常退出但没有关闭输出流，消费者会一直阻塞在 read 上。下面的模板展示了推荐的关闭方式。
public static void runPipe(PipedOutputStream out, PipedInputStream in) throws Exception {
Thread producer = new Thread(() -> {
try (out) {
out.write("done".getBytes(StandardCharsets.UTF_8));
} catch (IOException e) {
e.printStackTrace();
}
});
Thread consumer = new Thread(() -> {
try (in) {
byte[] buf = new byte[16];
int n;
while ((n = in.read(buf)) != -1) {
// 快速处理或转发
}
} catch (IOException e) {
e.printStackTrace();
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
}
还要避免多个线程同时写同一个 PipedOutputStream，因为单次 write 方法虽然同步，但多次 write 之间没有原子性，两个生产者的数据可能交叉成无法解析的字节序列。同样，多个消费者同时读取一个 PipedInputStream 会导致数据被随机瓜分。每对管道流只服务一个生产者与一个消费者，这是使用 PipedStream 实现稳定通信的重要约束。如果需要多生产者或多消费者，应在写端加锁或在进入管道前先通过 BlockingQueue 这类结构进行聚合。
四、轻量帧协议与选型边界
在低延迟数据流通信中，光有字节通道还不够。通常至少要定义消息类型、长度和负载格式。消息类型可以用一个字节表示，长度用两到四个字节表示，负载直接放原始数据。下面是一个更完整的写法，将类型字段和长度字段一起写入，读取端返回一个简单的 Message 对象。
public static void writeMessage(PipedOutputStream out, int type, byte[] payload) throws IOException {
int length = payload.length;
out.write(type);
out.write((length >>> 16) & 0xFF);
out.write((length >>> 8) & 0xFF);
out.write(length & 0xFF);
out.write(payload);
out.flush();
}
public static Message readMessage(PipedInputStream in) throws IOException {
int type = in.read();
if (type == -1) {
return null;
}
int len = (in.read() << 16) | (in.read() << 8) | in.read();
byte[] body = new byte[len];
int offset = 0;
while (offset < len) {
int n = in.read(body, offset, len - offset);
if (n == -1) {
throw new EOFException("pipe closed in message body");
}
offset += n;
}
return new Message(type, body);
}
负载内容尽量避免使用 Java 原生对象序列化。对象序列化会生成大量元数据，并且反序列化需要反射，延迟不低。可以根据业务选择 JSON、Protobuf 或自定义二进制格式。JSON 的文本解析速度不如二进制，但可读性和调试成本更低；Protobuf 的编码解码更快，适合对延迟敏感的内部服务；如果只是传输数值、坐标、状态码等基础数据，直接按大端字节序写入 DataOutputStream 会非常轻量。
PipedStream 的优势在于它是 JDK 标准库的一部分，接口是熟悉的 InputStream 与 OutputStream，可以和压缩、加密、解析等流式组件对接。但它仍然是同步阻塞模型，内部使用锁和 wait 通知，不是无锁队列。如果业务已经使用 Disruptor 这类高性能无锁组件，并且每条消息都追求微秒级延迟，PipedStream 不一定是最优解。对于大多数线程间流式传输和中等吞吐场景，通过合理设置缓冲、保持一对一模型和快速消费，PipedStream 完全可以把延迟控制在毫秒级以内，同时保持代码简单、易维护。
Java管道流PipedStream多线程通信修改时间：2026-10-06 05:09:48