首页 > 代码库 > Java之IO(八)PipedIutputStream和PipedOutputStream

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();
	}

技术分享

  再次声明,此代码没有经过仔细思考,只是提供一个思路,出现问题概不负责,使用管道流最好还是一次性写入所有数据比较好。

  

Java之IO(八)PipedIutputStream和PipedOutputStream