PipedInputStream和PipedOutputStream的源码分析和使用方法详细分析

一、PipedOutputStream生产者源码——向PipedInputStream消费者中的缓冲区byte[]数组写入字节数据的输出Stream生产者package java.io; import java.io.*; public class PipedOutputStream extends OutputStream { //与这个PipedOutputStream生产者相关联的 PipedInputStream 消费者 private PipedInputStream sink; //构造函数 public PipedOutputStream(PipedInputStream snk) throws IOException { connect(snk);//调用connect()函数来改变PipedInputStream 消费者中一些变量的值 } //构造函数 public PipedOutputStream() { } //线程同步函数用来改变将要关联的PipedInputStream 消费者中一些变量的值 public synchronized void connect(PipedInputStream snk) throws IOException { if (snk null) { throw new NullPointerException();//如果将要关联的PipedInputStream 消费者为null抛出NullPointerException } else if (sink ! null || snk.connected) { //如果与这个PipedOutputStream生产者相关联的 PipedInputStream 消费者!null或者将要关联的PipedInputStream 消费者的boolean connected变量为true则抛出IOException throw new IOException(Already connected); } sink snk;//将这个PipedOutputStream生产者与这个PipedInputStream 消费者相关联 snk.in -1;//改变PipedInputStream 消费者中的变量int in-1 snk.out 0;//改变PipedInputStream 消费者中的变量int out0 snk.connected true;//改变PipedInputStream 消费者中的变量boolean connectedtrue } //向与这个PipedOutputStream生产者相关联的 PipedInputStream 消费者的缓冲区byte[]数组写入1个字节 public void write(int b) throws IOException { if (sink null) { //如果与这个PipedOutputStream生产者相关联的 PipedInputStream 消费者 null抛出IOException throw new IOException(Pipe not connected); } sink.receive(b);//最终调用的是这个相关联的 PipedInputStream 消费者的receive(int b)函数 } //向与这个PipedOutputStream生产者相关联的 PipedInputStream 消费者的缓冲区byte[]数组写入byte[]数组b的[off,offlen)左闭右开不包括offlen索引位置的字节 public void write(byte b[], int off, int len) throws IOException { if (sink null) { //如果与这个PipedOutputStream生产者相关联的 PipedInputStream 消费者 null抛出IOException throw new IOException(Pipe not connected); } else if (b null) { throw new NullPointerException();//如果byte[]数组bnull抛出一个NullPointerException } else if ((off 0) || (off b.length) || (len 0) || ((off len) b.length) || ((off len) 0)) {//byte[]数组b的[off,offlen)左闭右开索引位置是否有越界的检查 throw new IndexOutOfBoundsException();//越界的话抛出一个IndexOutOfBoundsException } else if (len 0) { return;//如果len0结束本次函数调用 } sink.receive(b, off, len);//最终调用的是这个相关联的 PipedInputStream 消费者的receive(byte b[], int off, int len)函数 } //线程同步函数使用notifyAll()函数唤醒所有与这个PipedOutputStream生产者相关联的 PipedInputStream 消费者线程这个消费者可以绑定1~多个线程 public synchronized void flush() throws IOException { if (sink ! null) { synchronized (sink) { sink.notifyAll(); } } } //关闭这个PipedOutputStream生产者这个PipedOutputStream生产者不能再向与它相关联的PipedInputStream消费者中的缓冲区byte[]数组写入字节数据 public void close() throws IOException { if (sink ! null) { sink.receivedLast(); } } }二、PipedInputStream消费者源码——从自己的缓冲区byte[]数组读取字节数据的输入Stream消费者package java.io; public class PipedInputStream extends InputStream { //标记符true表示与这个 PipedInputStream 消费者相关联的PipedOutputStream生产者已经关闭反之反之 boolean closedByWriter false; //标记符true表示当前这个 PipedInputStream 消费者已经关闭了反之反之 volatile boolean closedByReader false; //标记符true表示与这个 PipedInputStream 消费者相关联的PipedOutputStream生产者已经持有了这个PipedInputStream 消费者对象或者叫已经连接上了反之反之 boolean connected false; Thread readSide;//当前消费的线程 Thread writeSide;//当前生产者的线程 //默认的PipedInputStream 消费者的缓冲区byte[]数组的长度 private static final int DEFAULT_PIPE_SIZE 1024; //PipedInputStream 消费者的缓冲区byte[]数组 protected byte buffer[]; //缓冲区byte[]数组的写指针 protected int in -1; //缓冲区byte[]数组的读指针 protected int out 0; //构造函数 public PipedInputStream(PipedOutputStream src) throws IOException { this(src, DEFAULT_PIPE_SIZE);//缓冲区byte[]数组的长度使用默认值1024 } //构造函数 public PipedInputStream(PipedOutputStream src, int pipeSize) throws IOException { initPipe(pipeSize);//缓冲区byte[]数组的长度使用指定的长度 //最终还是调用PipedOutputStream生产者的connect()函数并把自身对象this传递进去然后在PipedOutputStream生产者的connect()函数中改变自己的3个变量int in-1、int out0、boolean connectedtrue connect(src); } //构造函数缓冲区byte[]数组的长度使用默认值1024 public PipedInputStream() { initPipe(DEFAULT_PIPE_SIZE); } //构造函数缓冲区byte[]数组的长度使用指定的长度 public PipedInputStream(int pipeSize) { initPipe(pipeSize); } //初始化缓冲区byte[]数组 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); //最终还是调用PipedOutputStream生产者的connect()函数并把自身对象this传递进去然后在PipedOutputStream生产者的connect()函数中改变自己的3个变量int in-1、int out0、boolean connectedtrue } //线程同步函数该函数只被PipedOutputStream生产者的write(int b)函数调用 protected synchronized void receive(int b) throws IOException { checkStateForReceive();//检查PipedInputStream 消费者的状态 writeSide Thread.currentThread();//当前执行该函数的线程就是生产者线程 if (in out) //如果缓冲区byte[]数组的读指针缓冲区byte[]数组的写指针唤醒所有消费者线程自己这个生产者线程调用wait(1000)函数 awaitSpace(); if (in 0) {//缓冲区byte[]数组的写指针0时设置缓冲区byte[]数组的写指针0缓冲区byte[]数组的读指针0 in 0; out 0; } buffer[in] (byte)(b 0xFF);//向缓冲区的写指针位置写入1个字节 if (in buffer.length) { in 0;//如果缓冲区满了设置缓冲区的写指针0 } } //线程同步函数该函数只被PipedOutputStream生产者的write(byte b[], int off, int len)函数调用 synchronized void receive(byte b[], int off, int len) throws IOException { checkStateForReceive();//检查PipedInputStream 消费者的状态 writeSide Thread.currentThread();//当前执行该函数的线程就是生产者线程 int bytesToTransfer len;//生产者线程要写入到缓冲区byte[]数组中的字节总量 while (bytesToTransfer 0) { if (in out) //如果缓冲区byte[]数组的读指针缓冲区byte[]数组的写指针唤醒所有消费者线程自己这个生产者线程调用wait(1000)函数 awaitSpace(); int nextTransferAmount 0;//本次生产者线程要写入到缓冲区byte[]数组中的字节数量 if (out in) { //如果缓冲区的读指针缓冲区的写指针本次要写入到缓冲区byte[]数组中的字节数量缓冲区的长度-缓冲区的写指针 nextTransferAmount buffer.length - in; } else if (in out) { if (in -1) { in out 0; //如果缓冲区的读指针out 缓冲区的写指针in并且缓冲区的写指针in-1先设置缓冲区的读out、写in指针0本次要写入到缓冲区byte[]数组中的字节数量缓冲区的长度 nextTransferAmount buffer.length - in; } else { //如果缓冲区的读指针out 缓冲区的写指针in并且缓冲区的写指针in-1本次要写入到缓冲区byte[]数组中的字节数量读指针out-写指针in nextTransferAmount out - in; } } //本次生产者线程要写入到缓冲区byte[]数组中的字节数量最多为len下次为len-本次写入到缓冲区byte[]数组中的字节数量也就是每次写入的基于len个字节循环递减上一次写入的 if (nextTransferAmount bytesToTransfer) nextTransferAmount bytesToTransfer; assert(nextTransferAmount 0); System.arraycopy(b, off, buffer, in, nextTransferAmount);//向缓冲区byte[]数组的[in,innextTransferAmount)索引位置写入byte[]数组b中[off,offnextTransferAmount)索引位置的字节都是左闭右开。 bytesToTransfer - nextTransferAmount;//每一次都基于len个字节循环递减本次写入到缓冲区byte[]数组中的字节数量nextTransferAmount off nextTransferAmount;//将下次要从byte[]数组b中取字节的起始索引的位置偏移量本次写入到缓冲区byte[]数组中的字节数量nextTransferAmount in nextTransferAmount;//将缓冲区的写指针in本次写入到缓冲区byte[]数组中的字节数量nextTransferAmount if (in buffer.length) { in 0;//如果缓冲区的写指针in 缓冲区byte[]数组的长度设置缓冲区的写指针in0 } } } //检查PipedInputStream 消费者的状态 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); } } //如果缓冲区byte[]数组的读指针缓冲区byte[]数组的写指针唤醒所有消费者线程自己这个生产者线程调用wait(1000)函数 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(); } } } //关闭与这个 PipedInputStream 消费者相关联的PipedOutputStream生产者 synchronized void receivedLast() { closedByWriter true; notifyAll();//唤醒所有消费者线程 } //线程同步函数消费者线程每次从缓冲区byte[]数组中读取1个字节 public synchronized int read() throws IOException { if (!connected) {//检查标记符connected如果为false抛出IOException throw new IOException(Pipe not connected); } else if (closedByReader) {//检查标记符closedByReader如果为true抛出IOException throw new IOException(Pipe closed); } else if (writeSide ! null !writeSide.isAlive() !closedByWriter (in 0)) { //检查当前这个PipedInputStream 消费者对象中引用的生产者线程和生产者线程的状态如果和标记符closedByWriter还有缓冲区byte[]数组的写指针in不能对应的话抛出一个IOException throw new IOException(Write end dead); } readSide Thread.currentThread();//当前执行该函数的线程就是消费者线程 int trials 2;//这是一个多次检测的策略变量防止生产者线程没有关闭了与这个 PipedInputStream 消费者相关联的PipedOutputStream生产者时便抛出IOException //in-1的情况有种 //①、生产者线程还没有向缓冲区byte[]数组中写任何字节 //②、消费者线程从缓冲区byte[]数组中读完字节byte数据以后读指针out写指针in那么当前消费者线程会设置写指针in-1 //③、消费者线程执行PipedInputStream 的close()函数后关闭了这个 PipedInputStream 消费者 while (in 0) { if (closedByWriter) { /* closed by writer, return EOF */ return -1; } if ((writeSide ! null) (!writeSide.isAlive()) (--trials 0)) { //多个消费者线程从缓冲区byte[]数组中读的时候并且前一个消费者线程已经把缓冲区byte[]数组中写入的字节读完了并且前一个线程设置了写指针in-1生产者线程也关闭了与这个 PipedInputStream 消费者相关联的PipedOutputStream生产者时抛出一个IOException 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;//获取缓冲区byte[]数组中读指针out索引位置的字节,并且将读指针out1 if (out buffer.length) { out 0;//如果读指针out缓冲区byte[]数组的长度设置读指针out0 } if (in out) { /* now empty */ in -1;//如果消费者线程从缓冲区byte[]数组中读完字节byte数据以后读指针out写指针in那么当前消费者线程会设置写指针in-1 } return ret; } //线程同步函数如果缓冲区byte[]数组中有足够多的字节的话数量len消费者线程每次从缓冲区byte[]数组中读取len个字节放到byte[]数组b的[off, offlen)索引位置左闭右开不包括offlen //如果缓冲区byte[]数组中字节的数量len个比如有in写指针-out读指针个消费者线程每次从缓冲区byte[]数组中读取in-out个字节放到byte[]数组b的[off, offin-out)索引位置左闭右开不包括offin-out 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) {//byte[]数组b的[off,offlen)左闭右开索引位置是否有越界的检查 throw new IndexOutOfBoundsException();//越界的话抛出一个IndexOutOfBoundsException } else if (len 0) { return 0;//如果len0返回0 } /* possibly wait on the first character */ int c read();//先调用read()函数试探性从缓冲区byte[]数组中读1个字节 if (c 0) { return -1;//如果试探性的从缓冲区byte[]数组中都读不到1个字节返回-1 } b[off] (byte) c;//把试探性从缓冲区byte[]数组中读到的第1个字节放到byte[]数组b的off索引位置 int rlen 1;//累计从缓冲区byte[]数组中读到的所有字节数量 while ((in 0) (len 1)) { int available;//本次执行System.arraycopy()函数可以从缓冲区byte[]数组中读到byte[]数组b中的字节数量 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)) {//减掉试探性从缓冲区byte[]数组中读到的第1个字节 available len - 1; } System.arraycopy(buffer, out, b, off rlen, available); out available;//读指针outSystem.arraycopy()函数从缓冲区byte[]数组中读到byte[]数组b中的字节数量 rlen available;//累计从缓冲区byte[]数组中读到的所有字节数量 System.arraycopy()函数从缓冲区byte[]数组中读到byte[]数组b中的字节数量 len - available;//len - System.arraycopy()函数从缓冲区byte[]数组中读到byte[]数组b中的字节数量 if (out buffer.length) { out 0;//如果读指针out缓冲区byte[]数组的长度设置读指针out0 } if (in out) { /* now empty */ in -1;//如果消费者线程从缓冲区byte[]数组中读完字节byte数据以后读指针out写指针in那么当前消费者线程会设置写指针in-1 } } return rlen;//返回累计从缓冲区byte[]数组中读到的所有字节数量 } //线程同步函数返回缓冲区byte[]数组中可以被消费者线程读取的字节数量 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; } //关闭这个 PipedInputStream 消费者其实就是设置标记符closedByReadertrue 设置写指针in-1 public void close() throws IOException { closedByReader true; synchronized (this) { in -1; } } }三、1个线程向PipedOutputStream生产者写字节数据1个线程从PipedInputStream消费者读取字节数据的过程3.1、非循环直接写和非循环直接读package com.chelong.StreamAndReader; import java.io.IOException; import java.io.PipedInputStream; import java.io.PipedOutputStream; public class PipedTest { public static void main(String[] args) throws IOException { final PipedOutputStream output new PipedOutputStream(); final PipedInputStream input new PipedInputStream(output); Thread thread1 new Thread(new Runnable() { Override public void run() { try { output.write(Hello world, pipe!.getBytes());//write()函数是阻塞的 } catch (IOException e) { } } }); Thread thread2 new Thread(new Runnable() { Override public void run() { try { int data -1; while ((data input.read()) ! -1) {//read()函数是阻塞的 System.out.print((char) data); } } catch (IOException e) { } } }); thread1.start(); thread2.start(); } }程序运行结果如下所示main线程构造PipedOutputStream生产者和PipedInputStream消费者的过程如下向PipedOutputStream生产者写字节数据的生产者线程的执行过程如下从PipedInputStream消费者读取字节数据的消费者线程的执行过程如下3.1.1、非循环直接写和非循环直接读时1个生产者线程和1个消费者线程处理数据的过程Java 语言定义了 6 种线程状态, 在任意一个时间点, 一个线程只能有且只有其中的一种状态, 这 6 种状态分别如下这 6 种线程状态的简单介绍如下所示JVM运行时内存结构主要包含了五个部分程序计数器 PC寄存器、 JVM栈、Native方法栈、堆、 方法区。如下图所示图中红色部分是线程私有区域进入这个区域的数据不会出现线程竞争的关系。而绿色区域中的数据则被所有线程共享其中Java堆中存放的是大量对象方法区中存放class信息、常量、静态变量等数据。每个线程的线程栈中会存放函数方法的描述符成员本地变量等函数方法在线程栈中会通过压栈和弹栈来执行除了8种byte、short、int、long、float、double、boolean、char基本的数据类型存储在线程栈中以外其余的引用数据类型对象都存储在堆中然后通过引用将堆中的对象和线程栈中的变量关联起来也可以叫线程栈中的引用指向堆中的对象。那么当使用者执行3.1中的代码时1个生产者线程和1个消费者线程处理数据的过程如下①、main线程初始化一个缓冲区byte[]数组长度为1024默认值然后生产者线程通过不断的压栈来完成函数之间的调用最终执行PipedInputStream.class::receive(byte b[], int off, int len)函数来对缓冲区byte[]数组进行填充如下所示②、当生产者线程填充完缓冲区之后写指针变量int in17读指针变量int out0Thread writeSide 当前这个生产者线程Thread对象生产者线程会把自己线程栈中修改的变量最终刷新到堆中PipedInputStream对象中以确保其它消费者线程的线程栈从堆中读取这3个变量时这3个变量已经为修改后的值如下所示③、消费者线程读缓冲区byte[]数组的过程中会不断地执行out读指针以读取缓冲区byte[]数组中的可用字节并返回直到out读指针in写指针修改in写指针-1并且每次同步执行PipedInputStream.class::read()函数时都会更新Thread readSide 当前这个消费者线程Thread对象消费者线程也会把自己线程栈中修改的变量最终刷新到堆中PipedInputStream对象中以确保其它消费者线程的线程栈从堆中读取这3个变量时这3个变量已经为修改后的值如下所示④、更新in写指针-1后消费者线程再次同步执行PipedInputStream.class::read()函数时如果PipedInputStream::boolean closedByWriter变量为true则会返回-13.2、加锁循环写和非加锁循环读到byte[]数组b中再处理package com.chelong.pipe; import java.io.IOException; import java.io.PipedInputStream; import java.io.PipedOutputStream; public class PipeForTransferInThread { public static void main(String[] args) throws IOException, InterruptedException { final PipedOutputStream output new PipedOutputStream(); final PipedInputStream input new PipedInputStream(output); //生产者线程 Thread producer new Thread(new Runnable() { Override public void run() { for (int i 0; i 3; i) { synchronized (input) { try { // input.wait(); output.write(Hello world, pipe!.getBytes()); input.wait();//释放锁并无限等待直到消费者线程consumer 执行notifyAll()函数来唤醒当前阻塞 } catch (Exception e) { e.printStackTrace(); } } } } },生产者线程); //消费者线程 Thread consumer new Thread(new Runnable() { Override public void run() { try { byte[] b new byte[1024];//1KB int readBytes -1; long lastTime System.currentTimeMillis(); while ((readBytes input.read(b, 0, b.length)) ! -1) { long curTime System.currentTimeMillis(); System.out.print(Thread.currentThread().getName()本次读取花费时间 (curTime - lastTime) ms读到的数据是); lastTime curTime; for (int i 0; i readBytes; i) { System.out.print((char) b[i]);//模拟处理字节数据 } System.out.println(); } } catch (IOException e) { e.printStackTrace(); } } },消费者线程); producer.start();//生产者线程启动 consumer.start();//消费者线程启动 } }程序运行结果如下所示main线程构造PipedOutputStream生产者和PipedInputStream消费者的过程可以参考3.1向PipedOutputStream生产者写字节数据的生产者线程的执行过程可以参考3.1从PipedInputStream消费者读取字节数据的消费者线程的执行过程如下3.2.1、加锁循环写和非加锁循环读到byte[]数组b中再处理时1个生产者线程和1个消费者线程处理数据的过程标题3.2中的代码的整个执行过程如下①、main线程初始化一个缓冲区byte[]数组长度为1024默认值如下所示②、然后生产者线程通过不断的压栈来完成函数之间的调用最终执行PipedInputStream.class::receive(byte b[], int off, int len)函数来对缓冲区byte[]数组进行填充并且先在自己的线程栈中更新in写指针17out读指针0writeSide当前这个生产者线程Thread对象 如下所示当生产者线程对缓冲区byte[]数组填充完成之后再执行标题3.2中的代码input.wait();这行代码会释放锁并让生产者线程进入无限等待直到消费者线程consumer执行notifyAll()函数来唤醒当前这个生产者线程。在这之前生产者线程会将自己线程栈中的in写指针17out读指针0writeSide当前这个生产者线程这3个变量更新到主内存也就是堆中的PipedInputStream对象中。③、消费者线程读缓冲区byte[]数组的过程也是通过不断的压栈来完成函数之间的调用最终执行PipedInputStream::read()函数试探性的读取1个字节和PipedInputStream::read(byte b[], int off, int len)函数读取剩余其它的字节将步骤②中生产者线程写入到缓冲区byte[]数组中的17个字节读取出来附言最终消费者线程也会将自己线程栈中的in写指针 -1out读指针 17writeSide当前这个消费者线程这3个变量更新到主内存也就是堆中的PipedInputStream对象中。因此本次消费者线程从缓冲区byte[]数组中读数据的过程中没有执行read()函数中的wait(1000)这一行代码如下所以本次消费者线程从缓冲区byte[]数组中读取数据到消费者线程中自己创建的byte[]数组中时只花费了0ms接下来当消费者线程将步骤②中生产者线程写入到缓冲区byte[]数组中的17个字节读取出来以后通过System.arraycopy()函数复制到了消费者线程中自己创建的byte[]数组中消费者线程会遍历从缓冲区读到的这个byte[]数组来处理这些数据如下所示标题3.2中的代码片段//标题3.2中的代码片段 for (int i 0; i readBytes; i) { System.out.print((char) b[i]);//模拟处理字节数据 }然后当消费者线程再次执行//标题3.2中的代码片段 input.read(b, 0, b.length)从缓冲区byte[]数组中读数据到自己创建的byte[]数组中时由于此时in写指针-1并且当下图中的其它5个条件都不成立时唤醒执行了input.wait()的生产者线程然后当前这个正在从缓冲区(byte数组)中读数据的消费者线程执行wait 1000ms 如下④、当生产者线程被消费者线程执行的notifyAll();唤醒之后会再次通过不断的压栈来完成函数之间的调用再次执行PipedInputStream.class::receive(byte b[], int off, int len)函数来对缓冲区byte[]数组进行填充并且先在自己的线程栈中先更新in写指针17out读指针0writeSide当前这个生产者线程Thread对象 如下所示当生产者线程对缓冲区byte[]数组填充完成之后再执行标题3.2中的代码input.wait();这行代码会释放锁并让生产者线程进入无限等待直到消费者线程consumer执行notifyAll()函数来唤醒当前这个生产者线程。在这之前生产者线程会将自己线程栈中的in写指针17out读指针0writeSide当前这个生产者线程这3个变量更新到主内存也就是堆中的PipedInputStream对象中。如下所示⑤、消费者线程在第③步执行了wait(1000);在等待了1000ms之后消费者线程会自动唤醒继续执行此时自己线程栈中的in写指针 -1out读指针 17已经被第④步中的生产者线程修改为in写指针17out读指针0生产者线程不会直接修改消费者线程栈中的变量生产者线程会先将自己线程栈中in写指针out读指针变量的值修改到主内存中然后消费者线程会自己将主内存中的这2个变量值刷新到消费者自己的线程栈中如下所示