BufferedInputStream(緩沖輸入流)詳解_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理

BufferedInputStream(InputStream in)
BufferedInputStream(InputStream in, int size)

synchronized int   available()
void   close()
synchronized void   mark(int readlimit)
boolean   markSupported()
synchronized int   read()
synchronized int   read(byte[] buffer, int offset, int byteCount)
synchronized void   reset()
synchronized long   skip(long byteCount)

package java.io;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

public class BufferedInputStream extends FilterInputStream {

  // 默認(rèn)的緩沖大小是8192字節(jié)
  // BufferedInputStream 會(huì)根據(jù)“緩沖區(qū)大小”來(lái)逐次的填充緩沖區(qū)；
  // 即，BufferedInputStream填充緩沖區(qū)，用戶讀取緩沖區(qū)，讀完之后，BufferedInputStream會(huì)再次填充緩沖區(qū)。如此循環(huán)，直到讀完數(shù)據(jù)...
  private static int defaultBufferSize = 8192;

  // 緩沖數(shù)組
  protected volatile byte buf[];

  // 緩存數(shù)組的原子更新器。
  // 該成員變量與buf數(shù)組的volatile關(guān)鍵字共同組成了buf數(shù)組的原子更新功能實(shí)現(xiàn)，
  // 即，在多線程中操作BufferedInputStream對(duì)象時(shí)，buf和bufUpdater都具有原子性(不同的線程訪問(wèn)到的數(shù)據(jù)都是相同的)
  private static final
    AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater =
    AtomicReferenceFieldUpdater.newUpdater
    (BufferedInputStream.class, byte[].class, "buf");

  // 當(dāng)前緩沖區(qū)的有效字節(jié)數(shù)。
  // 注意，這里是指緩沖區(qū)的有效字節(jié)數(shù)，而不是輸入流中的有效字節(jié)數(shù)。
  protected int count;

  // 當(dāng)前緩沖區(qū)的位置索引
  // 注意，這里是指緩沖區(qū)的位置索引，而不是輸入流中的位置索引。
  protected int pos;

  // 當(dāng)前緩沖區(qū)的標(biāo)記位置
  // markpos和reset()配合使用才有意義。操作步驟：
  // (01) 通過(guò)mark() 函數(shù)，保存pos的值到markpos中。
  // (02) 通過(guò)reset() 函數(shù)，會(huì)將pos的值重置為markpos。接著通過(guò)read()讀取數(shù)據(jù)時(shí)，就會(huì)從mark()保存的位置開(kāi)始讀取。
  protected int markpos = -1;

  // marklimit是標(biāo)記的最大值。
  // 關(guān)于marklimit的原理，我們?cè)诤竺娴膄ill()函數(shù)分析中會(huì)詳細(xì)說(shuō)明。這對(duì)理解BufferedInputStream相當(dāng)重要。
  protected int marklimit;

  // 獲取輸入流
  private InputStream getInIfOpen() throws IOException {
    InputStream input = in;
    if (input == null)
      throw new IOException("Stream closed");
    return input;
  }

  // 獲取緩沖
  private byte[] getBufIfOpen() throws IOException {
    byte[] buffer = buf;
    if (buffer == null)
      throw new IOException("Stream closed");
    return buffer;
  }

  // 構(gòu)造函數(shù)：新建一個(gè)緩沖區(qū)大小為8192的BufferedInputStream
  public BufferedInputStream(InputStream in) {
    this(in, defaultBufferSize);
  }

  // 構(gòu)造函數(shù)：新建指定緩沖區(qū)大小的BufferedInputStream
  public BufferedInputStream(InputStream in, int size) {
    super(in);
    if (size <= 0) {
      throw new IllegalArgumentException("Buffer size <= 0");
    }
    buf = new byte[size];
  }

