Java之IO(八)PipedIutputStream和PipedOutputStream

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

转载请注明源出处：http://www.cnblogs.com/lighten/p/7056278.html

1.前言

本章介绍Java的IO体系中最后一对字节流--管道流。之前在字节数组流的时候就说过，其可以充当输入输出流的转换作用，Java中还有一个管道流可以完成相似的功能，但是其主要作用是用于不同线程间的通讯，下面就具体讲一下管道流是如何实现的，以及相关例子。

值得注意的是，在JDK源码注释中提到了，通常使用一个管道输出流关联一个管道输入流形成管道会话。通常输入流和输出流是不在一个线程中的，如果在同一个线程中使用，可能会造成死锁。当无法从输入流中读取数据的时候，输出流会被中断。

2.PipedOutputStream

管道输出流十分简单，其继承OutputStream，具体类结构如下：

其接受一个管道输入流，调用connect方法，判断通过后会重置相关数据并赋值。当然也可以直接new一个管道输出流，再通过connect方法关联。

两个write方法都和输出流没有关系，是通过输入流来进行操作的。

flush方法和close方法也是通过管道输入流进行操作的。

整个输入流的操作，都是通过输入流完成的，所以接下来我们主要关注输入流做了什么。

3.PipedInputStream

这个管道输入流的内容也不多：

上面就是主要的内容了，输入流有一个buffer字节数组，构造函数有输出流的时候做了两件事情，一个是initPipe初始化这个buffer，默认大小1024个字节，另一个就是调用connect方法，connect方法实际上就是调用输出流的connect，反过来修改了输入流的相关字段。如果输入流构造时没有输出流，就需要使用connect方法进行关联。但是切记，不管是输入流初始化和输出流初始化，connect方法无论是通过什么途径，都只能调用一次，也就是两个对象相互关联只能发生一次。

下面我们关注一下输出流write所调用输入流的receive都做了些什么：

管道流输出的时候就会调用输入管道流的receive方法，最终写入buffer中。注意buffer写完了就会重置写入下标in。receive一个数组也是判断buffer剩余空间是否足够而已。read方法就是读取这个数组中的内容了：

3.1 死锁探究

之前说了，如果写入和读取这两个操作是在同一线程可能会发生死锁，这里具体看下是如何死锁的，测试代码如下：

@Test
public void test() throws IOException {
PipedInputStream pis = new PipedInputStream();
PipedOutputStream pos = new PipedOutputStream(pis);
byte[] read = new byte[10];
byte[] write = new byte[100];
int count = 0;
while(true) {
pos.write(write);
pis.read(read);
System.out.println("完成写入读取次数："+(++count));
}
}

结果如下：

计数到11的时候就不进行下去了，很显然死锁了。这个是怎么产生的呢？回顾上面源码：

写入数据时，调用了receive方法，获得pis的对象锁。写完了数据就会释放锁了。read方法也需要获取锁，然后读完了就释放锁。乍一看好像没什么问题啊，不应该产生死锁的啊。但是问题出在write方法调用receive的一个方法awaitSpace上。因为写入比读取速度快，按照源代码的做法就会造成in的值追上out，然后就一直等待输入流使用，由于输入流和输出流在同一个线程，这里就变成了一个死循环了，wait之后依旧是在等写入消耗。写入比读取快也会造成wait。注释上说的死锁不知是否是这个样子，如果是那么说死锁就感觉有点不恰当了。

3.2 写入读取操作说明

读写操作在不同线程的时候，数据从管道输出流中写入，调用管道输入流的receive方法。里面有一个缓存数组buffer，用于接收write方法写入的字节。输入流有两个下标，一个in用于标记当前缓存到的字节数，一个out用于标记read方法读取buffer的位置。写入的时候，如果in==out，则暂停写入，因为此时判断写入快过读取，防止数据被覆盖，这个时候写入线程就会挂起，等待读取线程读取数据。读取线程如果读到in==out则认为写入完成，读取也就完成了。最初in是等于-1的，所以第一次读取的时候，其也是一个挂起读取线程等待写入的过程。之后in只有在读取线程读完了所有写入的时候才会为-1。下面这段在read()方法中，只有第一次读取可能进入，因为如果有写入的时候in就不会小于0。

问题来了，如果写入线程挂起了，读取线程读到目前写入位置，即in==out时，是判断读取完成的，这个时候会出现什么现象呢？

@Test
public void test2() throws IOException, InterruptedException {
PipedInputStream pis = new PipedInputStream();
PipedOutputStream pos = new PipedOutputStream(pis);
CountDownLatch latch = new CountDownLatch(2);
new Thread(new Runnable() {
@Override
public void run() {
byte[] b = new byte[1025];
b[1024] = 1;
try {
pos.write(b , 0, 1022);
Thread.sleep(2000);
pos.write(b, 1022, b.length);
pos.close();
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown();
}
}
}).start();new Thread(new Runnable() {
@Override
public void run() {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] buf = new byte[1024];
int len = 0;
try {
byte b;
while((b = (byte) pis.read()) != -1) {
baos.write(b);
}
} catch (IOException e) {
e.printStackTrace();
}
byte[] result = baos.toByteArray();
System.out.println(result.length + ":" + Arrays.toString(result));
latch.countDown();
}
}).start();
latch.await();
}

结果是：

会出现异常，且只能读取到第一次写入的1022个字节。这就会产生一个问题，在实际使用的时候就会发生先写入一批数据，再写入一批数据，但是可能读取判断第一批结束了的尴尬情况。有没有什么好的方法解决这个问题呢？至少我没有找到什么好办法。使用管道流的时候最好一次写入，并且初始化管道流的buffer大小最好是写入的大小。其它的方法并不能完全防止这种现象发生，只是概率小一点而已，比如控制读写速度。如果真的要实现多次写入并且要可靠，我能想到的办法就只有，读写线程共用一个锁，再加一个流结束标志符了。但是这种方法也要万分小心，在开始很容易产生死锁，如果读线程先获取公共锁，内部又获取了管道流的锁，即便释放了管道流的锁，写线程也拿不到外层的公共锁。下面给一个demo，仅供参考，可能会有问题：

static boolean end = false;
@Test
public void test3() throws IOException, InterruptedException {
PipedInputStream pis = new PipedInputStream();
PipedOutputStream pos = new PipedOutputStream(pis);
Object monitor = new Object();
CountDownLatch latch = new CountDownLatch(2);
new Thread(new Runnable() {
@Override
public void run() {
int count = 3;
while(count-->0) {
synchronized (monitor) {
System.out.println(count);
byte[] b = new byte[1022];
try {
pos.write(b , 0, b.length);
Thread.sleep(2000);
pos.write(1);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
monitor.notifyAll();
try {
monitor.wait(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
end = true;
try {
pos.close();
} catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
}
}).start();new Thread(new Runnable() {
@Override
public void run() {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] buf = new byte[1024];
int len = 0;
try {
System.out.println("reading");
synchronized (monitor) {
while (!end) {
len = pis.read(buf);
System.out.println("reading...");
baos.write(buf, 0, len);
monitor.notifyAll();
try {
monitor.wait(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
byte[] result = baos.toByteArray();
System.out.println(result.length + ":" + Arrays.toString(result));
latch.countDown();
}
}).start();
latch.await();
}