java io系列12之 BufferedInputStream(緩沖輸入流)的認知、源碼和示例

本章內容包括3個部分：BufferedInputStream介紹，BufferedInputStream源碼，以及BufferedInputStream使用示例。

轉載請注明出處：http://www.cnblogs.com/skywang12345/p/io_12.html

BufferedInputStream 介紹

BufferedInputStream 是緩沖輸入流。它繼承於FilterInputStream。
BufferedInputStream 的作用是為另一個輸入流添加一些功能，例如，提供“緩沖功能”以及支持“mark()標記”和“reset()重置方法”。
BufferedInputStream 本質上是通過一個內部緩沖區數組實現的。例如，在新建某輸入流對應的BufferedInputStream后，當我們通過read()讀取輸入流的數據時，BufferedInputStream會將該輸入流的數據分批的填入到緩沖區中。每當緩沖區中的數據被讀完之后，輸入流會再次填充數據緩沖區；如此反復，直到我們讀完輸入流數據位置。

BufferedInputStream 函數列表

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)

 

BufferedInputStream 源碼分析(基於jdk1.7.40)

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

public class BufferedInputStream extends FilterInputStream {

    // 默認的緩沖大小是8192字節
    // BufferedInputStream 會根據“緩沖區大小”來逐次的填充緩沖區；
    // 即，BufferedInputStream填充緩沖區，用戶讀取緩沖區，讀完之后，BufferedInputStream會再次填充緩沖區。如此循環，直到讀完數據...
    private static int defaultBufferSize = 8192;

    // 緩沖數組
    protected volatile byte buf[];

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

    // 當前緩沖區的有效字節數。
    // 注意，這里是指緩沖區的有效字節數，而不是輸入流中的有效字節數。
    protected int count;

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

    // 當前緩沖區的標記位置
    // markpos和reset()配合使用才有意義。操作步驟：
    // (01) 通過mark() 函數，保存pos的值到markpos中。
    // (02) 通過reset() 函數，會將pos的值重置為markpos。接着通過read()讀取數據時，就會從mark()保存的位置開始讀取。
    protected int markpos = -1;

    // marklimit是標記的最大值。
    // 關於marklimit的原理，我們在后面的fill()函數分析中會詳細說明。這對理解BufferedInputStream相當重要。
    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;
    }

    // 構造函數：新建一個緩沖區大小為8192的BufferedInputStream
    public BufferedInputStream(InputStream in) {
        this(in, defaultBufferSize);
    }

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

    // 從“輸入流”中讀取數據，並填充到緩沖區中。
    // 后面會對該函數進行詳細說明！
    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;
    }

    // 讀取下一個字節
    public synchronized int read() throws IOException {
        // 若已經讀完緩沖區中的數據，則調用fill()從輸入流讀取下一部分數據來填充緩沖區
        if (pos >= count) {
            fill();
            if (pos >= count)
                return -1;
        }
        // 從緩沖區中讀取指定的字節
        return getBufIfOpen()[pos++] & 0xff;
    }

    // 將緩沖區中的數據寫入到字節數組b中。off是字節數組b的起始位置，len是寫入長度
    private int read1(byte[] b, int off, int len) throws IOException {
        int avail = count - pos;
        if (avail <= 0) {
            // 加速機制。
            // 如果讀取的長度大於緩沖區的長度 並且沒有markpos，
            // 則直接從原始輸入流中進行讀取，從而避免無謂的COPY（從原始輸入流至緩沖區，讀取緩沖區全部數據，清空緩沖區， 
            //  重新填入原始輸入流數據）
            if (len >= getBufIfOpen().length && markpos < 0) {
                return getInIfOpen().read(b, off, len);
            }
            // 若已經讀完緩沖區中的數據，則調用fill()從輸入流讀取下一部分數據來填充緩沖區
            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;
    }

    // 將緩沖區中的數據寫入到字節數組b中。off是字節數組b的起始位置，len是寫入長度
    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;
        }

        // 讀取到指定長度的數據才返回
        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個字節
    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;
    }

    // 下一個字節是否存可讀
    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;
    }

    // 標記“緩沖區”中當前位置。
    // readlimit是marklimit，關於marklimit的作用，參考后面的說明。
    public synchronized void mark(int readlimit) {
        marklimit = readlimit;
        markpos = pos;
    }

    // 將“緩沖區”中當前位置重置到mark()所標記的位置
    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;
    }

    // 關閉輸入流
    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()
        }
    }
}

