PipedInputStream/PipedOutputStream原理

Posted 2020-07-01 邱明成

PipedInputStream类与PipedOutputStream类用于在应用程序中创建管道通信.一个PipedInputStream实例对象必须和一个PipedOutputStream实例对象进行连接而产生一个通信管道.PipedOutputStream可以向管道中写入数据,PipedIntputStream可以读取PipedOutputStream向管道中写入的数据.这两个类主要用来完成线程之间的通信.一个线程的PipedInputStream对象能够从另外一个线程的PipedOutputStream对象中读取数据. 

ps:使用这组I/O流必须在多线程环境下. 

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

        因为生产和消费的方法都是synchronized的，所以肯定是生产者先生产出一定数量的东西,消费者才可以开始消费,所以在生产的时候发现in==out,那一定是满了,同理,在消费的时候发现in==out,那一定是消费完了,因为生产的东西永远要比消费来得早,消费者最多可以消费和生产的数量相等的东西,而不会超出. 

        好了,介绍完之后,看看SUN高手是怎么实现这些功能的.由于buffer(存放产品的通道)这个关键变量在PipedInputStream消费者这个类中,所以要想对buffer操作,只能通过PipedInputStream来操作,因此将产品放入通道的操作是在PipedInputStream中. 

存放产品的行为: 

Java代码  

1. protected synchronized void receive(int b) throws IOException {// 这里好像有些问题,因为这个方法是在PipedOutputStream类中调用的,而这个方法是protected的,下面另一个receive方法就不是protected,可能是我的源码有些问题,也请大家帮我看看  
2.     checkStateForReceive();// 检测通道是否连接,准备好接收产品  
3.     writeSide = Thread.currentThread();// 当前线程是生产者  
4.     if (in == out)  
5.         awaitSpace();// 发现通道满了,没地方放东西啦,等吧~~  
6.     if (in < 0) {// in<0,表示通道是空的,将生产和消费的位置都置成第一个位置  
7.         in = 0;  
8.         out = 0;  
9.     }  
10.     buffer[in++] = (byte) (b & 0xFF);  
11.     if (in >= buffer.length) {// 如果生产位置到达了通道的末尾,为了循环利用通道,将in置成0  
12.         in = 0;  
13.     }  
14. }  
15.   
16. synchronized void receive(byte b[], int off, int len) throws IOException {// 看,这个方法不是protected的!  
17.     checkStateForReceive();  
18.     writeSide = Thread.currentThread();  
19.     int bytesToTransfer = len;// 需要接收多少产品的数量  
20.     while (bytesToTransfer > 0) {  
21.         if (in == out)  
22.             awaitSpace();  
23.         int nextTransferAmount = 0;// 本次实际可以接收的数量  
24.         if (out < in) {  
25.             nextTransferAmount = buffer.length - in;// 如果消费的当前位置<生产的当前位置,则还可以再生产buffer.length-in这么多  
26.         } else if (in < out) {  
27.             if (in == -1) {  
28.                 in = out = 0;// 如果已经消费完,则将in,out置成0,从头开始接收  
29.                 nextTransferAmount = buffer.length - in;  
30.             } else {  
31.                 nextTransferAmount = out - in;// 如果消费的当前位置>生产的当前位置,而且还没消费完,那么至少还可以再生产out-in这么多,注意,这种情况是因为通道被重复利用而产生的!  
32.             }  
33.         }  
34.         if (nextTransferAmount > bytesToTransfer)// 如果本次实际可以接收的数量要大于当前传过来的数量,  
35.             nextTransferAmount = bytesToTransfer;// 那么本次实际只能接收当前传过来的这么多了  
36.         assert (nextTransferAmount > 0);  
37.         System.arraycopy(b, off, buffer, in, nextTransferAmount);// 把本次实际接收的数量放进通道  
38.         bytesToTransfer -= nextTransferAmount;// 算出还剩多少需要放进通道  
39.         off += nextTransferAmount;  
40.         in += nextTransferAmount;  
41.         if (in >= buffer.length) {// 到末尾了,该从头开始了  
42.             in = 0;  
43.         }  
44.     }  
45. }  

