PipedInputStream类
Posted mgblogs
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了PipedInputStream类相关的知识,希望对你有一定的参考价值。
1 package java.io; 2 3 public class PipedInputStream extends InputStream 4 // “管道输出流”是否关闭的标记 5 boolean closedByWriter = false; 6 // “管道输入流”是否关闭的标记 7 volatile boolean closedByReader = false; 8 // “管道输入流”与“管道输出流”是否连接的标记 9 // 它在PipedOutputStream的connect()连接函数中被设置为true 10 boolean connected = false; 11 12 Thread readSide; // 读取“管道”数据的线程 13 Thread writeSide; // 向“管道”写入数据的线程 14 15 // “管道”的默认大小 16 private static final int DEFAULT_PIPE_SIZE = 1024; 17 18 protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE; 19 20 // 缓冲区 21 protected byte buffer[]; 22 23 //MG注 in为写指针 下一个写入字节的位置。in==out代表满,说明“写入的数据”全部被读取了或者是代表in多走了length。 24 protected int in = -1; 25 //MG注 out为读指针 下一个读取字节的位置。in==out代表满,说明“写入的数据”全部被读取了或者是代表in多走了length。 26 protected int out = 0; 27 28 // 构造函数:指定与“管道输入流”关联的“管道输出流” 29 public PipedInputStream(PipedOutputStream src) throws IOException 30 this(src, DEFAULT_PIPE_SIZE); 31 32 33 // 构造函数:指定与“管道输入流”关联的“管道输出流”,以及“缓冲区大小” 34 public PipedInputStream(PipedOutputStream src, int pipeSize) 35 throws IOException 36 initPipe(pipeSize); 37 connect(src); 38 39 40 // 构造函数:默认缓冲区大小是1024字节 41 public PipedInputStream() 42 initPipe(DEFAULT_PIPE_SIZE); 43 44 45 // 构造函数:指定缓冲区大小是pipeSize 46 public PipedInputStream(int pipeSize) 47 initPipe(pipeSize); 48 49 50 // 初始化“管道”:新建缓冲区大小 51 private void initPipe(int pipeSize) 52 if (pipeSize <= 0) 53 throw new IllegalArgumentException("Pipe Size <= 0"); 54 55 buffer = new byte[pipeSize]; 56 57 58 // 将“管道输入流”和“管道输出流”绑定。 59 // 实际上,这里调用的是PipedOutputStream的connect()函数 60 public void connect(PipedOutputStream src) throws IOException 61 src.connect(this); 62 63 64 // 接收int类型的数据b。 65 // 它只会在PipedOutputStream的write(int b)中会被调用 66 protected synchronized void receive(int b) throws IOException 67 // 检查管道状态 68 checkStateForReceive(); 69 // 获取“写入管道”的线程 70 writeSide = Thread.currentThread(); 71 //MG注 此处只有in比out多走了length的长度才会进入 72 if (in == out) 73 awaitSpace(); 74 if (in < 0) 75 in = 0; 76 out = 0; 77 78 // 将b保存到缓冲区 79 buffer[in++] = (byte)(b & 0xFF); 80 if (in >= buffer.length) 81 in = 0; 82 83 84 85 // 接收字节数组b。 86 synchronized void receive(byte b[], int off, int len) throws IOException 87 // 检查管道状态 88 checkStateForReceive(); 89 // 获取“写入管道”的线程 90 writeSide = Thread.currentThread(); 91 int bytesToTransfer = len; 92 while (bytesToTransfer > 0) 93 // 若“写入管道”的数据正好全部被读取完,则等待。 94 if (in == out) 95 awaitSpace(); 96 int nextTransferAmount = 0; 97 // 如果“管道中被读取的数据,少于写入管道的数据”; 98 // 则设置nextTransferAmount=“buffer.length - in” 99 if (out < in) 100 nextTransferAmount = buffer.length - in; 101 else if (in < out) // 如果“管道中被读取的数据,大于/等于写入管道的数据”,则执行后面的操作 102 // 若in==-1(即管道的写入数据等于被读取数据),此时nextTransferAmount = buffer.length - in; 103 // 否则,nextTransferAmount = out - in; 104 if (in == -1) 105 in = out = 0; 106 nextTransferAmount = buffer.length - in; 107 else 108 nextTransferAmount = out - in; 109 110 111 if (nextTransferAmount > bytesToTransfer) 112 nextTransferAmount = bytesToTransfer; 113 // assert断言的作用是,若nextTransferAmount <= 0,则终止程序。 114 assert(nextTransferAmount > 0); 115 // 将数据写入到缓冲中 116 System.arraycopy(b, off, buffer, in, nextTransferAmount); 117 bytesToTransfer -= nextTransferAmount; 118 off += nextTransferAmount; 119 in += nextTransferAmount; 120 if (in >= buffer.length) 121 in = 0; 122 123 124 125 126 // 检查管道状态 127 private void checkStateForReceive() throws IOException 128 if (!connected) 129 throw new IOException("Pipe not connected"); 130 else if (closedByWriter || closedByReader) 131 throw new IOException("Pipe closed"); 132 else if (readSide != null && !readSide.isAlive()) 133 throw new IOException("Read end dead"); 134 135 136 137 // 等待。 138 // 若“写入管道”的数据正好全部被读取完(例如,管道缓冲满),则执行awaitSpace()操作; 139 // 它的目的是让“读取管道的线程”管道产生读取数据请求,从而才能继续的向“管道”中写入数据。 140 private void awaitSpace() throws IOException 141 142 // 如果“管道中被读取的数据,等于写入管道的数据”时, 143 // 则每隔1000ms检查“管道状态”,并唤醒管道操作:若有“读取管道数据线程被阻塞”,则唤醒该线程。 144 while (in == out) 145 checkStateForReceive(); 146 147 /* full: kick any waiting readers */ 148 notifyAll(); 149 try 150 wait(1000); 151 catch (InterruptedException ex) 152 throw new java.io.InterruptedIOException(); 153 154 155 156 157 // 当PipedOutputStream被关闭时,被调用 158 synchronized void receivedLast() 159 closedByWriter = true; 160 notifyAll(); 161 162 163 // 从管道(的缓冲)中读取一个字节,并将其转换成int类型 164 public synchronized int read() throws IOException 165 if (!connected) 166 throw new IOException("Pipe not connected"); 167 else if (closedByReader) 168 throw new IOException("Pipe closed"); 169 else if (writeSide != null && !writeSide.isAlive() 170 && !closedByWriter && (in < 0)) 171 throw new IOException("Write end dead"); 172 173 174 readSide = Thread.currentThread(); 175 int trials = 2; 176 while (in < 0) 177 if (closedByWriter) 178 /* closed by writer, return EOF */ 179 return -1; 180 181 if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) 182 throw new IOException("Pipe broken"); 183 184 /* might be a writer waiting */ 185 notifyAll(); 186 try 187 wait(1000); 188 catch (InterruptedException ex) 189 throw new java.io.InterruptedIOException(); 190 191 192 int ret = buffer[out++] & 0xFF; 193 if (out >= buffer.length) 194 out = 0; 195 196 if (in == out) 197 /* now empty */ 198 in = -1; 199 200 201 return ret; 202 203 204 // 从管道(的缓冲)中读取数据,并将其存入到数组b中 205 public synchronized int read(byte b[], int off, int len) throws IOException 206 if (b == null) 207 throw new NullPointerException(); 208 else if (off < 0 || len < 0 || len > b.length - off) 209 throw new IndexOutOfBoundsException(); 210 else if (len == 0) 211 return 0; 212 213 214 /* possibly wait on the first character */ 215 int c = read(); 216 if (c < 0) 217 return -1; 218 219 b[off] = (byte) c; 220 int rlen = 1; 221 while ((in >= 0) && (len > 1)) 222 223 int available; 224 225 if (in > out) 226 available = Math.min((buffer.length - out), (in - out)); 227 else 228 available = buffer.length - out; 229 230 231 // A byte is read beforehand outside the loop 232 if (available > (len - 1)) 233 available = len - 1; 234 235 System.arraycopy(buffer, out, b, off + rlen, available); 236 out += available; 237 rlen += available; 238 len -= available; 239 240 if (out >= buffer.length) 241 out = 0; 242 243 if (in == out) 244 /* now empty */ 245 in = -1; 246 247 248 return rlen; 249 250 251 // 返回不受阻塞地从此输入流中读取的字节数。 252 public synchronized int available() throws IOException 253 if(in < 0) 254 return 0; 255 else if(in == out) 256 return buffer.length; 257 else if (in > out) 258 return in - out; 259 else 260 return in + buffer.length - out; 261 262 263 // 关闭管道输入流 264 public void close() throws IOException 265 closedByReader = true; 266 synchronized (this) 267 in = -1; 268 269 270
close()方法:此方法关闭流,读写线程都正常结束则此方法不起作用,但如一方线程非正常结束陷入挂起或出现问题导致程序运行不下去,则此方法会给挂起线程一个出口让其结束。
flush()方法:此方法,强制将缓冲区写入输出流,并唤醒等待的输入流线程读取数据。
以上是关于PipedInputStream类的主要内容,如果未能解决你的问题,请参考以下文章
JAVA IO流相关代码(Serializable接口,管道流PipedInputStream类,RandomAccessFile类相关代码)
JAVA IO流相关代码(Serializable接口,管道流PipedInputStream类,RandomAccessFile类相关代码)
Java 管道PipedInputStream PipedOutStream PipedReader PipedWriter