說明：
要想讀懂BufferedInputStream的源碼，就要先理解它的思想。BufferedInputStream的作用是為其它輸入流提供緩沖功能。創建BufferedInputStream時，我們會通過它的構造函數指定某個輸入流為參數。BufferedInputStream會將該輸入流數據分批讀取，每次讀取一部分到緩沖中；操作完緩沖中的這部分數據之后，再從輸入流中讀取下一部分的數據。
為什么需要緩沖呢？原因很簡單，效率問題！緩沖中的數據實際上是保存在內存中，而原始數據可能是保存在硬盤或NandFlash等存儲介質中；而我們知道，從內存中讀取數據的速度比從硬盤讀取數據的速度至少快10倍以上。
那干嘛不干脆一次性將全部數據都讀取到緩沖中呢？第一，讀取全部的數據所需要的時間可能會很長。第二，內存價格很貴，容量不像硬盤那么大。

下面，我就BufferedInputStream中最重要的函數fill()進行說明。其它的函數很容易理解，我就不詳細介紹了，大家可以參考源碼中的注釋進行理解。

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;
}

 

根據fill()中的if...else...，下面我們將fill分為5種情況進行說明。

 

情況1：讀取完buffer中的數據，並且buffer沒有被標記

執行流程如下，
(01) read() 函數中調用 fill()
(02) fill() 中的 if (markpos < 0) ...
為了方便分析，我們將這種情況下fill()執行的操作等價於以下代碼：

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;
}

說明：
這種情況發生的情況是 — — 輸入流中有很長的數據，我們每次從中讀取一部分數據到buffer中進行操作。每次當我們讀取完buffer中的數據之后，並且此時輸入流沒有被標記；那么，就接着從輸入流中讀取下一部分的數據到buffer中。
其中，判斷是否讀完buffer中的數據，是通過 if (pos >= count) 來判斷的；
          判斷輸入流有沒有被標記，是通過 if (markpos < 0) 來判斷的。

理解這個思想之后，我們再對這種情況下的fill()的代碼進行分析，就特別容易理解了。
(01) if (markpos < 0) 它的作用是判斷“輸入流是否被標記”。若被標記，則markpos大於/等於0；否則markpos等於-1。
(02) 在這種情況下：通過getInIfOpen()獲取輸入流，然后接着從輸入流中讀取buffer.length個字節到buffer中。
(03) count = n + pos; 這是根據從輸入流中讀取的實際數據的多少，來更新buffer中數據的實際大小。

 

情況2：讀取完buffer中的數據，buffer的標記位置>0，並且buffer中沒有多余的空間

執行流程如下，
(01) read() 函數中調用 fill()
(02) fill() 中的 else if (pos >= buffer.length) ...
(03) fill() 中的 if (markpos > 0) ...

為了方便分析，我們將這種情況下fill()執行的操作等價於以下代碼：

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;
}

說明：
這種情況發生的情況是 — — 輸入流中有很長的數據，我們每次從中讀取一部分數據到buffer中進行操作。當我們讀取完buffer中的數據之后，並且此時輸入流存在標記時；那么，就發生情況2。此時，我們要保留“被標記位置”到“buffer末尾”的數據，然后再從輸入流中讀取下一部分的數據到buffer中。
其中，判斷是否讀完buffer中的數據，是通過 if (pos >= count) 來判斷的；
          判斷輸入流有沒有被標記，是通過 if (markpos < 0) 來判斷的。
          判斷buffer中沒有多余的空間，是通過 if (pos >= buffer.length) 來判斷的。

理解這個思想之后，我們再對這種情況下的fill()代碼進行分析，就特別容易理解了。
(01) int sz = pos - markpos; 作用是“獲取‘被標記位置’到‘buffer末尾’”的數據長度。
(02) System.arraycopy(buffer, markpos, buffer, 0, sz); 作用是“將buffer中從markpos開始的數據”拷貝到buffer中(從位置0開始填充，填充長度是sz)。接着，將sz賦值給pos，即pos就是“被標記位置”到“buffer末尾”的數據長度。
(03) int n = getInIfOpen().read(buffer, pos, buffer.length - pos); 從輸入流中讀取出“buffer.length - pos”的數據，然后填充到buffer中。
(04) 通過第(02)和(03)步組合起來的buffer，就是包含了“原始buffer被標記位置到buffer末尾”的數據，也包含了“從輸入流中新讀取的數據”。

注意：執行過情況2之后，markpos的值由“大於0”變成了“等於0”！

 

