Java 输入与输出（I\O）之管道流介绍

管道流概述
在Java中，管道流是一种用于在两个线程之间进行通信的重要技术。管道流提供了一个单向数据流，其中一个线程将数据写入管道，而另一个线程将数据从管道读取。

管道流是用来在多个线程之间进行通信的Java流。管道流也可分为字节流和字符流：

• 字节管道流：PipedOutputStream 和 PipedInputStream。
• 字符管道流：PipedWriter 和 PipedReader。

PipedOutputStream、PipedWriter 是管道输出流，它们是写入者/生产者/发送者；
PipedInputStream、PipedReader 是管道输入流，它们是读取者/消费者/接收者。

Java的管道流在线程间进行通讯时通常是成对出现的，比如一个PipedInputStream必须和一个PipedOutputStream对象进行连接而产生一个通信管道，PipedOutputStream向管道中写入数据，PipedInputStream从管道中读取数据。
下面是管道流在进程间进行通讯的示意图：

管道流，用于在多个线程之间传递数据。它是先进先出（FIFO）的，确保了数据的顺序性。在使用管道前，我们需要先创建一个管道输入流和一个管道输出流，并将它们连接起来。
基本的管道流操作步骤：
1，创建管道输入流：PipedInputStream pis = new PipedInputStream();
2，创建管道输出流：PipedOutputStream pos = new PipedOutputStream();
3，将管道输入输出流连接起来： pos.connect(pis); // 连接管道流
这样就创建了一个管道，可以在不同的线程之间进行数据传递。例如，一个线程往管道写入数据，另一个线程从管道读取数据。
4，向管道写入数据： pos.write(data);
5，从管道读取数据： int data = pis.read();

字节管道流
这里我们只分析字节管道流，字符管道流原理跟字节管道流一样，只不过底层一个是 byte 数组存储 一个是 char 数组存储的。

java的管道输入与输出实际上使用的是一个循环缓冲数组来实现的。输入流PipedInputStream从这个循环缓冲数组中读数据，输出流PipedOutputStream往这个循环缓冲数组中写入数据。当这个缓冲数组已写满的时候，输出流PipedOutputStream所在的线程将阻塞；当这个缓冲数组为空的时候，输入流PipedInputStream所在的线程将阻塞。

注意事项
在使用管道流之前，需要注意以下要点：

• 管道流仅用于多线程之间的通讯传输信息，若用在同一个线程中可能会造成死锁；
• 管道流的输入输出是成对的，一个输出流只能对应一个输入流，使用构造器或者connect函数进行连接；
• 一对管道流包含一个缓冲区，其默认值为1024个字节，若要改变缓冲区大小，可以使用带有参数的构造函数；
• 管道的读写操作是互相阻塞的，当缓冲区为空时，读操作阻塞；当缓冲区满时，写操作阻塞；
• 管道依附于线程，因此若线程结束，则虽然管道流对象还在，仍然会报错“read dead end”；
• 管道流的读取方法与普通流不同，在阻塞状态pis.read()方法是不返回的。只有当输出流结束输出，调用方法close()关闭输出流时，输入流pis.read()方法才能读到-1值。

请看一个演示例程：
（一）、发送信息的线程

package IOStream.PipedStream;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.io.PipedOutputStream;
import java.io.Reader;
import java.util.concurrent.TimeUnit;public class SenderThread extends Thread {private String path;private PipedOutputStream pos;public SenderThread(PipedOutputStream pos,String path) {this.pos = pos;this.path = path;}@Overridepublic void run() {String message = null;try(FileInputStream in = new FileInputStream(path);Reader r = new InputStreamReader(in);BufferedReader reader = new BufferedReader(r)) {while ( (message = reader.readLine()) != null ) {byte[] buf = message.getBytes();pos.write(buf, 0, buf.length);TimeUnit.SECONDS.sleep(5);} pos.close();} catch (Exception e) {// TODO: handle exception}super.run();}}

（二）、接收信息的线程

package IOStream.PipedStream;import java.io.PipedInputStream;public class RecieverThread extends Thread {private PipedInputStream pis;public RecieverThread(PipedInputStream pis) {this.pis = pis;}@Overridepublic void run() {int len = 0;String msg = null;byte buf[] = new byte[1024];try {while ((len = pis.read(buf)) != -1) {msg = new String(buf, 0 , len);System.out.println("收到信息：" + msg);}pis.close();} catch (Exception e) {// TODO: handle exception}super.run();}
}

（三）、测试主程序

package IOStream.PipedStream;import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;public class PipedStreamTest {public static void main(String[] args) {String fPath = "D:/temp/凉州词.txt";//创建一个线程池ExecutorService executorService = Executors.newCachedThreadPool();try {//创建输入和输出管道流PipedOutputStream pos = new PipedOutputStream();PipedInputStream pis = new PipedInputStream(pos);//创建发送线程和接收线程SenderThread sender = new SenderThread(pos,fPath);RecieverThread reciever = new RecieverThread(pis);//提交给线程池运行发送线程和接收线程executorService.execute(sender);executorService.execute(reciever);} catch (IOException e) {e.printStackTrace();}//通知线程池，不再接受新的任务，并执行完成当前正在运行的线程后关闭线程池。executorService.shutdown();try {//shutdown 后可能正在运行的线程很长时间都运行不完成，这里设置超过1小时，强制执行 Interruptor 结束线程。executorService.awaitTermination(1, TimeUnit.HOURS);} catch (InterruptedException e) {e.printStackTrace();}}}

（四）、演示测试结果

参考文献：

Java IO7：管道流、对象流
Java IO（十三）PipedReader 和 PipedWriter
Java IO流详解（八）----其他流的使用
Java IO 之 管道流 原理分析