源码剖析之java.io.BufferedInputStream

java 提供了读写缓存api。
byte：java.io.BufferedInputStream 、 java.io.BufferedOutputStream
char: java.io.BufferedReader、java.io.BufferedWriter

好处：
1、可以避免一次性写入大量的数据，这样可能瞬间造成内存占用太多，导致系统不稳定。
2、可以对写入较少的数据进行缓冲，避免写入输入承载终端太过频繁。

缺点：几乎没有。

所以几乎任何时候我们都有必要使用缓存的流进行包装。

记得刚学习java io时，听到缓存，感觉很神秘。其实java提供的缓存实现的数据结构就是数组，java 缓存 api都是建立在byte[]　和 char[]基础之上的，也即是：先write（read） 数据到 byte[]（char[]）里，然后一旦缓存数据满，再一次性读取（写入）。


源码分析：

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

/**
 * BufferedInputStream 继承自 FilterInputStream，所以其本身需要一个 internal InputStream 来承载数据的读取源头
 * 其缓存实现的本质是：byte[] 对数据的缓存。
 */
public class BufferedInputStream extends FilterInputStream {

    private static int defaultBufferSize = 8192;

    /**
     * 缓存数据存储的地方，必要时候可以改变大小（当然也不是原来的buf数组了）
     */
    protected volatile byte buf[];

    /**
     * Atomic updater to provide compareAndSet for buf. This is
     * necessary because closes can be asynchronous. We use nullness
     * of buf[] as primary indicator that this stream is closed. (The
     * "in" field is also nulled out on close.)
     */
    private static final 
        AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater = 
        AtomicReferenceFieldUpdater.newUpdater
        (BufferedInputStream.class,  byte[].class, "buf");

    /**
     * 缓存的有效字节数
     */
    protected int count;

    /**
     * 下一个可读的字节在数组中的位置
     */
    protected int pos;
    
    /**
     * 最近一次mark的位置
     */
    protected int markpos = -1;

