java IO笔记（PipedInputStream/PipedOutputStream）

时间 2019-12-18

标签 java 笔记 pipedinputstream pipedoutputstream 栏目 Java 繁體版

原文原文链接

在上一篇笔记中讲述了java io 中的文件（file）以及如何用文件流来对文件进行读写操做，本篇则要讲述的是java IO中的管道流。java

java IO中的管道流可使得同一进程中的不一样线程进行通讯，若是不明白进程和线程的区别的话，能够去网上搜搜资料，能够看作提供同一jvm的通讯能力。在java IO中管道的建立须要经过PipedInputStream和PipedOutputStream两个类，能够经过二者的构造方法进行互相关联也能够经过其中的connect方法进行关联。数组

下面将用一个最简单的例子来代表其功能。缓存

package pipedIO;

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

public class PipedIO {

	private PipedInputStream pis = new PipedInputStream();
	private PipedOutputStream pos = new PipedOutputStream();

	public static void main(String[] args) throws IOException {
		PipedIO pipedIO = new PipedIO();
		pipedIO.initThread();
	}

	private void initThread() throws IOException {
		pos.connect(pis);
		Thread input = new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					while (true) {
						String time = new Date().toString();
						pos.write(time.getBytes());
						System.out.println("输出流完成一次数据写出,数据为"+time);
						Thread.sleep(1000);
					}
				} catch (IOException | InterruptedException e) {
					e.printStackTrace();
				}
			}
		});

		Thread output = new Thread(new Runnable() {
			@Override
			public void run() {
				byte[] temp = new byte[1024];
				try {
					int len;
					while ((len = pis.read(temp)) != -1) {
						System.out.println("输入流完成一次数据写入，数据为："+new String(temp, 0, len));
						Thread.sleep(1000);
					}
				} catch (IOException | InterruptedException e) {
					e.printStackTrace();
				}
			}
		});

		input.start();
		output.start();
	}

}

执行上述代码，能够看到以下打印：jvm

从控制台输出能够看出，两个线程之间完成了通讯，看上去十分简单。但管道流真正的使用时，还须要注意一些事项。下面说说管道流的工做原理吧。ide

PipedInputStream.java工具

package java.io;

public class PipedInputStream extends InputStream {
    boolean closedByWriter = false;
    volatile boolean closedByReader = false;
    boolean connected = false;

        /* REMIND: identification of the read and write sides needs to be
           more sophisticated.  Either using thread groups (but what about
           pipes within a thread?) or using finalization (but it may be a
           long time until the next GC). */
    Thread readSide;
    Thread writeSide;

    private static final int DEFAULT_PIPE_SIZE = 1024;

    protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE;

    protected byte buffer[];

    protected int in = -1;

    protected int out = 0;

    public PipedInputStream(PipedOutputStream src) throws IOException {
        this(src, DEFAULT_PIPE_SIZE);
    }

    public PipedInputStream(PipedOutputStream src, int pipeSize)
            throws IOException {
         initPipe(pipeSize);
         connect(src);
    }

    
    public PipedInputStream() {
        initPipe(DEFAULT_PIPE_SIZE);
    }

    public PipedInputStream(int pipeSize) {
        initPipe(pipeSize);
    }

    private void initPipe(int pipeSize) {
         if (pipeSize <= 0) {
            throw new IllegalArgumentException("Pipe Size <= 0");
         }
         buffer = new byte[pipeSize];
    }

    public void connect(PipedOutputStream src) throws IOException {
        src.connect(this);
    }

    protected synchronized void receive(int b) throws IOException {
        checkStateForReceive();
        writeSide = Thread.currentThread();
        if (in == out)
            awaitSpace();
        if (in < 0) {
            in = 0;
            out = 0;
        }
        buffer[in++] = (byte)(b & 0xFF);
        if (in >= buffer.length) {
            in = 0;
        }
    }

