BufferedInputStream 源码学习笔记

BufferedInputStream 是一个带有内存缓冲的 InputStream.

1.首先来看类结构 ：

BufferedInputStream是继承自FilterInputStream。
FilterInputStream继承自InputStream属于输入流中的链接流，同时引用了InputStream，将InputStream封装成一个内部变量，同时构造方法上需要传入一个InputStream。这是一个典型的装饰器模式，他的任何子类都可以对一个继承自InputStream的原始流或其他链接流进行装饰，如我们常用的使用BufferedInputStream对FileInputStream进行装饰，使普通的文件输入流具备了内存缓存的功能，通过内存缓冲减少磁盘io次数。

protected volatile InputStream in;
protected FilterInputStream(InputStream in) {
	this.in = in;
}

 注意：成员变量in使用了volatile关键字修饰，保障了该成员变量多线程情况下的可见性。

 

2.内存缓冲的实现
概要的了解完BufferedInputStream的继承关系，接下来详细理解BufferedInputStream是如何实现内存缓冲。既是内存缓冲，就涉及到内存的分配，管理以及如何实现缓冲。
通过构造方法可以看到：初始化了一个byte数组作为内存缓冲区，大小可以由构造方法中的参数指定，也可以是默认的大小。

protected volatile byte buf[];
private static int defaultBufferSize = 8192;
public BufferedInputStream(InputStream in, int size) {
	super(in);
    if (size <= 0) {
        throw new IllegalArgumentException("Buffer size <= 0");
    }
	buf = new byte[size];
}
public BufferedInputStream(InputStream in) {
	this(in, defaultBufferSize);
}

 

看完构造函数，大概可以了解其实现原理：通过初始化分配一个byte数组，一次性从输入字节流中读取多个字节的数据放入byte数组，程序读取部分字节的时候直接从byte数组中获取，直到内存中的数据用完再重新从流中读取新的字节。那么从api文档中我们可以了解到BufferedStream大概具备如下的功能：

 从api可以了解到BufferedInputStream除了使用一个byte数组做缓冲外还具备打标记，重置当前位置到标记的位置重新读取数据，忽略掉n个数据。这些功能都涉及到缓冲内存的管理，首先看下相关的几个成员变量：

protected int count;
protected int pos;
protected int markpos = -1;
protected int marklimit;

 count表示当前缓冲区内总共有多少有效数据；pos表示当前读取到的位置（即byte数组的当前下标，下次读取从该位置读取）；markpos：打上标记的位置；marklimit：最多能mark的字节长度，也就是从mark位置到当前pos的最大长度。

从最简单的read()读取一个字节的方法开始看：

public synchronized int read() throws IOException {
	if (pos >= count) {
	    fill();
	    if (pos >= count)
		return -1;
	}
	return getBufIfOpen()[pos++] & 0xff;
}

 当pos>=count的时候也就是表示当前的byte中的数据为空或已经被读完，他调用了一个fill()方法，从字面理解就是填充的意思，实际上是从真正的输入流中读取一些新数据放入缓冲内存中，之后直到缓冲内存中的数据读完前都不会再从真正的流中读取数据。
看源码中的fill()方法有很大一段是关于markpos的处理，其处理过程大致如下图：
a.没有markpos的情况很简单：
 

b.有mark的情况比较复杂：

3.read()方法返回值
以上即为内存缓冲管理的完全过程，再回过头看read()方法，当缓冲byte数组中有数据可以读时，直接从数组中读取一个字节，但最后的read方法返回的却是int，而且还和0xff做了与运算。

return getBufIfOpen()[pos++] & 0xff;

 为什么不直接返回一个byte，而是一个与运算后的int。首先宏观的看InputStream和Reader两个输入流的抽象类都定义了read接口而且都返回int，一个是字节流，一个是字符流。我们知道字节用byte表示，字符用char表示。首先看java中基本类型的取值范围：
 

 从取值范围来看int包含了char和byte，这为使用int作为返回值类型提供了可能。
在应用中我们一般用read()接口的返回值是-1则表示已经读到文件尾（EOF）。

char的取值范围本身不包含负数，所有用int的-1表示文件读完没问题，但byte的取值范围-128 ~ 127，包含了-1，读取的有效数据范围就是-128~127，没办法用这个取值范围中的任何一个数字表示异常或者数据已经读完，所以接口如果直接使用byte作为返回值不可行，直接将byte强制类型转换成int也不行，因为如果读到一个byte的-1，转为int了也是-1，会被理解为文件已经读完。所以这里做了一个特殊处理return getBufIfOpen()[pos++] & 0xff。