    /**
     * The maximum read ahead allowed after a call to the 
     * <code>mark</code> method before subsequent calls to the 
     * <code>reset</code> method fail. 
     * Whenever the difference between <code>pos</code>
     * and <code>markpos</code> exceeds <code>marklimit</code>,
     * then the  mark may be dropped by setting
     * <code>markpos</code> to <code>-1</code>.
     *
     * @see     java.io.BufferedInputStream#mark(int)
     * @see     java.io.BufferedInputStream#reset()
     */
    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;
    }

  
    public BufferedInputStream(InputStream in) {
	this(in, defaultBufferSize);
    }

    public BufferedInputStream(InputStream in, int size) {
	super(in);
        if (size <= 0) {
            throw new IllegalArgumentException("Buffer size <= 0");
        }
	buf = new byte[size];
    }

    /**
     * Fills the buffer with more data, taking into account
     * shuffling and other tricks for dealing with marks.
     * Assumes that it is being called by a synchronized method.
     * This method also assumes that all data has already been read in,
     * hence pos > count.
     */
    private void fill() throws IOException {
        byte[] buffer = getBufIfOpen();
				if (markpos < 0) //如果没有设置mark，那么pos的位置设置为0。（因为没有mark，算是全部读取新数据）
				    pos = 0;		/* no mark: throw away the buffer */
				else if (pos >= buffer.length)	/* 缓冲里没有空闲空间，并且有mark设置 */
				  if (markpos > 0) {	/* 此种情况必须保留markpos之后的数据 */
					int sz = pos - markpos;
					System.arraycopy(buffer, markpos, buffer, 0, sz);
					pos = sz; //pos 在新的buffer里的位置是sz
					markpos = 0; //markpos 在新的buffer里面为 0，所以必须需要设置为0
				  } else if (buffer.length >= marklimit) { //出错啦，这种情况不该出现的
					markpos = -1;	/* buffer got too big, invalidate mark */
					pos = 0;	/* drop buffer contents */
				  } else {		/* 增长buffer */
					int nsz = pos * 2;
					if (nsz > marklimit)//但是不能超过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 {
			if (pos >= count) {//如果到了缓冲的最大索引位置,那么去fill数据
			    fill();
			    if (pos >= count) //fill之后如果pos>=count，那么说明 underlying stream 里已经没有数据
				return -1;
			}
			return getBufIfOpen()[pos++] & 0xff; //正常返回
    }

    /**
     * Read characters into a portion of an array, reading from the underlying
     * stream at most once if necessary.
     */
    private int read1(byte[] b, int off, int len) throws IOException {
			int avail = count - pos;
			if (avail <= 0) {
			    if (len >= getBufIfOpen().length && markpos < 0) { //如果没有多余的缓存数据,并且没有设置mark标记，并且流里剩余的数据不足len长度，那么直接从数据流里读取，直接返回。
				    return getInIfOpen().read(b, off, len);
			    }
			    fill(); //其他情况则去填充数据
			    avail = count - pos; //查看缓存中剩下的数据
			    if (avail <= 0) return -1; //如果缓存里没有新数据（即：fill没有添加新数据）。
			}
			int cnt = (avail < len) ? avail : len; //取剩下的数据和len的较小值
			System.arraycopy(getBufIfOpen(), pos, b, off, cnt); //把数据读取 到b数组里。
			pos += cnt;//改变当前字节的位置
			return cnt;
    }

    /**
     * 读取流数据到b数组里，从b[off]开始放数据，长度为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) { //如果 off<0 或者  len <0 或者  (off + len) <0 或者  b.length - (off + len) <0
	         throw new IndexOutOfBoundsException();
	      } else if (len == 0) { //len =0，直接返回
            return 0;
        }

	      int n = 0; //记录已经读到的总byte数
        for (;;) {
            int nread = read1(b, off + n, len - n);
            if (nread <= 0) //如果没有读到数据，那么直接返回
                return (n == 0) ? nread : n;
            n += nread;
            if (n >= len) //如果已经到len 那么返回n（注意：事实上n=len,而不会n>len）
                return n;
            // if not closed but no bytes available, return
            InputStream input = in;
            if (input != null && input.available() <= 0) //每次读过之后判断 input 是否还有多余的可读，如果没有那么直接返回
                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) //如果没有设置mark，那么直接调用 underlying stream  的skip方法
                return getInIfOpen().skip(n);
            
            // 填充数据
            fill();
            avail = count - pos;
            if (avail <= 0) //如果avail<=0，说明underlying stream 中也没有剩余数据
                return 0;
        }
        
        long skipped = (avail < n) ? avail : n; //最avail 和 n 的最小值
        pos += skipped; //调整pos 是skip方法的本质
        return skipped;
    }

    /**
     * 返回可以的字节数。
     * <p>
     * This method returns the sum of the number of bytes remaining to be read in
     * the buffer (<code>count - pos</code>) and the result of calling the
     * {@link java.io.FilterInputStream#in in}.available().
     *
     * @return     an estimate of the number of bytes that can be read (or skipped
     *             over) from this input stream without blocking.
     * @exception  IOException  if this input stream has been closed by
     *                          invoking its {@link #close()} method,
     *                          or an I/O error occurs.
     */
    public synchronized int available() throws IOException {
	     return getInIfOpen().available() + (count - pos);
    }

    /** 
     * 打标记
     * markpos：把当前位置记录为标记
     * readlimit：在mark变的无线之前运行读取的最大字节数
     */
    public synchronized void mark(int readlimit) {
			marklimit = readlimit;
			markpos = pos;
    }

    /**
     * 重置操作。
     * 如果没有mark过，调用reset方法会抛出：IOException
     *
     */
    public synchronized void reset() throws IOException {
        getBufIfOpen(); // Cause exception if closed
			if (markpos < 0)
			    throw new IOException("Resetting to invalid mark");
			pos = markpos; //重置的效果是：下一次read读取继续从上次打标记的地方读取。
    }

    /**
     * 支持回溯操作
     */
    public boolean markSupported() {
	     return true;
    }

    /**
     * 关闭此流，并释放与此流相关的任何系统资源
     
     * 一旦close流后，任何 read(), available(), reset()将抛出异常
     */
    public void close() throws IOException {
        byte[] buffer;
        while ( (buffer = buf) != null) {
            if (bufUpdater.compareAndSet(this, buffer, null)) { //将buf设置为空，如果失败则循环知道buf 为空为止。因为close 不是线程安全的，所以适用原子类bufUpdater 配合操作。
                InputStream input = in;
                in = null;
                if (input != null)
                    input.close();
                return;
            }
            // Else retry in case a new buf was CASed in fill()
        }
    }
}