    synchronized void receive(byte b[], int off, int len)  throws IOException {
        checkStateForReceive();
        writeSide = Thread.currentThread();
        int bytesToTransfer = len;
        while (bytesToTransfer > 0) {
            if (in == out)
                awaitSpace();
            int nextTransferAmount = 0;
            if (out < in) {
                nextTransferAmount = buffer.length - in;
            } else if (in < out) {
                if (in == -1) {
                    in = out = 0;
                    nextTransferAmount = buffer.length - in;
                } else {
                    nextTransferAmount = out - in;
                }
            }
            if (nextTransferAmount > bytesToTransfer)
                nextTransferAmount = bytesToTransfer;
            assert(nextTransferAmount > 0);
            System.arraycopy(b, off, buffer, in, nextTransferAmount);
            bytesToTransfer -= nextTransferAmount;
            off += nextTransferAmount;
            in += nextTransferAmount;
            if (in >= buffer.length) {
                in = 0;
            }
        }
    }

    private void checkStateForReceive() throws IOException {
        if (!connected) {
            throw new IOException("Pipe not connected");
        } else if (closedByWriter || closedByReader) {
            throw new IOException("Pipe closed");
        } else if (readSide != null && !readSide.isAlive()) {
            throw new IOException("Read end dead");
        }
    }

    private void awaitSpace() throws IOException {
        while (in == out) {
            checkStateForReceive();

            /* full: kick any waiting readers */
            notifyAll();
            try {
                wait(1000);
            } catch (InterruptedException ex) {
                throw new java.io.InterruptedIOException();
            }
        }
    }

    synchronized void receivedLast() {
        closedByWriter = true;
        notifyAll();
    }

   
    public synchronized int read()  throws IOException {
        if (!connected) {
            throw new IOException("Pipe not connected");
        } else if (closedByReader) {
            throw new IOException("Pipe closed");
        } else if (writeSide != null && !writeSide.isAlive()
                   && !closedByWriter && (in < 0)) {
            throw new IOException("Write end dead");
        }

        readSide = Thread.currentThread();
        int trials = 2;
        while (in < 0) {
            if (closedByWriter) {
                /* closed by writer, return EOF */
                return -1;
            }
            if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) {
                throw new IOException("Pipe broken");
            }
            /* might be a writer waiting */
            notifyAll();
            try {
                wait(1000);
            } catch (InterruptedException ex) {
                throw new java.io.InterruptedIOException();
            }
        }
        int ret = buffer[out++] & 0xFF;
        if (out >= buffer.length) {
            out = 0;
        }
        if (in == out) {
            /* now empty */
            in = -1;
        }

        return ret;
    }

   
    public synchronized int read(byte b[], int off, int len)  throws IOException {
        if (b == null) {
            throw new NullPointerException();
        } else if (off < 0 || len < 0 || len > b.length - off) {
            throw new IndexOutOfBoundsException();
        } else if (len == 0) {
            return 0;
        }

        /* possibly wait on the first character */
        int c = read();
        if (c < 0) {
            return -1;
        }
        b[off] = (byte) c;
        int rlen = 1;
        while ((in >= 0) && (len > 1)) {

            int available;

            if (in > out) {
                available = Math.min((buffer.length - out), (in - out));
            } else {
                available = buffer.length - out;
            }

            // A byte is read beforehand outside the loop
            if (available > (len - 1)) {
                available = len - 1;
            }
            System.arraycopy(buffer, out, b, off + rlen, available);
            out += available;
            rlen += available;
            len -= available;

            if (out >= buffer.length) {
                out = 0;
            }
            if (in == out) {
                /* now empty */
                in = -1;
            }
        }
        return rlen;
    }

  
    public synchronized int available() throws IOException {
        if(in < 0)
            return 0;
        else if(in == out)
            return buffer.length;
        else if (in > out)
            return in - out;
        else
            return in + buffer.length - out;
    }

    public void close()  throws IOException {
        closedByReader = true;
        synchronized (this) {
            in = -1;
        }
    }
}

PipedOutputStream.javaoop

package java.io;

import java.io.*;


public class PipedOutputStream extends OutputStream {

    private PipedInputStream sink;

    public PipedOutputStream(PipedInputStream snk)  throws IOException {
        connect(snk);
    }

   
    public PipedOutputStream() {
    }

    public synchronized void connect(PipedInputStream snk) throws IOException {
        if (snk == null) {
            throw new NullPointerException();
        } else if (sink != null || snk.connected) {
            throw new IOException("Already connected");
        }
        sink = snk;
        snk.in = -1;
        snk.out = 0;
        snk.connected = true;
    }

    public void write(int b)  throws IOException {
        if (sink == null) {
            throw new IOException("Pipe not connected");
        }
        sink.receive(b);
    }

