Java PipedInputStream 类

最后修改时间：2025 年 4 月 16 日

java.io.PipedInputStream 类是一个专门的输入流，可以连接到 PipedOutputStream。它在两个线程之间创建一个通信管道，允许一个线程写入数据，另一个线程可以读取数据。

PipedInputStream 通常用于线程间通信。管道具有有限的缓冲区大小，如果缓冲区已满，写入操作将会阻塞。同样，如果没有数据可用，读取操作将会阻塞。两个流必须连接才能正常工作。

PipedInputStream 类概述

PipedInputStream 扩展了 InputStream 并提供基于管道的输入操作。它必须在构造时或之后连接到 PipedOutputStream。默认管道大小为 1024 字节。

public class PipedInputStream extends InputStream {
    public PipedInputStream();
    public PipedInputStream(int pipeSize);
    public PipedInputStream(PipedOutputStream src);
    public PipedInputStream(PipedOutputStream src, int pipeSize);
    public void connect(PipedOutputStream src);
    public synchronized int read();
    public synchronized int read(byte[] b, int off, int len);
    public synchronized int available();
    public void close();
}

上面的代码显示了 PipedInputStream 提供的关键方法。这些方法允许从连接的输出流读取数据。该类是线程安全的，可以从多个线程并发访问。

创建 PipedInputStream

可以以几种方式创建 PipedInputStream - 可以连接到 PipedOutputStream，也可以不连接。可以指定管道大小，也可以将其保留为默认值。始终确保在使用前正确连接。

Main.java

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

public class Main {