情況3：讀取完buffer中的數據，buffer被標記位置=0，buffer中沒有多余的空間，並且buffer.length>=marklimit

執行流程如下，
(01) read() 函數中調用 fill()
(02) fill() 中的 else if (pos >= buffer.length) ...
(03) fill() 中的 else if (buffer.length >= marklimit) ...

為了方便分析，我們將這種情況下fill()執行的操作等價於以下代碼：

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;
}

說明：這種情況的處理非常簡單。首先，就是“取消標記”，即 markpos = -1；然后，設置初始化位置為0，即pos=0；最后，再從輸入流中讀取下一部分數據到buffer中。

 

情況4：讀取完buffer中的數據，buffer被標記位置=0，buffer中沒有多余的空間，並且buffer.length<marklimit

執行流程如下，
(01) read() 函數中調用 fill()
(02) fill() 中的 else if (pos >= buffer.length) ...
(03) fill() 中的 else { int nsz = pos * 2; ... }

為了方便分析，我們將這種情況下fill()執行的操作等價於以下代碼：

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;
}

說明：
這種情況的處理非常簡單。
(01) 新建一個字節數組nbuf。nbuf的大小是“pos*2”和“marklimit”中較小的那個數。

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

(02) 接着，將buffer中的數據拷貝到新數組nbuf中。通過System.arraycopy(buffer, 0, nbuf, 0, pos)
(03) 最后，從輸入流讀取部分新數據到buffer中。通過getInIfOpen().read(buffer, pos, buffer.length - pos);
注意：在這里，我們思考一個問題，“為什么需要marklimit，它的存在到底有什么意義？”我們結合“情況2”、“情況3”、“情況4”的情況來分析。

假設，marklimit是無限大的，而且我們設置了markpos。當我們從輸入流中每讀完一部分數據並讀取下一部分數據時，都需要保存markpos所標記的數據；這就意味着，我們需要不斷執行情況4中的操作，要將buffer的容量擴大……隨着讀取次數的增多，buffer會越來越大；這會導致我們占據的內存越來越大。所以，我們需要給出一個marklimit；當buffer>=marklimit時，就不再保存markpos的值了。

 

情況5：除了上面4種情況之外的情況

執行流程如下，
(01) read() 函數中調用 fill()
(02) fill() 中的 count = pos...

為了方便分析，我們將這種情況下fill()執行的操作等價於以下代碼：

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;
}

說明：這種情況的處理非常簡單。直接從輸入流讀取部分新數據到buffer中。

 

示例代碼

關於BufferedInputStream中API的詳細用法，參考示例代碼(BufferedInputStreamTest.java)：

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 測試程序
 *
 * @author skywang
 */
public class BufferedInputStreamTest {

    private static final int LEN = 5;

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

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

        // 創建BufferedInputStream字節流，內容是ArrayLetters數組
        try {
            File file = new File("bufferedinputstream.txt");
            InputStream in =
                  new BufferedInputStream(
                      new FileInputStream(file), 512);

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

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

            // 跳過22個字節。
            in.skip(22);

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

            // 重置“輸入流的索引”為mark()所標記的位置，即重置到“f”處。
            in.reset();
            // 從“重置后的字節流”中讀取5個字節到buf中。即讀取“fghij”
            in.read(buf, 0, LEN);
            // 將buf轉換為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();
       }
    }
}

程序中讀取的bufferedinputstream.txt的內容如下：

abcdefghijklmnopqrstuvwxyz
0123456789
ABCDEFGHIJKLMNOPQRSTUVWXYZ

運行結果：

0 : 0x61
1 : 0x62
2 : 0x63
3 : 0x64
4 : 0x65
str1=01234
str2=fghij

 

---

更多內容

01. java 集合系列目錄(Category)

02. java io系列01之 IO框架

03. java io系列02之 ByteArrayInputStream的簡介,源碼分析和示例(包括InputStream)

04. java io系列03之 ByteArrayOutputStream的簡介,源碼分析和示例(包括OutputStream)

05. java io系列04之 管道(PipedOutputStream和PipedInputStream)的簡介,源碼分析和示例

06. java io系列05之 ObjectInputStream 和 ObjectOutputStream

07. java io系列06之 序列化總結(Serializable 和 Externalizable)

08. java io系列07之 FileInputStream和FileOutputStream

09. java io系列08之 File總結

10. java io系列09之 FileDescriptor總結

11. java io系列10之 FilterInputStream

12. java io系列11之 FilterOutputStream

13. java io系列12之 BufferedInputStream(緩沖輸入流)的認知、源碼和示例