  // 從“輸入流”中讀取數(shù)據(jù)，并填充到緩沖區(qū)中。
  // 后面會(huì)對(duì)該函數(shù)進(jìn)行詳細(xì)說(shuō)明！
  private void fill() throws IOException {
    byte[] buffer = getBufIfOpen();
    if (markpos < 0)
      pos = 0;      /* no mark: throw away the buffer */
    else if (pos >= buffer.length) /* no room left in buffer */
      if (markpos > 0) { /* can throw away early part of the buffer */
        int sz = pos - markpos;
        System.arraycopy(buffer, markpos, buffer, 0, sz);
        pos = sz;
        markpos = 0;
      } else if (buffer.length >= marklimit) {
        markpos = -1;  /* buffer got too big, invalidate mark */
        pos = 0;    /* drop buffer contents */
      } else {      /* grow buffer */
        int nsz = pos * 2;
        if (nsz > marklimit)
          nsz = marklimit;
        byte nbuf[] = new byte[nsz];
        System.arraycopy(buffer, 0, nbuf, 0, pos);
        if (!bufUpdater.compareAndSet(this, buffer, nbuf)) {
          throw new IOException("Stream closed");
        }
        buffer = nbuf;
      }
    count = pos;
    int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
    if (n > 0)
      count = n + pos;
  }

  // 讀取下一個(gè)字節(jié)
  public synchronized int read() throws IOException {
    // 若已經(jīng)讀完緩沖區(qū)中的數(shù)據(jù)，則調(diào)用fill()從輸入流讀取下一部分?jǐn)?shù)據(jù)來(lái)填充緩沖區(qū)
    if (pos >= count) {
      fill();
      if (pos >= count)
        return -1;
    }
    // 從緩沖區(qū)中讀取指定的字節(jié)
    return getBufIfOpen()[pos++] & 0xff;
  }

  // 將緩沖區(qū)中的數(shù)據(jù)寫入到字節(jié)數(shù)組b中。off是字節(jié)數(shù)組b的起始位置，len是寫入長(zhǎng)度
  private int read1(byte[] b, int off, int len) throws IOException {
    int avail = count - pos;
    if (avail <= 0) {
      // 加速機(jī)制。
      // 如果讀取的長(zhǎng)度大于緩沖區(qū)的長(zhǎng)度 并且沒(méi)有markpos，
      // 則直接從原始輸入流中進(jìn)行讀取，從而避免無(wú)謂的COPY（從原始輸入流至緩沖區(qū)，讀取緩沖區(qū)全部數(shù)據(jù)，清空緩沖區(qū)， 
      // 重新填入原始輸入流數(shù)據(jù)）
      if (len >= getBufIfOpen().length && markpos < 0) {
        return getInIfOpen().read(b, off, len);
      }
      // 若已經(jīng)讀完緩沖區(qū)中的數(shù)據(jù)，則調(diào)用fill()從輸入流讀取下一部分?jǐn)?shù)據(jù)來(lái)填充緩沖區(qū)
      fill();
      avail = count - pos;
      if (avail <= 0) return -1;
    }
    int cnt = (avail < len) ? avail : len;
    System.arraycopy(getBufIfOpen(), pos, b, off, cnt);
    pos += cnt;
    return cnt;
  }

  // 將緩沖區(qū)中的數(shù)據(jù)寫入到字節(jié)數(shù)組b中。off是字節(jié)數(shù)組b的起始位置，len是寫入長(zhǎng)度
  public synchronized int read(byte b[], int off, int len)
    throws IOException
  {
    getBufIfOpen(); // Check for closed stream
    if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
      throw new IndexOutOfBoundsException();
    } else if (len == 0) {
      return 0;
    }

