java中PipedInputStream和PipedOutputStream分别是管道输入流和管道输出流,它的作用是让多线程可以通过管道进行线程间的通讯,在使用管道通信时,必须将PipedInputStream和PipedOutputStream配套使用。大致的流程是:当在线程A中向PipedOutputStream中写入数据,会自动发送到与PipedOutputStream对应的PipedInputStream中,进而存储在PipedInputStream的缓冲区中,此时线程B通过读取PipedInputStream中的数据,实现线程的通信。
PipedInputStream类的主要函数有:
public PipedInputStream(PipedOutputStream src) public PipedInputStream(PipedOutputStream src, int pipeSize) public PipedInputStream() public PipedInputStream(int pipeSize) public void connect(PipedOutputStream src) protected synchronized void receive(int b) synchronized void receive(byte b[], int off, int len) public synchronized int read() public synchronized int read(byte b[], int off, int len) public synchronized int available()
PipedOutputStream类的主要函数有:
public PipedOutputStream(PipedInputStream snk) public PipedOutputStream() public synchronized void connect(PipedInputStream snk) public void write(int b) public void write(byte b[], int off, int len)基于PipedInputStream和PipedOutputStream线程通信示例:
Send类:
public class Send extends Thread { private PipedOutputStream outputStream=new PipedOutputStream(); public PipedOutputStream getOutputStream() { return outputStream; } @Override public void run() { writeMessage(); } public void writeMessage() { String string="hello pipedstream"; try { outputStream.write(string.getBytes()); outputStream.close(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
Receiver类:
public class Receiver extends Thread{ private PipedInputStream inputStream=new PipedInputStream(); public PipedInputStream getInputStream() { return inputStream; } @Override public void run() { readMessage(); } public void readMessage() { byte[] buf=new byte[2048]; int len; try { len = inputStream.read(buf); System.out.println(new String(buf,0,len)); inputStream.close(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
Main函数:
public class Hello { public static void main(String [] args) { Send s1=new Send(); Receiver r1=new Receiver(); PipedInputStream inputStream=r1.getInputStream(); PipedOutputStream outputStream=s1.getOutputStream(); try { inputStream.connect(outputStream); s1.start(); r1.start(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }
运行结果:
hello pipedstream
结果分析:
(1)inputStream.connect(outputStream);将管道输入流和输出流关联起来,代码换成outputStream.connect(inputStream)是同样的效果
(2)
当调用PipedOutputStream的write()方法时,其实是调用的PipedInputStream的receiver()方法。
基于JDK8的PipedOutputStream的源码:
public class PipedOutputStream extends OutputStream { (but it may be a long time until the next GC). */ 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); } //想输入流写b,起始为off,长度为len 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(); } } }
基于JDK8的PipedInputStream的源码:
public class PipedInputStream extends InputStream { boolean closedByWriter = false; volatile boolean closedByReader = false; boolean connected = false; Thread readSide;//读线程 Thread writeSide;//写线程 private static final int DEFAULT_PIPE_SIZE = 1024;//默认管道大小 protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE; //字节数组,循环数组,放置数据 protected byte buffer[]; //写的标志位为-1 protected int in = -1; //读的标志位为0,当in==out表示为空, 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; } //从输入流中读取数据到b。其实为off,大小为len 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; } } }
转载:https://blog.csdn.net/qq924862077/article/details/52692870