java io系列04之 管道(PipedOutputStream和PipedInputStream)的简介,源码分析和示例
- 格式：doc
- 大小：159.50 KB
- 文档页数：17
java 管道介绍在java中，PipedOutputStream和PipedInputStream分别是管道输出流和管道输入流。
它们的作用是让多线程可以通过管道进行线程间的通讯。
在使用管道通信时，必须将PipedOutputStream和PipedInputStream配套使用。
使用管道通信时，大致的流程是：我们在线程A中向PipedOutputStream中写入数据，这些数据会自动的发送到与 PipedOutputStream对应的PipedInputStream中，进而存储在PipedInputStream的缓冲中；此时，线程B通过读取PipedInputStream中的数据。
就可以实现，线程A和线程B的通信。
PipedOutputStream和PipedInputStream源码分析下面介绍PipedOutputStream和PipedInputStream的源码。
在阅读它们的源码之前，建议先看看源码后面的示例。
待理解管道的作用和用法之后，再看源码，可能更容易理解。
此外，由于在“java io系列03之 ByteArrayOutputStream的简介,源码分析和示例(包括OutputStream)”中已经对PipedOutputStream的父类OutputStream进行了介绍，这里就不再介绍OutputStream。
在“java io系列02之 ByteArrayInputStream的简介,源码分析和示例(包括InputStream)”中已经对PipedInputStream的父类InputStream进行了介绍，这里也不再介绍InputStream。
1. PipedOutputStream 源码分析(基于jdk1.7.40)1package java.io;23import java.io.*;45public class PipedOutputStream extends OutputStream {67// 与PipedOutputStream通信的PipedInputStream对象8private PipedInputStream sink;910// 构造函数，指定配对的PipedInputStream11public PipedOutputStream(PipedInputStream snk) throws IOException {12 connect(snk);13 }1415// 构造函数16public PipedOutputStream() {17 }1819// 将“管道输出流” 和“管道输入流”连接。
20public synchronized void connect(PipedInputStream snk) throws IOException {21if (snk == null) {22throw new NullPointerException();23 } else if (sink != null || snk.connected) {24throw new IOException("Already connected");25 }26// 设置“管道输入流”27 sink = snk;28// 初始化“管道输入流”的读写位置29// int是PipedInputStream中定义的，代表“管道输入流”的读写位置30 snk.in = -1;31// 初始化“管道输出流”的读写位置。
32// out是PipedInputStream中定义的，代表“管道输出流”的读写位置33 snk.out = 0;34// 设置“管道输入流”和“管道输出流”为已连接状态35// connected是PipedInputStream中定义的，用于表示“管道输入流与管道输出流”是否已经连接36 snk.connected = true;37 }3839// 将int类型b写入“管道输出流”中。
40// 将b写入“管道输出流”之后，它会将b传输给“管道输入流”41public void write(int b) throws IOException {42if (sink == null) {43throw new IOException("Pipe not connected");44 }45 sink.receive(b);46 }4748// 将字节数组b写入“管道输出流”中。
49// 将数组b写入“管道输出流”之后，它会将其传输给“管道输入流”50public void write(byte b[], int off, int len) throws IOException {51if (sink == null) {52throw new IOException("Pipe not connected");53 } else if (b == null) {54throw new NullPointerException();55 } else if ((off < 0) || (off > b.length) || (len < 0) ||56 ((off + len) > b.length) || ((off + len) < 0)) {57throw new IndexOutOfBoundsException();58 } else if (len == 0) {59return;60 }61// “管道输入流”接收数据62 sink.receive(b, off, len);63 }6465// 清空“管道输出流”。
66// 这里会调用“管道输入流”的notifyAll()；67// 目的是让“管道输入流”放弃对当前资源的占有，让其它的等待线程(等待读取管道输出流的线程)读取“管道输出流”的值。
68public synchronized void flush() throws IOException {69if (sink != null) {70synchronized (sink) {71 sink.notifyAll();72 }73 }74 }7576// 关闭“管道输出流”。
77// 关闭之后，会调用receivedLast()通知“管道输入流”它已经关闭。
78public void close() throws IOException {79if (sink != null) {80 sink.receivedLast();81 }82 }83 }2. PipedInputStream 源码分析(基于jdk1.7.40)1package java.io;23public class PipedInputStream extends InputStream {4// “管道输出流”是否关闭的标记5boolean closedByWriter = false;6// “管道输入流”是否关闭的标记7volatile boolean closedByReader = false;8// “管道输入流”与“管道输出流”是否连接的标记9// 它在PipedOutputStream的connect()连接函数中被设置为true10boolean connected = false;1112 Thread readSide; // 读取“管道”数据的线程13 Thread writeSide; // 向“管道”写入数据的线程1415// “管道”的默认大小16private static final int DEFAULT_PIPE_SIZE = 1024;1718protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE;1920// 缓冲区21protected byte buffer[];2223//下一个写入字节的位置。
in==out代表满，说明“写入的数据”全部被读取了。
24protected int in = -1;25//下一个读取字节的位置。
in==out代表满，说明“写入的数据”全部被读取了。
26protected int out = 0;2728// 构造函数：指定与“管道输入流”关联的“管道输出流”29public PipedInputStream(PipedOutputStream src) throws IOException {30this(src, DEFAULT_PIPE_SIZE);31 }3233// 构造函数：指定与“管道输入流”关联的“管道输出流”，以及“缓冲区大小”34public PipedInputStream(PipedOutputStream src, int pipeSize) 35throws IOException {36 initPipe(pipeSize);37 connect(src);38 }3940// 构造函数：默认缓冲区大小是1024字节41public PipedInputStream() {42 initPipe(DEFAULT_PIPE_SIZE);43 }4445// 构造函数：指定缓冲区大小是pipeSize46public PipedInputStream(int pipeSize) {47 initPipe(pipeSize);48 }4950// 初始化“管道”：新建缓冲区大小51private void initPipe(int pipeSize) {52if (pipeSize <= 0) {53throw new IllegalArgumentException("Pipe Size <= 0");54 }55 buffer = new byte[pipeSize];56 }5758// 将“管道输入流”和“管道输出流”绑定。
59// 实际上，这里调用的是PipedOutputStream的connect()函数60public void connect(PipedOutputStream src) throws IOException {61 src.connect(this);62 }6364// 接收int类型的数据b。
65// 它只会在PipedOutputStream的write(int b)中会被调用66protected synchronized void receive(int b) throws IOException { 67// 检查管道状态68 checkStateForReceive();69// 获取“写入管道”的线程70 writeSide = Thread.currentThread();71// 若“写入管道”的数据正好全部被读取完，则等待。