消费产品的行为: 

Java代码  

1. public synchronized int read() throws IOException {// 消费单个产品  
2.     if (!connected) {  
3.         throw new IOException("Pipe not connected");  
4.     } else if (closedByReader) {  
5.         throw new IOException("Pipe closed");  
6.     } else if (writeSide != null && !writeSide.isAlive() && !closedByWriter  
7.             && (in < 0)) {  
8.         throw new IOException("Write end dead");  
9.     }  
10.   
11.     readSide = Thread.currentThread();  
12.     int trials = 2;  
13.     while (in < 0) {// in<0,表示通道是空的,等待生产者生产  
14.         if (closedByWriter) {  
15.             /**//* closed by writer, return EOF */  
16.             return -1;// 返回-1表示生产者已经不再生产产品了,closedByWriter为true表示是由生产者将通道关闭的  
17.         }  
18.         if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) {  
19.             throw new IOException("Pipe broken");  
20.         }  
21.         /**//* might be a writer waiting */  
22.         notifyAll();  
23.         try {  
24.             wait(1000);  
25.         } catch (InterruptedException ex) {  
26.             throw new java.io.InterruptedIOException();  
27.         }  
28.     }  
29.     int ret = buffer[out++] & 0xFF;  
30.     if (out >= buffer.length) {  
31.         out = 0;// 如果消费到通道的末尾了,从通道头开始继续循环消费  
32.     }  
33.     if (in == out) {  
34.         /**//* now empty */  
35.         in = -1;// 消费的位置和生产的位置重合了,表示消费完了,需要生产者生产,in置为-1  
36.     }  
37.     return ret;  
38. }  
39.   
40. public synchronized int read(byte b[], int off, int len) throws IOException {  
41.     if (b == null) {  
42.         throw new NullPointerException();  
43.     } else if ((off < 0) || (off > b.length) || (len < 0)  
44.             || ((off + len) > b.length) || ((off + len) < 0)) {  
45.         throw new IndexOutOfBoundsException();  
46.     } else if (len == 0) {  
47.         return 0;  
48.     }  
49.   
50.     /**//* possibly wait on the first character */  
51.     int c = read();// 利用消费单个产品来检测通道是否连接,并且通道中是否有东西可消费  
52.     if (c < 0) {  
53.         return -1;// 返回-1表示生产者生产完了,消费者也消费完了,消费者可以关闭通道了  
54.     }  
55.     b[off] = (byte) c;  
56.     int rlen = 1;  
57.   
58.     // 这里没有采用receive(byte [], int ,  
59.     // int)方法中System.arrayCopy()的方法,其实用System.arrayCopy()的方法也可以实现  
60.     /**//* 
61.      * 这是用System.arrayCopy()实现的方法 int bytesToConsume = len - 1; while 
62.      * (bytesToConsume > 0 && in >= 0) { int nextConsumeAmount = 0; if (out < 
63.      * in) { nextConsumeAmount = in - out; // System.arraycopy(buffer, out, 
64.      * b, off, nextConsumeAmount); } else if (in < out) { nextConsumeAmount = 
65.      * buffer.length - out; } 
66.      *  
67.      * if (nextConsumeAmount > bytesToConsume) nextConsumeAmount = 
68.      * bytesToConsume; assert (nextConsumeAmount > 0); 
69.      * System.arraycopy(buffer, out, b, off, nextConsumeAmount); 
70.      * bytesToConsume -= nextConsumeAmount; off += nextConsumeAmount; out += 
71.      * nextConsumeAmount; rlen += nextConsumeAmount; if (out >= 
72.      * buffer.length) { out = 0; } if(in == out) { in = -1; } } 
73.      */  
74.   
75.     while ((in >= 0) && (--len > 0)) {  
76.         b[off + rlen] = buffer[out++];  
77.         rlen++;  
78.         if (out >= buffer.length) {  
79.             out = 0;  
80.         }  
81.         if (in == out) {  
82.             /**//* now empty */  
83.             in = -1;// in==out,表示满了,将in置成-1  
84.         }  
85.     }  
86.     return rlen;  
87. }