0xff是int类型，二进制为0000 0000 0000 0000 0000 0000 1111 1111。

上述的与运算实际上读取的byte先被强制转换成了int，例如byte的-1（最高位表示符号位，以补码的形式表示负数为：1111 1111）

转换为int之后的二进制1111 1111 1111 1111 1111 1111 1111 1111

& 0xff之后高位去0

最后返回的结果是0000 0000 0000 0000 0000 0000 1111 1111, 为int值为256

其-128~-1被转为int中128~256的正数表示。

 

这样解决了可以用-1表示文件已经读完。但关键是数据的值发生了变化，真正要用读取的数据时是否还能拿到原始的byte。还拿上面那个例子来看，当读取返回一个256时，将其强制类型转换为byte，(byte)256得到byte的-1，因为byte只有8位，当int的高位被丢弃后就只剩下1111 1111，在byte中高位的1表示符号位为负数，最终的结果即是byte的-1；同样byte的-128（1000 0000）被转为int的128（0000 0000 0000 0000 0000 0000 1000 0000），强制类型转换后还原byte的1000 0000。

 

4.线程安全

返回值中还有一个细节是getBufIfOpen()[pos++]，直接将pos++来获取下一个未读取的数据，这里涉及到的两个元素：一个内存数组，一个当前读取的数据下标都是全局变量，pos++也不是线程安全。那么BufferedInputStream如何保证对内存缓冲数组的操作线程安全？源码中有操作的public方法除了close方法之外，其他方法上都加上了synchronized关键字，以保障上面描述的整个内存缓存数组的操作是线程安全的。但为什么close方法没有synchronized，我们看这个方法做了些什么事情：

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

 简单来看做了两个操作：把内存数组置为null，将引用的inputStream置为null，同时将引用的inputStream.close()；
这两个操作的核心都是关闭原始流，释放资源，如果加了synchronized关键字，会导致当前线程正在执行read方法，而且系统消耗很大时，想释放资源无法释放。此时read方法还没执行完，我们知道synchronized的锁是加在整个对象上的，所以close方法就必须等到read结束后才能执行，这样很明显不能满足close的需求，甚至会导致大量的io资源被阻塞不能关闭。
但该方法用一个while循环，而且只有当bufUpdater.compareAndSet(this, buffer, null)成功时，才执行上述的资源释放。
先看bufUpdater这个全局变量：

protected volatile byte buf[];
private static final 
        AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater = 
        AtomicReferenceFieldUpdater.newUpdater
        (BufferedInputStream.class,  byte[].class, "buf");

 AtomicReferenceFieldUpdater是一个抽象类，但该类的内部已经给出了包访问控制级别的一个实现AtomicReferenceFieldUpdaterImpl，原理是利用反射将一个 被声明成volatile 的属性通过JNI调用，使用cpu指令级的命令将一个变量进行更新，保障该操作是原子的。也就是通过上面定义的bufUpdater将buf这个byte数组的跟新变为原子操作，其作用是保障其原子更新。
BufferedInputStream源代码中总共有两个地方用到了这个bufUpdater，一个是我们上面看到的close方法中，另外一个是再前面说道的fill()方法中。既然BufferedInputStream的所有操作上都用了synchronized来做同步，那为什么这里还需要用这个原子更新器呢？带着问题上面提到过fill()方法中的最后一个步骤：当有mark，而且markLimit的长度又大于初始数组的长度时，需要对内存数组扩容，即创建一个尺寸更大的数组，将原来数组中的数据拷贝到新数组中，再将指向原数组的应用指向新的数组。bufUpdater正是用在了将原数组引用指向新数组的操作上，同样close的方法使用的bufUpdater也是用在对数组引用的改变上，这样看来就比较清晰了，主要是为了防止一个线程在执行close方法时，将buffer赋值为null这个时候另外一个线程正在执行fill()方法的最后一个步骤又将buffer赋值给了一个新的数组，从而导致资源没有释放掉。

 

5.结束
到这里BufferedInputStream的源码每个细节都已经分析完，看似简单的一些方法，返回值和调用中其实蕴藏着很多不简单的东西，通过阅读一些好的源代码可以学到不少东西。

 

评论

1 楼 nergpchen 2015-11-18  

我也正在寻找return getBufIfOpen()[pos++] & 0xff 这个代码的含有，找到了你这里，谢谢博主的分享