    public static void main(String[] args) {
        try {
            // Create unconnected pipe
            PipedInputStream pis1 = new PipedInputStream();
            
            // Create with default pipe size (connected)
            PipedOutputStream pos1 = new PipedOutputStream();
            PipedInputStream pis2 = new PipedInputStream(pos1);
            
            // Create with custom pipe size (4KB)
            PipedOutputStream pos2 = new PipedOutputStream();
            PipedInputStream pis3 = new PipedInputStream(pos2, 4096);
            
            System.out.println("Created three PipedInputStream instances");
            
            pis1.close();
            pis2.close();
            pis3.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

此示例演示了创建 PipedInputStream 的不同方法。第一个创建未连接的管道，而其他管道在构造期间连接。完成后务必关闭流以释放资源。连接的输出流不会自动关闭。

基本管道通信

此示例显示了 PipedInputStream 与 PipedOutputStream 的基本用法。一个线程写入数据，而另一个线程读取数据。管道自动处理线程之间的同步。

Main.java

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

public class Main {

    public static void main(String[] args) {
        try {
            PipedOutputStream pos = new PipedOutputStream();
            PipedInputStream pis = new PipedInputStream(pos);
            
            // Writer thread
            new Thread(() -> {
                try {
                    pos.write("Hello from pipe!".getBytes());
                    pos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            
            // Reader thread
            new Thread(() -> {
                try {
                    int data;
                    while ((data = pis.read()) != -1) {
                        System.out.print((char) data);
                    }
                    pis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

此示例演示了使用管道进行的基本线程间通信。 writer 线程通过 PipedOutputStream 发送数据。 reader 线程通过 PipedInputStream 接收数据。使用后必须正确关闭两个流。

将字节读入数组

为了获得更好的性能，一次将多个字节读入字节数组。这减少了方法调用并提高了效率。 read 方法返回实际读取的字节数。

Main.java

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

public class Main {

    public static void main(String[] args) {
        try {
            PipedOutputStream pos = new PipedOutputStream();
            PipedInputStream pis = new PipedInputStream(pos, 2048);
            
            // Writer thread
            new Thread(() -> {
                try {
                    for (int i = 0; i < 100; i++) {
                        pos.write(("Data " + i + "\n").getBytes());
                    }
                    pos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            
            // Reader thread
            new Thread(() -> {
                try {
                    byte[] buffer = new byte[50];
                    int bytesRead;
                    
                    while ((bytesRead = pis.read(buffer)) != -1) {
                        System.out.print(new String(buffer, 0, bytesRead));
                    }
                    pis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

此示例显示了从管道批量读取到字节数组中。 writer 发送 100 行数据。 reader 以 50 字节的块处理数据。管道大小设置为 2048 字节，以适应更大的数据传输。

检查可用字节

available 方法返回可以在不阻塞的情况下读取的字节数。这对于检查是否准备好从管道读取数据非常有用。

Main.java

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

public class Main {

    public static void main(String[] args) {
        try {
            PipedOutputStream pos = new PipedOutputStream();
            PipedInputStream pis = new PipedInputStream(pos);
            
            // Writer thread
            new Thread(() -> {
                try {
                    System.out.println("Writer: Sending data...");
                    pos.write("Sample data".getBytes());
                    Thread.sleep(2000); // Simulate delay
                    pos.write("More data".getBytes());
                    pos.close();
                } catch (IOException | InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
            
            // Reader thread
            new Thread(() -> {
                try {
                    while (true) {
                        int available = pis.available();
                        if (available > 0) {
                            byte[] data = new byte[available];
                            pis.read(data);
                            System.out.println("Reader: Got " + 
                                new String(data));
                        }
                        if (available == -1) break;
                        Thread.sleep(500);
                    }
                    pis.close();
                } catch (IOException | InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

此示例演示了检查管道中的可用字节。 writer 分两部分发送数据，并有延迟。 reader 定期检查可用数据。当您想避免阻塞读取时，此方法很有用。

创建后连接流

可以使用 connect 方法在创建后将 PipedInputStream 连接到 PipedOutputStream。调用此方法时，两个流都必须未连接。

Main.java

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

public class Main {

    public static void main(String[] args) {
        try {
            PipedInputStream pis = new PipedInputStream();
            PipedOutputStream pos = new PipedOutputStream();
            
            // Connect them after creation
            pis.connect(pos);
            
            // Writer thread
            new Thread(() -> {
                try {
                    pos.write("Data sent through connected pipes".getBytes());
                    pos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            
            // Reader thread
            new Thread(() -> {
                try {
                    int data;
                    while ((data = pis.read()) != -1) {
                        System.out.print((char) data);
                    }
                    System.out.println();
                    pis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

此示例显示了如何在创建后连接流。 connect 方法在现有流之间建立管道。要使此方法有效，两个流都必须处于未连接状态。通信的工作方式与基于构造函数的连接相同。

处理管道断开

当管道断开（writer 在 reader 完成之前关闭）时，会发生 IOException。正确的错误处理可确保可靠的线程间通信。

Main.java

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

public class Main {

    public static void main(String[] args) {
        try {
            PipedOutputStream pos = new PipedOutputStream();
            PipedInputStream pis = new PipedInputStream(pos);
            
            // Writer thread (closes immediately)
            new Thread(() -> {
                try {
                    pos.write("Partial data".getBytes());
                    pos.close(); // Closes before reader finishes
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            
            // Reader thread (slow reader)
            new Thread(() -> {
                try {
                    Thread.sleep(1000); // Delay reading
                    int data;
                    while ((data = pis.read()) != -1) {
                        System.out.print((char) data);
                        Thread.sleep(500); // Slow processing
                    }
                    pis.close();
                } catch (IOException | InterruptedException e) {
                    System.err.println("Pipe error: " + e.getMessage());
                }
            }).start();
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

此示例演示了管道断开处理。 writer 在 reader 完成处理之前关闭。当 reader 尝试从断开的管道读取时，会收到 IOException。正确的错误处理可以防止应用程序崩溃。

来源

Java PipedInputStream 类文档

在本文中，我们介绍了 Java PipedInputStream 类的基本方法和功能。理解这些概念对于在 Java 应用程序中使用线程间通信至关重要。

作者

我叫 Jan Bodnar，是一位拥有多年经验的专注程序员。我于 2007 年开始撰写编程文章，至今已撰写了 1,400 多篇文章和八本电子书。凭借八年以上的教学经验，我致力于分享我的知识并帮助他人掌握编程概念。

列出所有Java教程。