    // 讀取到指定長(zhǎng)度的數(shù)據(jù)才返回
    int n = 0;
    for (;;) {
      int nread = read1(b, off + n, len - n);
      if (nread <= 0)
        return (n == 0) ? nread : n;
      n += nread;
      if (n >= len)
        return n;
      // if not closed but no bytes available, return
      InputStream input = in;
      if (input != null && input.available() <= 0)
        return n;
    }
  }

  // 忽略n個(gè)字節(jié)
  public synchronized long skip(long n) throws IOException {
    getBufIfOpen(); // Check for closed stream
    if (n <= 0) {
      return 0;
    }
    long avail = count - pos;

    if (avail <= 0) {
      // If no mark position set then don't keep in buffer
      if (markpos <0)
        return getInIfOpen().skip(n);

      // Fill in buffer to save bytes for reset
      fill();
      avail = count - pos;
      if (avail <= 0)
        return 0;
    }

    long skipped = (avail < n) ? avail : n;
    pos += skipped;
    return skipped;
  }

  // 下一個(gè)字節(jié)是否存可讀
  public synchronized int available() throws IOException {
    int n = count - pos;
    int avail = getInIfOpen().available();
    return n > (Integer.MAX_VALUE - avail)
          ? Integer.MAX_VALUE
          : n + avail;
  }

  // 標(biāo)記“緩沖區(qū)”中當(dāng)前位置。
  // readlimit是marklimit，關(guān)于marklimit的作用，參考后面的說(shuō)明。
  public synchronized void mark(int readlimit) {
    marklimit = readlimit;
    markpos = pos;
  }

  // 將“緩沖區(qū)”中當(dāng)前位置重置到mark()所標(biāo)記的位置
  public synchronized void reset() throws IOException {
    getBufIfOpen(); // Cause exception if closed
    if (markpos < 0)
      throw new IOException("Resetting to invalid mark");
    pos = markpos;
  }

  public boolean markSupported() {
    return true;
  }

  // 關(guān)閉輸入流
  public void close() throws IOException {
    byte[] buffer;
    while ( (buffer = buf) != null) {
      if (bufUpdater.compareAndSet(this, buffer, null)) {
        InputStream input = in;
        in = null;
        if (input != null)
          input.close();
        return;
      }
      // Else retry in case a new buf was CASed in fill()
    }
  }
}

private void fill() throws IOException {
  byte[] buffer = getBufIfOpen();
  if (markpos < 0)
    pos = 0;
  else if (pos >= buffer.length) {
    if (markpos > 0) { /* can throw away early part of the buffer */
      int sz = pos - markpos;
      System.arraycopy(buffer, markpos, buffer, 0, sz);
      pos = sz;
      markpos = 0;
    } else if (buffer.length >= marklimit) {
      markpos = -1;  /* buffer got too big, invalidate mark */
      pos = 0;    /* drop buffer contents */
    } else {      /* grow buffer */
      int nsz = pos * 2;
      if (nsz > marklimit)
        nsz = marklimit;
      byte nbuf[] = new byte[nsz];
      System.arraycopy(buffer, 0, nbuf, 0, pos);
      if (!bufUpdater.compareAndSet(this, buffer, nbuf)) {
        // Can't replace buf if there was an async close.
        // Note: This would need to be changed if fill()
        // is ever made accessible to multiple threads.
        // But for now, the only way CAS can fail is via close.
        // assert buf == null;
        throw new IOException("Stream closed");
      }
      buffer = nbuf;
    }
  }

  count = pos;
  int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
  if (n > 0)
    count = n + pos;
}

private void fill() throws IOException {
  byte[] buffer = getBufIfOpen();
  if (markpos < 0)
    pos = 0;

  count = pos;
  int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
  if (n > 0)
    count = n + pos;
}

private void fill() throws IOException {
  byte[] buffer = getBufIfOpen();
  if (markpos >= 0 && pos >= buffer.length) {
    if (markpos > 0) {
      int sz = pos - markpos;
      System.arraycopy(buffer, markpos, buffer, 0, sz);
      pos = sz;
      markpos = 0;
    }
  }

  count = pos;
  int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
  if (n > 0)
    count = n + pos;
}

