频道栏目
首页 > 程序开发 > 软件开发 > Java > 正文
Java 7之传统I/O - PipedInputStream和PipedOutputStream
2014-02-18 11:28:40      个评论    来源:Java 7之传统I/O - PipedInputStream和PipedOutputStream  
收藏   我要投稿
       PipedInputStream类与PipedOutputStream类用于在应用程序中创建管道通信.一个PipedInputStream实例对象必须和一个PipedOutputStream实例对象进行连接而产生一个通信管道.PipedOutputStream可以向管道中写入数据,PipedIntputStream可以读取PipedOutputStream向管道中写入的数据.这两个类主要用来完成线程之间的通信.一个线程的PipedInputStream对象能够从另外一个线程的PipedOutputStream对象中读取数据.如下图所示。

\

PipedInputStream和PipedOutputStream的实现原理类似于"生产者-消费者"原理,PipedOutputStream是生产者,PipedInputStream是消费者,在PipedInputStream中有一个buffer字节数组,默认大小为1024,作为缓冲区,存放"生产者"生产出来的东西.还有两个变量in和out。in是用来记录"生产者"生产了多少,out是用来记录"消费者"消费了多少,in为-1表示消费完了,in==out表示生产满了.当消费者没东西可消费的时候,也就是当in为-1的时候,消费者会一直等待,直到有东西可消费.


1、创建连接与初始化


在两者的构造函数中,都相互提供了连接的构造方法,分别用于接收对方的管道实例,然后调用各自的connect()方法进行连接,如PipedInputStream:

PipedInputStream

    private static final int DEFAULT_PIPE_SIZE = 1024;
    protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE;
    // 省略部分构造函数
    public PipedInputStream(PipedOutputStream src, int pipeSize)  throws IOException {
         initPipe(pipeSize);
         connect(src);
    }
    private void initPipe(int pipeSize) {  // 初始化buffer的大小,pipeSize可以通过构造方法指定,也可以使用默认的PIPE_SIZE的大小
         if (pipeSize <= 0) {
            throw new IllegalArgumentException("Pipe Size <= 0");
         }
         buffer = new byte[pipeSize];      // 初始化缓冲区的大小
    }
    public void connect(PipedOutputStream src) throws IOException {
        src.connect(this);                //  连接输入管道
    }

看一下PipedOutputStream的构造函数,如下:

PipedOutputStream

    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;         // buffer数组中无数据
        snk.out = 0;         // 取出的数据为0
        snk.connected = true;// 表示连接成功 
    }

可以看到,相互之间会调用connect()方法来连接,其效果是一样的。


2、数据的写入与读取


连接成功就,就可以进行数据的写入与读出操作了,在PipedOutputStream中的write()写法如下:

PipedOutputStream

   public void write(int b)  throws IOException {
        if (sink == null) {
            throw new IOException("Pipe not connected");
        }
        sink.receive(b);             // 调用receive方法进行数据的输出
    }
    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);  
    }

方法在写入到byte[]数组缓存区数据后,就会调用PipedInputStream中的receive方法。输出管道中的receive()方法如下:

PipedInputStream

    protected synchronized void receive(int b) throws IOException { // 只会在PipedOutputStream类的write()中调用
        checkStateForReceive();
        writeSide = Thread.currentThread();
        if (in == out)         // in==out表示buffer数组已满
            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 { // 将下标off开始的len个数组数据写入到输出管道中
        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);                          // nextTransferAmount<=0,则终止程序的执行
            System.arraycopy(b, off, buffer, in, nextTransferAmount);// 拷贝数组中数据到缓冲区
            bytesToTransfer -= nextTransferAmount;
            off += nextTransferAmount;
            in += nextTransferAmount;
            if (in >= buffer.length) {
                in = 0;                
            }
        }
    }

输入管理通过如上的对应方法接收到数据并保存到输入缓冲区后,下面就可以使用read()方法读出这些数据了。

看一下awaitSpace()方法的实现源代码:

PipedInputStream

     /*
      *  若写入管道的数据正好全部被读取完(例如,管道缓冲满),则执行awaitSpace()操作;
      *  让读取管道的线程管道产生读取数据请求,从而才能继续的向“管道”中写入数据
      */
     private void awaitSpace() throws IOException {
         /*
          * 如果管道中被读取的数据,等于写入管道的数据时,
          * 则每隔1000ms检查“管道状态”,并唤醒管道操作:若有读取管道数据线程被阻塞,则唤醒该线程
          */
         while (in == out) {
             checkStateForReceive();
             /* full: kick any waiting readers */
             notifyAll();
             try {
                 wait(1000);
             } catch (InterruptedException ex) {
                 throw new java.io.InterruptedIOException();
             }
         }
     }



PipedInputStream类中的read()方法源代码如下:

PipedInputStream

    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) {  // 如果in<0(表示管道中无数据)且closedByWriter为true(表示输入管道已经关闭)则直接返回-1
                return -1;
            }
            if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) {
                throw new IOException("Pipe broken");
            }
            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) {
                in = -1;
            }
        }
        return rlen;
    }




3、刷新与关闭管道


来看一下管道输出流中的刷新和关闭方法,源代码如下:

PipedOutputStream

    // 刷回管道输出流
    public synchronized void flush() throws IOException {
        if (sink != null) {
            synchronized (sink) {
            	/*
            	 * 调用管道输入流的notifyAll(),通知管道输入流放弃对当前资源的占有,
            	 * 让其它的等待线程(等待读取管道输出流的线程)读取管道输出流的值。
            	 */
                sink.notifyAll(); 
            }
        }
    }
    // 关闭管道输出流
    public void close()  throws IOException {
        if (sink != null) {
            sink.receivedLast();// 通知管道输入流,输出管理已经关闭
        }
    }

看一下receivedLast()方法,如下:

PipedInputStream

 synchronized void receivedLast() {
        closedByWriter = true;  // 输出管道标志为true,表示关闭
        notifyAll();            // 唤醒所有的等待线程
 }
通知所有的等待线程,最后的数据已经全部到达。

PipedInputStream
public void close()  throws IOException {  // 关闭管道输出流
        closedByReader = true;
        synchronized (this) {
            in = -1;                      // 清空缓冲区数据 
        }
    }


下面来具体举一个例子,如下:

public class test04 {
    public static void main(String [] args) {  
        Sender sender = new Sender();  
        Receiver receiver = new Receiver();  
          
        PipedOutputStream outStream = sender.getOutStream();  
        PipedInputStream inStream = receiver.getInStream();  
        try {  
            //inStream.connect(outStream); // 与下一句一样  
            outStream.connect(inStream);  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
        sender.start();  
        receiver.start();  
    }  
}  
  
class Sender extends Thread {  
    private PipedOutputStream outStream = new PipedOutputStream();  
    public PipedOutputStream getOutStream() {  
        return outStream;  
    }  
    public void run() {  
        String info = "hello, receiver";  
        try {  
            outStream.write(info.getBytes());  
            outStream.close();  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}  
  
class Receiver extends Thread {  
    private PipedInputStream inStream = new PipedInputStream();  
    public PipedInputStream getInStream() {  
        return inStream;  
    }  
    public void run() {  
        byte[] buf = new byte[1024];  
        try {  
            int len = inStream.read(buf);  
            System.out.println("receive message from sender : " + new String(buf, 0, len));  
            inStream.close();  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }     
}  

最后运行后输出的结果如下:receive message from sender : hello, receiver




参考文献:

1、https://www.cnblogs.com/lich/archive/2011/12/11/2283928.html

2、https://www.cnblogs.com/skywang12345/p/io_04.html





点击复制链接 与好友分享!回本站首页
相关TAG标签 传统
上一篇:java按指定编码写入和读取文件内容的类
下一篇:字典序问题的解决方案
相关文章
图文推荐
点击排行

关于我们 | 联系我们 | 广告服务 | 投资合作 | 版权申明 | 在线帮助 | 网站地图 | 作品发布 | Vip技术培训 | 举报中心

版权所有: 红黑联盟--致力于做实用的IT技术学习网站