    public void write(byte b[], int off, int len) throws IOException {
        if (sink == null) {
            throw new IOException("Pipe not connected");
        } else if (b == null) {
            throw new NullPointerException();
        } else if ((off < 0) || (off > b.length) || (len < 0) ||
                   ((off + len) > b.length) || ((off + len) < 0)) {
            throw new IndexOutOfBoundsException();
        } else if (len == 0) {
            return;
        }
        sink.receive(b, off, len);
    }

    public synchronized void flush() throws IOException {
        if (sink != null) {
            synchronized (sink) {
                sink.notifyAll();
            }
        }
    }

    
    public void close()  throws IOException {
        if (sink != null) {
            sink.receivedLast();
        }
    }
}

上面贴出了PipedInputStream和PipedOutputStream的源码，从源码中不难看出，在PipedOutputStream中封装了一个PipedInputStream，当调用两个类中的的connect方法时，最终都回到了PipedOutputStream中的connect方法，使得两个流进行链接关系。优化

咱们能够看出不管是PipedInputStream的read方法，仍是PipedOutputStream中的write方法，数据都是存放在PipedInputStream中的一个byte数组的缓存中的，该缓存默认大小为1024字节。那么这就有一个问题了，当PipedInputStream中的缓存区已经装满的时候，必需要等到读取PipedInputStream数据缓存并清除相对应的数据时，才能继续往缓存中写入，若是一直等不到，将再次产生相似死锁的状况。this

所以向PipedOutputStream中写入数据的线程不该是负责从对应PipedInputStream中读取数据的惟一线程，下面举例说明：线程

package pipedIO;

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

public class PipedIO1 {

	private PipedInputStream pis = new PipedInputStream();
	private PipedOutputStream pos = new PipedOutputStream();

	public static void main(String[] args) throws IOException {
		PipedIO1 pipedIO1 = new PipedIO1();
		pipedIO1.initThread();
	}

	private void initThread() throws IOException {
		pos.connect(pis);
		Thread input = new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					byte[] inBytes = new byte[500];
					byte[] outBytes = new byte[1000];
					pos.write(inBytes);
					System.out.println("输出流完成一次数据写出");
					int len = pis.read(outBytes);
					System.out.println("输入流完成一次数据读出");
					while (len != -1) {
						pos.write(outBytes);
						System.out.println("输出流完成一次数据写出");
						len = pis.read(inBytes);
						System.out.println("输入流完成一次数据读出");
						Thread.sleep(1000);
					}
				} catch (IOException | InterruptedException e) {
					e.printStackTrace();
					System.out.println("there are some mistakes");
				}
			}
		});
		input.start();
	}

}

执行上述代码后，控制台能够看到以下打印：

上面的代码模拟了PipedInputStream和PipedOutputStream在同一线程中同时工做的状况，PipedOutputStream每次向buffer中写入1000字节的数据，PipedInputStream每次向buffer中度却500字节的数据，没执行一次读写操做，buffer中都会剩余500字节的数据未读取，由于没有调用PipedInputStream中的void initPipe(int pipeSize)方法，因此buffer默认的大小为1024字节，当执行完第二次读写操做时，缓冲区只剩余24字节的空间，并不足够再一次写入1000字节的数据了，因此此时PipedInputStream的write操做就会阻塞等待有人从buffer中读取并清空缓存区。然而由于读写在同一线程中，PipedInputStream的read操做又在等待PipedOutputStream的write操做完成后，再执行read,从而形成相似死锁的状况，从控制台能够看出，程序卡在了第三次读写时。要解决这个状况也很简单，只要读写操做不在一个线程中就能够了，能够参考第一个例子。

除了上面避免进入死锁的状况还要注意，在进行数据传输的时候，读写线程是否一直保存存活(isAlive)，不管是任何一个线程不在活跃，此时进行读写操做就颇有可能抛出IOEception。

固然你也能够本身写一个工具类，优化一下，好比使用ByteArrayOutputStream的自动扩充缓存的特色来避免管道流由于缓存区空间不够而形成的死锁状况。只有最适合本身需求的才是最棒的，不是吗。

除了管道以外，java中不一样线程之间还有不少的通讯方式，若是须要在线程之间传递字节数据，管道流就是要一个不错的选择，尽管大部分时候通讯之间可能直接传递的是对象而不是简单的字节数据。