private void fill() throws IOException {
  byte[] buffer = getBufIfOpen();
  if (markpos >= 0 && pos >= buffer.length) {
    if ( (markpos <= 0) && (buffer.length >= marklimit) ) {
      markpos = -1;  /* buffer got too big, invalidate mark */
      pos = 0;    /* drop buffer contents */
    }
  }

  count = pos;
  int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
  if (n > 0)
    count = n + pos;
}

private void fill() throws IOException {
  byte[] buffer = getBufIfOpen();
  if (markpos >= 0 && pos >= buffer.length) {
    if ( (markpos <= 0) && (buffer.length < marklimit) ) {
      int nsz = pos * 2;
      if (nsz > marklimit)
        nsz = marklimit;
      byte nbuf[] = new byte[nsz];
      System.arraycopy(buffer, 0, nbuf, 0, pos);
      if (!bufUpdater.compareAndSet(this, buffer, nbuf)) {
        throw new IOException("Stream closed");
      }
      buffer = nbuf;
    }
  }

  count = pos;
  int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
  if (n > 0)
    count = n + pos;
}

int nsz = pos * 2;
if (nsz > marklimit)
  nsz = marklimit;
byte nbuf[] = new byte[nsz];

private void fill() throws IOException {
  byte[] buffer = getBufIfOpen();

  count = pos;
  int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
  if (n > 0)
    count = n + pos;
}

import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.InputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.FileNotFoundException;
import java.lang.SecurityException;

/**
 * BufferedInputStream 測(cè)試程序
 *
 * @author skywang
 */
public class BufferedInputStreamTest {

  private static final int LEN = 5;

  public static void main(String[] args) {
    testBufferedInputStream() ;
  }

  /**
   * BufferedInputStream的API測(cè)試函數(shù)
   */
  private static void testBufferedInputStream() {

    // 創(chuàng)建BufferedInputStream字節(jié)流，內(nèi)容是ArrayLetters數(shù)組
    try {
      File file = new File("bufferedinputstream.txt");
      InputStream in =
         new BufferedInputStream(
           new FileInputStream(file), 512);

      // 從字節(jié)流中讀取5個(gè)字節(jié)。“abcde”，a對(duì)應(yīng)0x61，b對(duì)應(yīng)0x62，依次類推...
      for (int i=0; i<LEN; i++) {
        // 若能繼續(xù)讀取下一個(gè)字節(jié)，則讀取下一個(gè)字節(jié)
        if (in.available() >= 0) {
          // 讀取“字節(jié)流的下一個(gè)字節(jié)”
          int tmp = in.read();
          System.out.printf("%d : 0x%s\n", i, Integer.toHexString(tmp));
        }
      }

      // 若“該字節(jié)流”不支持標(biāo)記功能，則直接退出
      if (!in.markSupported()) {
        System.out.println("make not supported!");
        return ;
      }
       
      // 標(biāo)記“當(dāng)前索引位置”，即標(biāo)記第6個(gè)位置的元素--“f”
      // 1024對(duì)應(yīng)marklimit
      in.mark(1024);

      // 跳過(guò)22個(gè)字節(jié)。
      in.skip(22);

      // 讀取5個(gè)字節(jié)
      byte[] buf = new byte[LEN];
      in.read(buf, 0, LEN);
      // 將buf轉(zhuǎn)換為String字符串。
      String str1 = new String(buf);
      System.out.printf("str1=%s\n", str1);

      // 重置“輸入流的索引”為mark()所標(biāo)記的位置，即重置到“f”處。
      in.reset();
      // 從“重置后的字節(jié)流”中讀取5個(gè)字節(jié)到buf中。即讀取“fghij”
      in.read(buf, 0, LEN);
      // 將buf轉(zhuǎn)換為String字符串。
      String str2 = new String(buf);
      System.out.printf("str2=%s\n", str2);

      in.close();
    } catch (FileNotFoundException e) {
      e.printStackTrace();
    } catch (SecurityException e) {
      e.printStackTrace();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}