JDK源码阅读之BufferedInputStream

前言：

  BufferedInputStream 的作用是为另一个输入流添加一些功能，例如，提供“缓冲功能”以及支持“mark()标记”和“reset()重置方法”

BufferedInputStream

  现在看源码分析

public
class BufferedInputStream extends FilterInputStream {

    //默认的缓存数组大小
    private static int DEFAULT_BUFFER_SIZE = 8192;

    /*
                     要分配的最大数组大小。
       有些VM会在数组中保留一些头信息。
       尝试分配更大的数组可能会导致
       OutOfMemoryError：请求的数组大小超过VM限制
     */
    private static int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8;

    //缓存数组，变化对线程可见
    protected volatile byte buf[];

    /*  原子更新操作
                       与buf数组的volatile关键字共同组成了buf数组的原子更新功能实现
     */
    private static final
        AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater =
        AtomicReferenceFieldUpdater.newUpdater
        (BufferedInputStream.class,  byte[].class, "buf");

   //当前缓冲区的有效字节数
   //比缓冲区中最后一个有效字节的索引大 1 的索引。此值始终处于 0 到 buf.length 的范围内；
        //从 buf[0] 到 buf[count-1] 的元素包含从底层输入流中获取的缓冲输入数据。
    protected int count;

   /*
    *   缓冲区中的当前位置。这是将从 buf 数组中读取的下一个字符的索引。
         此值始终处于 0 到 count 的范围内。如果此值小于 count，则 buf[pos] 将作为下一个输入字节；
         如果此值等于 count，则下一次 read 或 skip 操作需要从包含的输入流中读取更多的字节。
    */
    protected int pos;

    /*
     * 最后一次调用 mark 方法时 pos 字段的值。
       此值始终处于 -1 到 pos 的范围内。如果输入流中没有被标记的位置，则此字段为 -1。
       如果输入流中有被标记的位置，则 buf[markpos] 将用作 reset 操作后的第一个输入字节。
       如果 markpos 不是 -1，则从位置 buf[markpos] 到 buf[pos-1] 之间的所有字节都必须保留在缓冲区数组中（尽管对 count、pos 和 markpos 的值进行适当调整后，
       这些字节可能移动到缓冲区数组中的其他位置）；除非 pos 与 markpos 的差超过 marklimit，否则不能将其丢弃。
     */
    protected int markpos = -1;

/*
 * 调用 mark 方法后，在后续调用 reset 方法失败之前所允许的最大提前读取量。
 * 只要 pos 与 markpos 之差超过 marklimit，就可以通过将 markpos 设置为 -1 来删除该标记。
 */
    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;
    }

  //创建默认大小的缓存输入流，包装指定的输入流
    //关于装饰器模式即装饰器模式在Java I/O里的应用，请看下一篇博文
    public BufferedInputStream(InputStream in) {
        this(in, DEFAULT_BUFFER_SIZE);
    }

  //创建指定大小的缓存输入流，包装指定的输入流
    public BufferedInputStream(InputStream in, int size) {
        super(in);
        if (size <= 0) {
            throw new IllegalArgumentException("Buffer size <= 0");
        }
        buf = new byte[size];
    }

 /*
  * fill()函数是整个 BufferedInputStream类的核心所在，我们重点来分析这个函数
  */
    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 if (buffer.length >= MAX_BUFFER_SIZE) {
                throw new OutOfMemoryError("Required array size too large");
               }
       
         else {            /* grow buffer */
                int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?
                        pos * 2 : MAX_BUFFER_SIZE;
                                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;
    }

   
    public synchronized int read() throws IOException {
        // 若已经读完缓冲区中的数据，则调用fill()从输入流读取下一部分数据来填充缓冲区
        if (pos >= count) {
            fill();
            //如果填充操作执行完后pos>=count,这时证明已经无数据向数组中填充
            if (pos >= count)
                return -1;
        }
        return getBufIfOpen()[pos++] & 0xff;//保持二进制数据一致性
    }

    private int read1(byte[] b, int off, int len) throws IOException {
       
        int avail = count - pos;
        if (avail <= 0) {
            /* 如果请求的长度至少与缓冲区一样大，并且
                如果没有标记/重置活动，那么就不要费心去复制
                字节进入本地缓冲区。 而是直接从流中读取，这其实是一种加快读取的机制*/
            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;
            // 如果真的没有可以读取的数据了，此时也会返回
            InputStream input = in;
            if (input != null && input.available() <= 0)
                return 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;
    }

 
    public synchronized void mark(int readlimit) {
        marklimit = readlimit;
        markpos = pos;
    }

    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()
        }
    }
}

对于fill()函数

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 if (buffer.length >= MAX_BUFFER_SIZE) {
                throw new OutOfMemoryError("Required array size too large");
               }
         else {            /* grow buffer */
                int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?
                        pos * 2 : MAX_BUFFER_SIZE;
                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方法，这个方法主要是对缓存数组进行操作，可以看到这个方法存在5条判断分支，并且除了抛出异常外，都会执行最后一段代码，所以这里我们可以将fill方法分解成5个小方法分别进行分析！暂时以fill+序号的方式命名这五个函数。

 一

  private void fill0() 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;

 这种情况指的是我们每次将输入流中的数据读取一部分到缓存数组中，当缓存数组中的数据被读完后，此时也没有进行任何标记，那么我们继续将输入流中的数据读取一部分到缓存数组中，结合read方法，我们通过pos >= count来判断缓存数组中的数据是否已经读完，通过markpos<0来判断是否存在标记！getInIfOpen().read用来向缓存数组中放入数据，最后通过count变量记录放入的数据个数。

二

private void fill1() throws IOException {
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;
               }
count = pos;
        int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
        if (n > 0)
            count = n + pos;
}

这种情况指的是我们读完了缓存数组中的数据（通过pos >=count判断）,数组中也没有多余的空间（通过pos >= buffer.length判断），而缓存中存在标记，我们需要将标记位置到数组结尾的数据保存下来已供读取，如下图：数组下标从0到7.

-1

0  data

1  data

2  data markpos

3 data

4 data

5  data

6 data

7  data

pos

标记及其后面的 数据都需要保留

-1

0 data

1data

2 data

3 data

4 data

5 data

pos

getInIfOpen().read方法; 从输入流中读取出“buffer.length - pos”的数据，然后填充到buffer中

 

三

private void fill2() throws IOException {
byte[] buffer = getBufIfOpen();
if (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;
    }

这种情况指的是读取完buffer中的数据，buffer被标记位置=0，buffer中没有多余的空间，并且buffer.length>=marklimit,这时缓存区太大了，标记无效，重置markpos和pos即可！

四

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

if (markpos >= 0 && pos >= buffer.length)
{            /* grow buffer */
                int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?
                        pos * 2 : MAX_BUFFER_SIZE;
                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;
            }

这种情况指的是读取完buffer中的数据，buffer被标记位置=0，buffer中没有多余的空间，并且buffer.length<marklimit，这时我们需要扩充数组，对于

int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?

                        pos * 2 : MAX_BUFFER_SIZE;

这条语句，为方便观察，我们改一下

int nsz=(pos<=MAX_BUFFER_SIZE/2) ?pos*2:MAX_BUFFER_SIZE;

很明显，我们扩充的数组不能超过MAX_BUFFER_SIZE，我们现在来考虑一种情况

，如果再执行fill()方法的过程中一直要执行fill3()方法，那么缓存数组就会越来越大，需要的内存越来越多，所以我们需要一个变量来控制数组的增长，而这个变量就是marklimit, 当buffer>=marklimit时，就不再保存markpos的值了