首页 文章 精选 留言 我的

精选列表

搜索[源码阅读]，共10000篇文章
优秀的个人博客,低调大师

Java PipedInputStream PipedOutputStream类源码解析

管道流主要是用于不同线程间的数据交互，可以通过一个PipedInputStream和一个PipedOutputStream相互连接来进行通信，从PipedOutputStream写入字节到PipedInputStream中，所以PipedOutputStream是writer端，PipedInputStream是reader端。 一个PipedInputStream只能与一个PipedOutputStream连接，但是可以通过多个线程共享同一个管道来达到复数生产者和复数消费者的模式，同一个角色相互之间会存在竞争。下面的例子是一个writer和两个reader共用同一个管道流，可以看到两个reader之间根据线程调度随机接收一部分数据。 public class PipeStreamTest implements Runnable { private PipedInputStream in; private PipedOutputStream out; public PipeStreamTest(PipedInputStream in) { this.in = in; } public PipeStreamTest(PipedOutputStream out) { this.out = out; } @Override public void run() { if (this.in != null) { boolean flag = true; while (flag) { try { int a = in.read(); if (a == -1) { flag = false; break; } System.out.println(Thread.currentThread().getName() + " calculate " + String.valueOf(2 * a + 1)); } catch (IOException e) { e.printStackTrace(); flag = false; } } } else { try { Thread.sleep(1000);// 这里阻塞了out，导致in内没有可读取数据被阻塞 } catch (InterruptedException e1) { e1.printStackTrace(); } int a = 5; try { for (int i = 0; i < 5; i++) { out.write(a + i); System.out.println(Thread.currentThread().getName() + " write " + String.valueOf(a)); } out.close(); } catch (IOException e) { e.printStackTrace(); } } } public static void main(String args[]) throws IOException { PipedInputStream in = new PipedInputStream(10); PipedOutputStream out = new PipedOutputStream(in); new Thread(new PipeStreamTest(in)).start(); new Thread(new PipeStreamTest(in)).start(); new Thread(new PipeStreamTest(out)).start(); /*Thread-0 calculate 11 Thread-2 write 5 Thread-2 write 5 Thread-0 calculate 13 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-0 calculate 15 Thread-0 calculate 19 Thread-1 calculate 17*/ } } 对run()进行一些修改，阻塞reader线程，使得writer可以写满缓存区，此时writer会被阻塞，等待有reader读取数据后缓冲区有空余空间再写入，由于Thread.sleep不会被notifyAll()影响，所以在写入了10bytes数据后会存在5s的明显间隔 public void run() { if (this.in != null) { try { Thread.sleep(5000);// 这里阻塞了in，使得out会因没有空间写入而阻塞 } catch (InterruptedException e1) { e1.printStackTrace(); } boolean flag = true; while (flag) { try { int a = in.read(); if (a == -1) { flag = false; break; } System.out.println(Thread.currentThread().getName() + " calculate " + String.valueOf(2 * a + 1)); } catch (IOException e) { e.printStackTrace(); flag = false; } } } else { try { Thread.sleep(1000);// 这里阻塞了out，导致in内没有可读取数据被阻塞 } catch (InterruptedException e1) { e1.printStackTrace(); } int a = 5; try { for (int i = 0; i < 15; i++) { out.write(a + i); System.out.println(Thread.currentThread().getName() + " write " + String.valueOf(a)); } out.close(); } catch (IOException e) { e.printStackTrace(); } } } /* Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 Thread-2 write 5 这里出现明显的间隔 Thread-0 calculate 11 Thread-0 calculate 15 Thread-2 write 5 Thread-1 calculate 13 Thread-1 calculate 19 Thread-1 calculate 21 Thread-1 calculate 23 Thread-1 calculate 25 Thread-1 calculate 27 Thread-1 calculate 29 Thread-2 write 5 Thread-2 write 5 Thread-0 calculate 17 Thread-0 calculate 33 Thread-0 calculate 35 Thread-0 calculate 37 Thread-2 write 5 Thread-2 write 5 Thread-1 calculate 31 Thread-0 calculate 39*/ PipedInputStream 下面先来分析一下管道字节输入流PipedInputStream，它继承了InputStream。一个管道输入流需要连接一个管道输出流，管道输入流会提供任何数据字节写入管道输出流。典型地，数据被一个线程从一个PipedInputStream对象中读取，然后数据被另一个线程写到对应的PipedOutputStream。试图由一个线程使用所有对象是不推荐的，这可能导致线程死锁。 管道输入流含有一个缓冲区来解耦一个读操作和写操作。一个管道如果提供数据给相连接的管道输出流的线程不再活动，这个管道状态被称为BROKEN，也就是说如果上面的例子中把in对应的线程循环去掉使它运行完后被终止，此时out再尝试写入就会出现管道BROKEN的异常。 先来看一下跟管道状态有关的内部变量，流的关闭需要输入和输出端都关闭，所以需要两个变量来标志。同时，所有读写操作都需要管道两端建立连接，connected表示是否有连接建立。最后两个线程是输入和输出的操作线程，由于接收需要读取线程是运行状态，读取在缓冲区内为空时需要 /** * 输出流关闭，仅PipedOutputStream.close()调用receivedLast()方法会将它置为true */ boolean closedByWriter = false; /** * 输入流关闭，仅PipedInputStream.close()会将它置为true */ volatile boolean closedByReader = false; /** * 是否有一对输入输出流相互连接 */ boolean connected = false; /* 识别读和写两边需要更加复杂。使用线程组（但是管道在一个线程中怎么办）或者使用final化（但是可能到下一次GC的时间更长） */ Thread readSide;//input流的线程 Thread writeSide;//output流的线程 下面这些内部变量和缓冲区有关。buffer数组是循环缓冲区，这里有in和out两个下标指针，in负责接收，out负责读取，两个下标都是循环的如果达到末端会重新回到0的位置，读取不能超过接收数据的范围。 private static final int DEFAULT_PIPE_SIZE = 1024; /** * 管道循环输入缓冲区默认大小1K */ // 这个值在管道大小允许改变前被用作常数。这个值会为了向下兼容性被持续保持 protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE; /** * 循环缓冲区，接下来的数据会被放入其中 */ protected byte buffer[]; /** * 循环缓冲区中下一个从连接的管道输出流接收的字节数据将会被存储的位置下标。in<0说明缓冲区是空的，in==out说明缓冲区满了 */ protected int in = -1; /** * 循环缓冲区中下一个将会被这个管道输入流读取的字节下标位置。 */ protected int out = 0; 构造函数方面，可以直接给出管道输出流直接在构造时建立连接，也可以先构造输入流在需要使用时再连接。循环缓冲区的大小可以使用默认的1K，也可以手动指定大小。connect方法用于建立两端的连接，可以在构造时调用也可以手动调用。 /** * 创建一个PipedInputStream连接到管道输出流src。数据字节写入到src中将会成为这个流的输入。 */ public PipedInputStream(PipedOutputStream src) throws IOException { this(src, DEFAULT_PIPE_SIZE); } /** * 跟上面相比这个管道缓冲区的大小是指定的，其他相同 */ public PipedInputStream(PipedOutputStream src, int pipeSize) throws IOException { initPipe(pipeSize); connect(src); } /** * 创建一个还没有连接的PipedInputStream，在使用前必须连接到一个PipedOutputStream */ public PipedInputStream() { initPipe(DEFAULT_PIPE_SIZE); } /** * 创建一个还没有连接的PipedInputStream指定它的缓冲区大小，在使用前必须连接到一个PipedOutputStream */ public PipedInputStream(int pipeSize) { initPipe(pipeSize); } private void initPipe(int pipeSize) { if (pipeSize <= 0) { throw new IllegalArgumentException("Pipe Size <= 0"); } buffer = new byte[pipeSize]; } /** * 引起这个管道输入流连接到管道输出流src。如果这个对象已经连接到某个其他的管道输出流会抛出IOException。 * 如果src是一个未连接的管道输出流，snk是一个未连接的管道输入流，它们可以通过snk.connect(src)或者src.connect(snk)连接，两者效果相同。 */ public void connect(PipedOutputStream src) throws IOException { src.connect(this); } receive方法将字节写入到缓冲区，这是protected方法，再没有继承类的情况下由PipedOutputStream.write方法调用，接收字节会导致下标指针in增加，如果到达右边界，会重置为0。如果当前缓冲区没有空余的空间，这个方法会被阻塞，直到有线程读取了数据使得有空间写入时再继续。 /** * 接收一字节数据，这个方法如果没有有效输入时会阻塞。 */ protected synchronized void receive(int b) throws IOException { checkStateForReceive();//检查管道状态 writeSide = Thread.currentThread();//写线程设为当前线程 if (in == out) awaitSpace();//缓冲区满了，通知所有线程使得读取端读取字节给当前流缓冲区空出位置来接收 if (in < 0) {//in<0表示当前缓冲区为空 in = 0; out = 0; } buffer[in++] = (byte)(b & 0xFF);//将b存储到缓冲区中 if (in >= buffer.length) { in = 0;//因为是循环缓冲区，所以in超过buffer.length时重新回到0的位置 } } /** * 将数据接收到一个字节数组中，这个方法会阻塞到一些输入变得有效。 */ synchronized void receive(byte b[], int off, int len) throws IOException { checkStateForReceive();//检查管道状态 writeSide = Thread.currentThread();//写线程设为当前线程 int bytesToTransfer = len; while (bytesToTransfer > 0) {//还有需要写入的字节 if (in == out) awaitSpace();//缓冲区满了，通知其他线程读取字节空出空间 int nextTransferAmount = 0; if (out < in) {//out<in说明out到in这段是未读取的数据，所以空余空间是buffer.length-in nextTransferAmount = buffer.length - in; } else if (in < out) { if (in == -1) { //当前缓冲区为空，可用空间为buffer.length in = out = 0; nextTransferAmount = buffer.length - in; } else { //in已经到达一次右边界重置为0之后in<out，out到右边界是未读取内容，in不能超过out的值，所以空余空间为out-in nextTransferAmount = out - in; } } if (nextTransferAmount > bytesToTransfer) nextTransferAmount = bytesToTransfer;//要读取的字节数是剩余空间和参数指定长度间的较小值 assert(nextTransferAmount > 0); System.arraycopy(b, off, buffer, in, nextTransferAmount);//将接收的字节复制到缓冲区in开始的位置 bytesToTransfer -= nextTransferAmount; off += nextTransferAmount; in += nextTransferAmount;//in增加读取的字节数 if (in >= buffer.length) { in = 0;//in到达缓冲区边界时重置为0 } } } 然后看下这用到的两个内部方法。checkStateForReceive这个内部方法，确定当前有连接，管道两端都没有被关闭，有活动的读取端线程。当in==out时说明缓冲区满了不能再写入，需要有线程读取使得out增大之后才能再写入，awaitSpace会唤醒所有的等待读取线程，然后等待1s，再检查in==out是否成立，不断循环这个过程。 private void checkStateForReceive() throws IOException { if (!connected) { throw new IOException("Pipe not connected"); } else if (closedByWriter || closedByReader) { throw new IOException("Pipe closed"); } else if (readSide != null && !readSide.isAlive()) { throw new IOException("Read end dead"); } } private void awaitSpace() throws IOException { while (in == out) { checkStateForReceive(); /* full: kick any waiting readers */ notifyAll();//唤醒所有线程，使得有reader读取字节将当前流缓冲区空出来 try { wait(1000);//当前线程等待1s } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } } } read方法从缓冲区中读取字节，要求存在连接且读取端没有关闭，如果缓冲区内没有可读取的字节则还需要writer端线程是活跃的。读取多个字节时会先尝试允许阻塞读取一个字节，成功后再读取后面部分，此时不会再阻塞等待所以如果要读取的长度太长则只读取缓冲区内所有可读取的内容，返回的是实际读取的字节数。 /** * 从这个管道输入流读取下一个字节。返回值作为一个整数范围在0-255之间。这个方法一直阻塞到输入数据变得有效，或者探知到流结束或者抛出异常。 */ public synchronized int read() throws IOException { //存在连接，读取端没有关闭，缓存区为空时写入端线程必须活动 if (!connected) { throw new IOException("Pipe not connected"); } else if (closedByReader) { throw new IOException("Pipe closed"); } else if (writeSide != null && !writeSide.isAlive() && !closedByWriter && (in < 0)) { throw new IOException("Write end dead"); } readSide = Thread.currentThread();//读取端线程设为当前线程 int trials = 2;//加起来等待2s，第三次循环缓冲区依然为空没有写入数据则认为管道BROKEN while (in < 0) { if (closedByWriter) { /* 输出管道被writer关闭返回-1 */ return -1; } if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) { throw new IOException("Pipe broken"); } /* 可能有一个writer在等待 */ notifyAll(); try { wait(1000); } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } } int ret = buffer[out++] & 0xFF;//读取out位置的字节并增加out if (out >= buffer.length) { out = 0;//如果out到达buffer右边界，将它重置为0 } if (in == out) { /* in==out说明缓冲区空了 */ in = -1; } return ret; } /** * 从这个管道输入流读取最大len程度的字节数据到字节数组中。如果到达了数据量末端或者len超过了管道缓冲区大小，少于len字节的数据会被读取。 * 如果len是0，没有数据会被读取返回0；否则这个方法会阻塞直到至少1字节输入是有效的，或者到达了流末端或者抛出异常。 */ public synchronized int read(byte b[], int off, int len) throws IOException { //存在连接，读取端没有关闭，缓存区为空时写入端线程必须活动 if (b == null) { throw new NullPointerException(); } else if (off < 0 || len < 0 || len > b.length - off) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return 0; } /* 可能等待第一个字节 */ int c = read();//尝试读取一个字节 if (c < 0) { return -1;//没有读取到直接返回-1 } b[off] = (byte) c;//将读取到的字节存入数组 int rlen = 1; while ((in >= 0) && (len > 1)) { int available; if (in > out) { available = Math.min((buffer.length - out), (in - out));//in>out则可读取的数据是in-out，正常情况下in不能超过buffer.length } else { available = buffer.length - out;//in<out则可读取数据是out到buffer右边界，一次循环只能读取到右边界，从头开始的部分要下一次循环读取 } // 在循环外事先读取的字节 if (available > (len - 1)) { available = len - 1;//读取的字节不能超过参数指定的数量-1因为循环外已经读取了一个字节 } System.arraycopy(buffer, out, b, off + rlen, available);//复制可读取字节 out += available;//out位置移动读取的字节数 rlen += available;//目标位置移动 len -= available;//待读取长度减少 if (out >= buffer.length) { out = 0; } if (in == out) { /* in==out缓冲区为空，将in置为-1，所以循环会跳出 */ in = -1; } } return rlen; } available返回从这个输入流可以读取多少字节不用阻塞 public synchronized int available() throws IOException { if(in < 0) return 0; else if(in == out) return buffer.length;//因为在接收和读取时没有可读取的字节会把in置为-1，所以in==out是缓冲区满了的状态 else if (in > out) return in - out;//in>out时可读取内容是in到out之间 else return in + buffer.length - out;//in<out时是out到buffer右边界再加上buffer左边界到in的内容 } close关闭管道输入流，释放任何和这个流关联的资源 public void close() throws IOException { closedByReader = true;//输入流关闭 synchronized (this) { in = -1;//缓冲区所有数据失效 } } PipedOutputStream PipedOutputStream继承了OutputStream。管道输出流可以连接到一个管道输入流来创建一个通信管道。管道输出流是发送端。一般来说，一个线程将数据写入到一个PipedOutputStream对象，其他线程从连接的PipedInputStream对象读取数据。不推荐使用同一个线程来同时使用两个对象，可能会引起这个线程死锁。如果一个从连接的管道输入流读取数据的线程不再活动，这个管道被称为broken状态。 PipedOutputStream的代码很少，基本上全部都是调用PipedInputStream的方法。 内部变量只有一个private PipedInputStream sink用来判定输入端的状态。 构造方法可以指定输入流，也可以直接用无参构造使用前再手动调用连接方法 /** * 创建一个管道输出流连接到指定的管道输入流。数据字节写入到这个流中将会称为snk的有效输入。 */ public PipedOutputStream(PipedInputStream snk) throws IOException { connect(snk); } /** * 创建一个管道输出流还没有连接管道输入流。它在使用前必须通过接收者或者发送者连接到一个管道输入流。 */ public PipedOutputStream() { } connect方法要求两端都不能有已有的连接，否则会抛出异常 /** * 连接这个管道输出流到一个接收者。如果这个对象已经连接到了某个其他的管道输入流，会抛出IOException。 * 如果snk是一个未连接的管道输入流，src是一个未连接的管道输出流，它们可以通过两种方式连接： * src.connect(snk)或者snk.connect(src)。这两种方式是同样的效果。 */ public synchronized void connect(PipedInputStream snk) throws IOException { if (snk == null) { throw new NullPointerException(); } else if (sink != null || snk.connected) { throw new IOException("Already connected");//如果自身或者snk已经连接到某个管道，则抛出异常 } sink = snk;//因为snk内的属性是protected所以可以被同package的PipedOutputStream直接修改 snk.in = -1; snk.out = 0; snk.connected = true;//连接状态修改 } write方法写入数据到缓冲区，需要检查是否有连接的输入端，输入字符数组时还要检查数组和位置参数是否在正确范围内，最后调用输入流的receive方法 /** * 将指定的字节写入到管道输出流，实现了OutputStream.write */ public void write(int b) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } sink.receive(b);//调用PipedInputStream.receive } /** * 将指定的数组中从off偏移开始长度len的字节写入到这个管道输出流。这个方法会阻塞，直到所有的字节都写入到输出流中。 */ public void write(byte b[], int off, int len) throws IOException { if (sink == null) { throw new IOException("Pipe not connected"); } else if (b == null) { throw new NullPointerException(); } else if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) > b.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return; } sink.receive(b, off, len); } flush()刷新这个输出流，并且促使任何缓冲的输出数据被写出。这个方法会唤醒所有的字节在管道中等待的读取端进入就绪状态。 public synchronized void flush() throws IOException { if (sink != null) { synchronized (sink) { sink.notifyAll(); } } } close关闭这个管道输出流并释放所有关联的系统资源。这个流不能再用于写字节。 public void close() throws IOException { if (sink != null) { sink.receivedLast(); } } /** * 通知所有等待的线程最后一个字节数据已经被接收 */ synchronized void receivedLast() { closedByWriter = true; notifyAll(); } PipedReader PipedReader是管道字符输入流，总体设计逻辑基本上和PipedInputStream是一致的，它继承了Reader。因为总体上都是一样的，就只挑区别来讲了。 第一个区别，因为是字符流，缓冲区变为字符数组 char buffer[]; receive单个字符把PipedInputStream里两个内部方法直接写到代码里了，然后因为类型的改变这里有强制转换类型的区别。 synchronized void receive(int c) throws IOException { if (!connected) { throw new IOException("Pipe not connected"); } else if (closedByWriter || closedByReader) { throw new IOException("Pipe closed"); } else if (readSide != null && !readSide.isAlive()) { throw new IOException("Read end dead"); } writeSide = Thread.currentThread(); while (in == out) { if ((readSide != null) && !readSide.isAlive()) { throw new IOException("Pipe broken"); } /* full: kick any waiting readers */ notifyAll(); try { wait(1000); } catch (InterruptedException ex) { throw new java.io.InterruptedIOException(); } } if (in < 0) { in = 0; out = 0; } buffer[in++] = (char) c;//int转为char if (in >= buffer.length) { in = 0; } } receive多个字符代码很简单，前面分析过PipedInputStream是在缓冲区满时等待读取，然后写入直到再次填满缓冲区或者写完所有数据，而PipedReader则是一个个字符存储到缓冲区，每一次写入都有可能发生阻塞等待。 synchronized void receive(char c[], int off, int len) throws IOException { while (--len >= 0) { receive(c[off++]); } } read设计逻辑上没有区别，也是单个字符读取最多等待2个循环也就是2s。多个字符读取会先尝试读取一个字符，后面无阻塞的尽可能读取多的字符来满足要求的长度，返回实际读取的字符个数。 PipedWriter设计和PipedOutputStream逻辑上无区别，就不说了。
2018-09-09
优秀的个人博客,低调大师

Java BufferedReader BufferedWriter类源码解析

BufferedReader 像BufferedInputStream为FileInputStream提供了缓冲区一样，BufferedReader为InputStreamReader提供了缓冲区。一般情况下，所有的读取都是先从下层输入流读取到缓冲区，然后再从缓冲区读取到目标数组，除非出现要读取的长度超过了缓冲区大小且缓冲区没有有效数据和有效mark，可以直接从下层输入流读取字符。 来看一下这个类的头部注释：从字符输入流读取文本，缓冲字符来提供高效地读取字符、数组和行。缓冲区的大小可能是指定的或者是默认大小，默认大小对于大部分情况足够大了。一般来说，Reader的每一个读取请求引起一个下层的字节或者字符输入流的读取请求。所以建议用BufferedReader来包裹一个read方法花费大的Reader，比如FileReaders和InputStreamReaders。例如 BufferedReader in = new BufferedReader(new FileReader("foo.in"));将会缓存从特定文件来的输入。没有缓冲，每一次调用read()或者readLine()可能引起从文件读取字节，转换为字符后返回，这样效率不高。使用DataInputStreams进行文本输入的程序可以通过替换每一个DataInputStream为合适的BufferedReader进行局部化。 前面已经解释过为什么可以通过从文件读取大块内容存储到内存缓冲区来提高整体读取速度，BufferedReader就是字符缓冲输入流。它提供了一个字符数组来作为缓冲区，默认大小是8K，在构造时初始化，fill()方法可能会出现需要重新分配一个更大数组的情况。从构造函数和内部变量中，我们可以看到它也具有内部包裹的下层输入流并且需要是Reader及其子类。 /** * 下层输入流 */ private Reader in; /** * 缓冲区 */ private char cb[]; /** * 最后一个有效字符的位置 */ private int nChars; /** * 下一个要读取的字符位置 */ private int nextChar; private static final int INVALIDATED = -2; private static final int UNMARKED = -1; /** * 标记的位置，小于等于-1时代表不存在mark */ private int markedChar = UNMARKED; private int readAheadLimit = 0; /* 只有markedChar > 0时是有效的 */ /** 如果下一个字符是换行符，跳过它 */ private boolean skipLF = false; /** 设定mark时skipLF的标记 */ private boolean markedSkipLF = false; private static int defaultCharBufferSize = 8192; private static int defaultExpectedLineLength = 80; /** * 创建一个缓冲字符输入流使用一个指定大小的输入缓冲区 */ public BufferedReader(Reader in, int sz) { super(in); if (sz <= 0) throw new IllegalArgumentException("Buffer size <= 0"); this.in = in; cb = new char[sz]; nextChar = nChars = 0; } /** * 创建一个缓冲字符输入流使用一个默认大小8K的输入缓冲区 */ public BufferedReader(Reader in) { this(in, defaultCharBufferSize); } ensureOpen这个内部方法的作用就是检查流有没有被关闭，标志是下层输入流为null private void ensureOpen() throws IOException { if (in == null) throw new IOException("Stream closed"); } 无参数的read()读取单个字符，返回的是转为int的字符，返回是0-0xffff，因为存在两个字节的字符，到达文件末尾返回-1。这里存在一种情况，所有提到跳过换行符都是这个意思，上一次读取到的是'r'此时如果当前要读取的是'n'则直接跳过读取下一个字符，这是LRLF代表换行的情况，Windows平台下输入的文本很多是这种换行方式。 public int read() throws IOException { synchronized (lock) { ensureOpen(); for (;;) { if (nextChar >= nChars) { fill();//读取到最后一个有效字符时，从输入流读取字符到缓冲区 if (nextChar >= nChars) return -1; } if (skipLF) {//如果下一个字符是换行符要跳过 skipLF = false; if (cb[nextChar] == '\n') { nextChar++; continue; } } return cb[nextChar++];//从缓冲区返回要读取的字符 } } } 内部方法fill()填充输入缓冲区，如果mark是有效的需要将它加入考虑，只有在可被读取的字符被全部读取完后才会调用这个方法。该类中所有要从下层输入流读取字符到缓冲区都需要经过该方法。没有mark时，会直接将内容从数组头部开始尽可能填满缓冲区。存在mark时，如果从mark开始读取的内容超过了readAheadLimit的限制，mark失效。mark未超过范围时，检查readAheadLimit是否超过了缓冲区大小，没有超过时将mark开始的部分移动到缓冲区头部，然后读取数据尽可能填满缓冲区；mark超过了缓冲区大小，需要分配一个大小等于readAheadLimit的新数组，复制mark开始的内容，然后再读取字符，这是唯一引起分配新数组的方法。所有调用fill()的方法需要保证线程的安全性，我们可以看到这里分配新数组是简单赋值而不是CAS方法。 private void fill() throws IOException { int dst; if (markedChar <= UNMARKED) { /* 没有mark */ dst = 0; } else { /* 存在mark */ int delta = nextChar - markedChar; if (delta >= readAheadLimit) { /* 从mark开始到读取位置的字节数超过了限制，将markedChar设为-2使它失效 */ markedChar = INVALIDATED; readAheadLimit = 0; dst = 0; } else { if (readAheadLimit <= cb.length) {//能够保留的最大mark部分长度超过了缓冲区大小 /* 在当前缓冲区中洗牌 */ System.arraycopy(cb, markedChar, cb, 0, delta);//将从markedChar开始的所有有效字符移到缓冲区头部 markedChar = 0; dst = delta; } else { /* 重新分配缓冲区的大小来满足预读要求 */ char ncb[] = new char[readAheadLimit];//新大小是readAheadLimit System.arraycopy(cb, markedChar, ncb, 0, delta);//将有效字符复制到新的数组 cb = ncb; markedChar = 0;//因为从标记位开始的有效字符被移动到了头上，所以标记位从0开始 dst = delta; } nextChar = nChars = delta; } } int n; do { n = in.read(cb, dst, cb.length - dst);//从下层输入流读取字符 } while (n == 0); if (n > 0) { nChars = dst + n; nextChar = dst; } } int read(char cbuf[], int off, int len)读取字符到数组的一部分中，这个方法实现了Reader.read(char[], int, int)方法的一般约束。作为额外的便利，它会试图读取尽可能多的字符，通过重复调用下层输入流的read方法。这个重复的read会持续到一下一种情况为true：指定的字符数已经被读取，下层输入流的read方法返回-1说明到达了文件结束符，或者下层输入流的ready方法返回false说明后面的输入请求会阻塞。如果第一个下层输入流的read方法返回-1，这个方法会返回-1.否则这个方法返回实际读取的字符数。这个类的子类鼓励但不是必须去尝试用同样的方法读取尽可能多的字符。一般这个方法从这个输入流的字符缓冲区获取字符，如果需要的话从下层输入流来填充缓冲区。但是，如果缓冲区是空的，mark是无效的，请求的长度超过了缓冲区的长度，这个方法会从下层输入流直接读取字符到给定的数组。因此多余的BufferedReader不会进行不必要的复制。 public int read(char cbuf[], int off, int len) throws IOException { synchronized (lock) {//同步操作 ensureOpen(); if ((off < 0) || (off > cbuf.length) || (len < 0) || ((off + len) > cbuf.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return 0; } int n = read1(cbuf, off, len);//从缓冲区读取数据到数组，如果调用时缓冲区就已经没有有效数据从下层输入流读取 if (n <= 0) return n; while ((n < len) && in.ready()) { int n1 = read1(cbuf, off + n, len - n);//循环尝试从下层输入流读取数据到缓冲区 if (n1 <= 0) break;//下层输入流没有数据时返回 n += n1; } return n; } } private int read1(char[] cbuf, int off, int len) throws IOException { if (nextChar >= nChars) { /* If the requested length is at least as large as the buffer, and if there is no mark/reset activity, and if line feeds are not being skipped, do not bother to copy the characters into the local buffer. In this way buffered streams will cascade harmlessly.如果需要的长度超过了缓冲区大小，并且没有mark/reset活动，并且换行没有被跳过 不要麻烦地将字符复制到本地缓冲区。这样可以避免没必要的损耗 */ if (len >= cb.length && markedChar <= UNMARKED && !skipLF) { return in.read(cbuf, off, len); } fill();//读取到最后一个缓冲区内的有效字符时，从下层输入流读取字符到缓冲区 } if (nextChar >= nChars) return -1;//下层输入流没有有效字符时返回-1 if (skipLF) { skipLF = false; if (cb[nextChar] == '\n') { nextChar++;//下一个字符时换行符时跳过 if (nextChar >= nChars)//如果跳过换行后有效字符不足则尝试从下层输入流中读取 fill(); if (nextChar >= nChars) return -1; } } int n = Math.min(len, nChars - nextChar);//读取的字符长度是len和缓冲区内有效字符的较小值 System.arraycopy(cb, nextChar, cbuf, off, n);//从缓冲区复制字符到目标数组中 nextChar += n; return n;//返回实际读取的字符数 } readLine读取一行文本。一行通过一个换行符或者一个回车或者一个回车跟着一个换行符来终止。返回一个包含行内容但是不包含行终止字符的字符串，如果已经到达流结尾的话返回null。 public String readLine() throws IOException { return readLine(false); } String readLine(boolean ignoreLF) throws IOException { StringBuffer s = null; int startChar; synchronized (lock) { ensureOpen(); boolean omitLF = ignoreLF || skipLF; bufferLoop: for (;;) { if (nextChar >= nChars) fill();//缓冲区内没有有效字符时，从下层输入流读取字符到缓冲区 if (nextChar >= nChars) { /* 到达了EOF */ if (s != null && s.length() > 0) return s.toString();//读取到了字符则返回字符串 else return null;//没有读取到字符返回null } boolean eol = false;//缓冲区内有有效数据，还没有到达行结束 char c = 0; int i; /* 如果有必要的话跳过一个 '\n'*/ if (omitLF && (cb[nextChar] == '\n')) nextChar++; skipLF = false;//只有第一个字符是'\n'时才会跳过 omitLF = false; charLoop: for (i = nextChar; i < nChars; i++) {//读取范围不超过缓冲区内的所有有效字符 c = cb[i]; if ((c == '\n') || (c == '\r')) { eol = true;//到达了行结束 break charLoop;//这里跳出了循环，所以nextChar依然在换行符或者回车的位置上 } } startChar = nextChar; nextChar = i; if (eol) {//到达了行结束 String str; if (s == null) {//将新读取的内容添加到返回结果中 str = new String(cb, startChar, i - startChar); } else { s.append(cb, startChar, i - startChar); str = s.toString(); } nextChar++;//此时nextChar向前移动一位，经过了之前的换行符或回车 if (c == '\r') {//如果行结束标记是回车，跳过下一个换行符对应LRLF这种情况 skipLF = true; } return str; } if (s == null) s = new StringBuffer(defaultExpectedLineLength); s.append(cb, startChar, i - startChar);//没有到达行结尾时，直接将新增内容添加到返回的字符串中 } } } skip跳过字符的操作是通过读取字符到缓冲区然后改变缓冲区的当前读取位置下标nextChar实现的，跳过的字符数为参数n和缓冲区内剩余有效字符的较小值 public long skip(long n) throws IOException { if (n < 0L) { throw new IllegalArgumentException("skip value is negative"); } synchronized (lock) { ensureOpen(); long r = n; while (r > 0) { if (nextChar >= nChars) fill();//缓冲区内没有有效字符时尝试从下层输入流读取字符到缓冲区 if (nextChar >= nChars) /* 到达了EOF */ break; if (skipLF) { skipLF = false; if (cb[nextChar] == '\n') { nextChar++;//调用时第一个有效字符时'\n'时跳过它 } } long d = nChars - nextChar;//当前缓冲区内的有效字符个数 if (r <= d) {//缓冲区内的有效字符数超过了要跳过的字符数，直接增加nextChar nextChar += r; r = 0; break; } else {//缓冲区内数据不足时，nextChar移动到最后一个有效字符的位置 r -= d; nextChar = nChars; } } return n - r;//返回跳过的字符数 } } ready告知这个流是否准备完读取。一个字符缓冲流在缓冲区非空或者下层输入流准备完时是准备完成的状态。 public boolean ready() throws IOException { synchronized (lock) { ensureOpen(); /* * 如果一个新行需要跳过并且下一个要读取的字符是新行的字符，立刻跳过它 */ if (skipLF) { /* * in.ready()仅当流中的下一个读取不会被阻塞时会返回true */ if (nextChar >= nChars && in.ready()) { fill();//缓冲区没有有效数据且下层输入流准备完时进行读取 } if (nextChar < nChars) { if (cb[nextChar] == '\n')//如果缓冲区内第一个有效字符是'\n'则跳过 nextChar++; skipLF = false; } } return (nextChar < nChars) || in.ready(); } } mark/reset操作只需要操作内部变量就行了，另外readAheadLimit是在mark时输入的，当这个限制超过了缓冲区大小时会引起新分配一个不小于这个大小的缓冲区，因此要小心输入一个大值。 public void mark(int readAheadLimit) throws IOException { if (readAheadLimit < 0) { throw new IllegalArgumentException("Read-ahead limit < 0"); } synchronized (lock) { ensureOpen(); this.readAheadLimit = readAheadLimit; markedChar = nextChar; markedSkipLF = skipLF; } } public void reset() throws IOException { synchronized (lock) { ensureOpen(); if (markedChar < 0) throw new IOException((markedChar == INVALIDATED) ? "Mark invalid" : "Stream not marked"); nextChar = markedChar; skipLF = markedSkipLF; } } close方法会关闭下层输入流并将in和cb设为null，所以关闭之后所有操作都会抛出异常，即使只涉及缓冲区，因为缓冲区被清除了。 public void close() throws IOException { synchronized (lock) { if (in == null) return; try { in.close(); } finally { in = null; cb = null; } } } 最后lines()方法是JDK1.8引入了流式操作，具体流式操作的内容以后再讲吧，这里就翻译下注释：返回一个流，元素是从这个BufferedReader读取的行。流是延迟构成的，比如读取只发生在终止流操作中。reader在执行终止流操作期间不能被操作，否则，终止流的结果会不确定。如果访问下层BufferedReader时抛出了IOException，它会被包装为UncheckedIOException由Stream抛出。这个方法如果调用在一个关闭的BufferedReader上会返回stream，任何请求从关闭后的BufferedReader中读取的请求都会引起抛出UncheckedIOException。 BufferedWriter BufferedWriter是对应的缓冲字符输出流。将文本写到字符输出流，缓冲字符来提供单个字符、数组、字符串的高效写入。缓冲区的大小可能是指定的，或者默认大小8K，默认大小对于大多数情况足够大。 newLine()方法使用平台定义的行分隔符line.separator。不是所有平台使用'n'来表示行结束。调用这个方法来结束每个输入行比直接写一个'n'字符更好。 一般来说，一个Writer发送它的输出直接到下层的字符流或者字节流。除非要求立刻输出，否则建议用BufferedWriter来包装一个write()操作花费大的Writer，比如FileWriters和OutputStreamWriters。例如，PrintWriter out = new PrintWriter(new BufferedWriter(new FileWriter("foo.out")));会缓冲PrintWriter到一个文件的输出。没有缓冲，每一次调用print()方法会引起字符被转换成字节然后直接写入到文件，这样效率低。 含有一个底层输入流Writer，在构造时提供，并且Writer的锁对象会是它自身。 /** * 下层输出流 */ private Writer out; /** * 缓冲区 */ private char cb[]; /** * 缓冲区大小 */ private int nChars; /** * 下一个有效字符 */ private int nextChar; private static int defaultCharBufferSize = 8192; /** * 行分隔符，在这个流创建时行分隔符的值 */ private String lineSeparator; /** * 创建一个使用默认大小8K输出缓冲区的缓冲字符输出流 */ public BufferedWriter(Writer out) { this(out, defaultCharBufferSize); } /** * 创建一个使用指定大小sz输出缓冲区的缓冲字符输出流 */ public BufferedWriter(Writer out, int sz) { super(out); if (sz <= 0) throw new IllegalArgumentException("Buffer size <= 0"); this.out = out; cb = new char[sz]; nChars = sz; nextChar = 0; lineSeparator = java.security.AccessController.doPrivileged( new sun.security.action.GetPropertyAction("line.separator")); } ensureOpen内部方法用于确认这个流是否被关闭，标志是out为null private void ensureOpen() throws IOException { if (out == null) throw new IOException("Stream closed"); } write(int c)写入单个字符，在缓冲区满了时需要将缓冲区内的字符写入到下层输入流 public void write(int c) throws IOException { synchronized (lock) { ensureOpen(); if (nextChar >= nChars) flushBuffer();//如果缓冲区已满，将缓冲区内的字符写入到下层输出流 cb[nextChar++] = (char) c;//将c复制到数组中 } } flushBuffer将这个输出缓冲区刷新到下层字符流中，但不刷新下层流本身。这个类不是private，所以可以被PrintStream调用 void flushBuffer() throws IOException { synchronized (lock) { ensureOpen(); if (nextChar == 0) return; out.write(cb, 0, nextChar);//下层输出流输出缓冲区内的所有有效字符 nextChar = 0; } } write(char cbuf[], int off, int len)写入一个字符数组的一部分。一般来说这个方法从数组存储字符到流的缓冲区，如果需要的话刷新缓冲区到下层输出流。如果请求的长度超过了缓冲区大小，这个方法会刷新缓冲区并将字符直接写入到下层输出流。因此多余的BufferedWriter不会不必要的复制数据。 public void write(char cbuf[], int off, int len) throws IOException { synchronized (lock) { ensureOpen(); if ((off < 0) || (off > cbuf.length) || (len < 0) || ((off + len) > cbuf.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return; } if (len >= nChars) { /* 如果请求长度超过了输出缓冲区大小，刷新缓冲区直接写数据。这样缓冲输出流是级联无害的 */ flushBuffer(); out.write(cbuf, off, len);//直接使用下层输出流的write方法 return; } int b = off, t = off + len; while (b < t) { int d = min(nChars - nextChar, t - b); System.arraycopy(cbuf, b, cb, nextChar, d);//复制字符到缓冲区，长度为缓冲区剩余空间和剩余要输入字符的较小值 b += d; nextChar += d; if (nextChar >= nChars) flushBuffer();//缓冲区满了时刷新 } } } write(String s, int off, int len)写一个字符串的一部分，一定会经过复制到缓冲区这个过程。需要注意的是，如果len参数是负数，没有字符会被写入。这个和java.io.Writer.write(java.lang.String,int,int)超类是相反的，它要求抛出IndexOutOfBoundsException public void write(String s, int off, int len) throws IOException { synchronized (lock) { ensureOpen(); int b = off, t = off + len; while (b < t) { int d = min(nChars - nextChar, t - b); s.getChars(b, b + d, cb, nextChar); b += d; nextChar += d; if (nextChar >= nChars) flushBuffer(); } } } 在上面的写入方法中，我们注意到有个min方法来获得两个数中的最小值，这里的min是自身的内部方法，避免加载java.lang.Math在用尽文件描述符和尝试打印堆栈追踪时 private int min(int a, int b) { if (a < b) return a; return b; } newLine()写一个行分隔符，行分隔符通过系统属性line.separator来定义，不一定是一个'n'字符 public void newLine() throws IOException { write(lineSeparator); } flush刷新流，同时会刷新下层输入流 public void flush() throws IOException { synchronized (lock) { flushBuffer(); out.flush();//下层输入流的刷新在这里调用 } } close关闭这个输出流和下层输出流，重复关闭没有效果。关闭后由于out和cb都为null，所以一切写入操作都会抛出异常。这里关闭out也是使用了JDK7引入的try-with-resources写法 public void close() throws IOException { synchronized (lock) { if (out == null) {//out为null说明已经被关闭了 return; } try (Writer w = out) {//out随着这个代码块结束而关闭 flushBuffer();//先做一次刷新，避免有没有输入到下层输出流的数据 } finally { out = null; cb = null;//缓冲区被移除 } } }
2018-09-04
优秀的个人博客,低调大师

Java BufferedInputStream BufferedOutputStream类源码解析

BufferedInputStream ​ BufferedInputStream是一个缓冲输入流，继承的是FilterInputStream。FilterInputStream包含了另一个InputStream作为它的基础数据源，并且FilterInputStream重写了InputStream的所有方法。作为FilterInputStream需要重写其中的部分方法，如果没有重写的话默认调用InputStream的同名方法。 ​ 对于BufferedInputStream来说，常规的用法是以FileInputStream作为它的下层输入流，原因是FileInputStream是从文件中读取字节，而文件处于硬盘之中，每次从硬盘读取数据的速度相对于内存很慢。如果多次通过FileInputStream读取或跳过一段字符，就需要多次访问硬盘，如果能够通过一次访问将整块内容读取到内存中，再从内存中多次读取，效率就可以大幅度提高，这就是内存缓冲区。举个极端例子来说，连续调用FileInputStream.read()一千次，得到文件中一千个连续字节，访问硬盘1000次。如果使用FileInputStream.read(buff[1000])，然后从buff[]中读取1000个字节，只需要访问硬盘1-2（跨页）次，访问内存1000次，毫无疑问由于内存对硬盘的数量级速度优势是后者快得多。 ​ BufferedInputStream对另一个输出流增加了功能，支持缓存输入和mark/reset操作。当创建BufferedInputStream时，一个内部缓冲数组被建立，随着字节从流中被读取或者跳过，内部缓冲区根据需要从包含的输入流中重新装填许多字节。mark操作记录了输入流中的一个位置，reset操作引起在最近一个mark操作前的所有读取的字节在流中新的字节被获取之前被重新读取。涉及数据的方法是线程同步的。 先来看一下内部变量，可以看到BufferedInputStream新增了很多内部变量来支持缓存以及mark/reset操作。buf数组作为缓冲区，其中的有效字节数是count，当前读取到的位置是pos，所以count-pos是剩余可以从缓冲区内直接读取的字节数。markpos是标记的位置，-1代表没有标记过，pos-markpos是标记之后的字节数，该值不能超过marklimit，超过的话会导致markpos被设为-1也就是mark内容被丢弃。 private static int DEFAULT_BUFFER_SIZE = 8192;//默认大小8K /** * 分配的最大数组大小。一些虚拟机保留了数组中一些头部字。 * 试图分配更大的数组可能导致OutOfMemoryError：请求的数组大小超过了虚拟机限制 */ private static int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8; /** * 内部缓冲数组存储数据，必要时，它会被替换为另一个不同大小的数组 */ protected volatile byte 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; 构造函数需要提供输入流，通常是FileInputStream，可选参数是缓冲区初始大小，该值不能为负数，不提供时使用默认值8K /** * 创建一个BufferedInputStream并存储它的参数供之后使用，输入流是in。一个大小为8K内部的缓冲数组被建立存储在buf */ public BufferedInputStream(InputStream in) { this(in, DEFAULT_BUFFER_SIZE); } /** * 创建一个指定缓冲区大小的BufferedInputStream并存储它的参数供之后使用，输入流是in。一个大小为size的内部的缓冲数组被建立存储在buf */ public BufferedInputStream(InputStream in, int size) { super(in); if (size <= 0) { throw new IllegalArgumentException("Buffer size <= 0"); } buf = new byte[size]; } BufferedInputStream中的方法除了close和markSupported外都需要保证流是打开的，所以有两个内部方法负责检查状态。调用方法在发现返回值为null时即表示流已经关闭。 /** * 检查确认下层输入流还没有因为关闭变成null，不是null的话返回输入流 */ private InputStream getInIfOpen() throws IOException { InputStream input = in; if (input == null) throw new IOException("Stream closed"); return input; } /** * 检查确认缓冲区还没有因为关闭变成null，非null的话返回它 */ private byte[] getBufIfOpen() throws IOException { byte[] buffer = buf; if (buffer == null) throw new IOException("Stream closed"); return buffer; } 下面分析read方法，先从读取单个字节开始，可以看到它从方法级别加了同步锁。如果缓冲区内的数据足够则直接返回数组中的元素，否则需要先通过fill()将下层输入流中的数据读取到缓冲区中，如果流中也没有数据了则返回-1。 public synchronized int read() throws IOException { if (pos >= count) {//pos>=count说明缓冲区内没有可读取的数据，需要从流中读取数据到缓冲区 fill(); if (pos >= count)//流中也没有数据可读取了 return -1; } return getBufIfOpen()[pos++] & 0xff; } 然后来看一下fill方法，它会从下层输入流中读取字节填满缓冲区直到输入流中也没有有效的字节。如果当前没有标记，说明缓冲区不再需要存储已经读取过的内容，可以直接情况缓冲区。如果pos达到了缓冲区大小上限，此时如果markpos>0则丢弃缓冲区中markpos之前的部分；如果markpos等于0且缓冲区大小超过了marklimit需要丢弃标记和缓冲区内容；如果缓冲区大小达到了最大限制抛出OOM错误；如果缓冲区还没有达到大小限制，通过分配新数组再复制将缓冲区大小扩大两倍但不能超过最大限制，扩大仅在标记位置为0且读取到缓冲区最后一个字节，缓冲区大小没有超过marklimit和MAX_BUFFER_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);//被丢弃的是从0到markpos之间的数据 pos = sz; markpos = 0;//标记位置设为0 } else if (buffer.length >= marklimit) { markpos = -1; /* buffer got too big, invalidate mark buffer太大，无效标记缓冲区*/ pos = 0; /* drop buffer contents 丢弃缓冲区内容*/ } else if (buffer.length >= MAX_BUFFER_SIZE) { throw new OutOfMemoryError("Required array size too large"); } else { /* grow buffer buffer增长*/ int nsz = (pos <= MAX_BUFFER_SIZE - pos) ? pos * 2 : MAX_BUFFER_SIZE;//pos大小乘以2除非超过最大上限 if (nsz > marklimit) nsz = marklimit;//增长后的大小不能超过marklimit byte nbuf[] = new byte[nsz]; System.arraycopy(buffer, 0, nbuf, 0, pos);//将原buffer中的内容复制到新的数组中 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.如果fill()曾被设置为多线程可进入，这里需要改变 // But for now, the only way CAS can fail is via close. // assert buf == null;但是现在，CAS唯一失败的原因是关闭，断言buf==null throw new IOException("Stream closed"); } buffer = nbuf; } count = pos;//从开始pos到count都是有效字符 int n = getInIfOpen().read(buffer, pos, buffer.length - pos);//从输入流中读取数据填满缓冲区 if (n > 0) count = n + pos;//读取到数据则增加count } int read(byte b[], int off, int len)从这个字节输入流读取字节到指定的字节数组中，从给出的偏移量开始。这个方法实现了InputStream.read(byte[], int, int)抽象方法。作为额外的便利，这个方法尝试读取尽可能多的字节，通过重复调用下层输入流的read方法。这个重复的read会一直持续直到以下几种情况：读取到了指定数量的字节；下层输入流read方法返回-1说明到达文件末尾；下层输入流的available方法返回0说明后续的读取请求会被阻塞。如果下层输入流的第一次read方法返回-1说明到达文件末尾，这个方法会返回-1，否则返回实际读取的字节数。 public synchronized int read(byte b[], int off, int len) throws IOException { getBufIfOpen(); // 检查流有没有关闭 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;//什么都没读取到时返回-1 n += nread; if (n >= len) return n;//达到需要读取的字节数，返回实际读取到的字节，正常情况下等于len // 如果没有关闭但是没有可读取的字节，返回 InputStream input = in; if (input != null && input.available() <= 0)//下层输入流没有可以读取的字节 return n;//返回读取到的字节数，该值小于len } } 上面的方法调用了内部方法read1，它将字符读入到数组的一部分，如果需要的话最多从下层输入流读取一次。先检查缓冲区内有没有数据，若没有数据且len超过了缓冲区大小上限并且缓冲区内没有标记，则直接从输入流读取到数组b，不再经过缓冲区并返回；否则从输入流读取数据填满缓冲区。除非已经返回，否则随后会从缓冲区复制数据到数组b，长度为len和缓冲区内剩余字节数的较小值。也就是说，read1会在不超过len和缓冲区上限的情况下读取尽可能多的字节。 private int read1(byte[] b, int off, int len) throws IOException { int avail = count - pos;//缓冲区内有效的字节数 if (avail <= 0) {//缓冲区内没有数据了 /* 如果需要的长度至少跟buffer一样大，并且没有mark/reset活动，不会打扰复制字节到本地缓冲区。这样缓冲流是级联无害的 */ if (len >= getBufIfOpen().length && markpos < 0) {//流超过了缓冲区的数组长度且不存在标记 return getInIfOpen().read(b, off, len);//从输入流中直接读取字节复制到数组b } fill();//从输入流中读取数据到缓冲区，在缓冲区数据被全部读取到b中且未超过marklimit的时候，会扩大缓冲区 avail = count - pos; if (avail <= 0) return -1;//输入流中也没有数据了，返回-1 } int cnt = (avail < len) ? avail : len; System.arraycopy(getBufIfOpen(), pos, b, off, cnt);//将缓冲区内的字节复制到数组b，数据量为len和缓冲区内剩余字节数的较小值 pos += cnt; return cnt;//返回读取到数组b的字节数 } skip操作跳过指定数量的字节除非达到文件末尾。如果存在标记或者缓冲区存在有效内容则通过读取字节到缓冲区在设置pos的值来完成跳过，这个跳过的数量不能超过缓冲区的有效字节数。如果没有标记也没有缓冲区有效数据，直接通过下层输入流的skip来跳过，对于FileInputStream是通过底层方法IO_Lseek。 public synchronized long skip(long n) throws IOException { getBufIfOpen(); // 检查流是否关闭 if (n <= 0) { return 0; } long avail = count - pos; if (avail <= 0) { // 如果没有标记则不再保留buffer if (markpos <0) return getInIfOpen().skip(n);//调用了下层输入流的skip方法 // 填充缓冲区，保存用于reset的字节 fill(); avail = count - pos; if (avail <= 0) return 0; } long skipped = (avail < n) ? avail : n;//跳过的字节数最大不能超过缓冲区的可填充字节数 pos += skipped;//通过读取到缓冲区并设置pos的位置来skip return skipped;//返回实际跳过的字节 } available()返回从输入流中不需要因为下一个操作阻塞的可以读取或者跳过的字节数的估计值。下一个操作可能由这个线程或者其他线程来调用。一个单位的读取或者跳过操作不会阻塞，但是可能读取或跳过更少的字节。这个方法返回缓冲区内剩余可读取的字节数count-pos加上FilterInputStream.available() 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;//不超过Interger范围时，返回值是缓冲区的有效数据加上输入流中的有效数据量 } mark/reset相关方法前面基本提到了，是通过直接修改pos的值来进行的 public synchronized void mark(int readlimit) { marklimit = readlimit;//marklimit设为给出的参数 markpos = pos;//标记当前位置 } /** * markpos是-1时没有标记，抛出IOException。否则pos=markpos */ public synchronized void reset() throws IOException { getBufIfOpen(); // 如果关闭了会引起异常 if (markpos < 0) throw new IOException("Resetting to invalid mark"); pos = markpos; } /** * 检查这个输入流是否支持mark/reset操作，BufferedInputStream总是返回true */ public boolean markSupported() { return true; } close关闭这个输入流并释放任何关联的系统资源。一旦流被关闭，后面的read(), available(), reset(), skip()调用都会抛出IOException。关闭一个已经关闭的流没有作用 public void close() throws IOException { byte[] buffer; while ( (buffer = buf) != null) { if (bufUpdater.compareAndSet(this, buffer, null)) {//将buf设为null InputStream input = in; in = null;//将in设为null if (input != null) input.close();//关闭下层输入流 return; } // 或者一个新的buf在fill()中更新时重试 } } 最后，简单讲一下前面提到的CAS，compareAndSet简单来说就是先获取内存中的某个值，比较是否与给出的原值相等，如果相等的话尝试将这个值更新为目标值，再检查内存中的值是否变为目标值，成功则返回，任何一个环节失败则直接返回false。这里主要涉及到的问题是多线程之间的不同步问题，CAS一定会保证这次操作是原子性的，并且这个操作是无锁算法所以比起synchronized在碰撞较少时更加高效。总之，CAS的作用是确保方法是线程安全的，对于BufferedInputStream来说，唯一可能造成线程不安全的地方是在其他方法没有进行完时进行关闭。对于close来说，它在发现有其他线程在通过fill修改buf时，会检测到buffer不相等或者修改后不是null，会继续循环尝试关闭。而对于fill方法，如果检查到冲突会直接抛出IO异常，结束这次操作。 /** * 原子更新操作为缓冲区提供CAS操作。这是必要的，因为关闭是异步的。 * 我们使用null的buf[]作为流关闭的主要指示器，输入关闭时输入区域也是null */ private static final AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater = AtomicReferenceFieldUpdater.newUpdater (BufferedInputStream.class, byte[].class, "buf"); BufferedOutputStream BufferedOutputStream继承了FilterOutputStream，FilterOutputStream是所有过滤输出流类的超级父类，它含有一个下层的输出流，并且简单重写了OutputStream的全部方法。 跟缓冲输入流相对应，BufferedOutputStream实现了一个缓冲输出流，通过设置这样的输出流，应用可以在写入字节到下层输出流时不需要在写每个字节都调用一次下层系统，它通常以FileOutputStream作为下层输出流，通过一个在构造时分配的字节数组作为缓冲区来存储数据，避免每次write都直接涉及到底层的写入操作。前面讲到过，直接读写硬盘的速度和内存读取速度差距极大，所以通过在内存中缓存数据再一口气按块写入可以提高写入到硬盘的速度。该类的刷新和写操作是方法级加锁同步的。 BufferedOutputStream的内部变量只有缓冲区buf[]，记录有效字节数量的count和继承自父类的下层输出流out /** * 存储数据的内部缓冲区 */ protected byte buf[]; /** * 缓冲区中有效字节的数量。这个值得范围总是从0到buf.length，元素buf[0]到buf[count-1]含有有效的字节数据 */ protected int count; protected OutputStream out; 构造函数必须提供OutputStream，一般是FileOutputStream，可选参数是数组大小，数组在构造时进行分配，之后不会再重新分配 /** * 创建一个新的缓冲输出流来写入数据到指定的下层输出流当中 */ public BufferedOutputStream(OutputStream out) { this(out, 8192);//默认缓冲区大小是8K } /** * 创建一个新的缓冲输出流来写入数据到指定的下层输出流当中，指定它的缓冲区大小 */ public BufferedOutputStream(OutputStream out, int size) { super(out); if (size <= 0) { throw new IllegalArgumentException("Buffer size <= 0");//size必须大于0 } buf = new byte[size]; } write写单个字节先检查缓冲区有没有满，满了先将缓冲区内的数据全部写入到输出流中，然后将要写的数据存储到缓冲区数组中。而从指定的字节数组中写从off偏移开始len长度的字节到这个缓冲输出流中时，一般来说这个方法存储给出数组中的字节到这个流的缓冲区，必要的时候刷新缓冲区到下层输出流。 如果请求的长度大于鞥与缓冲区大小，这个方法会刷新缓冲区，然后将字节直接写入到下层输出流中。这样多余的BufferedOutputStream就不会无必要的复制数据。 public synchronized void write(int b) throws IOException { if (count >= buf.length) { flushBuffer();//如果缓冲区满了则将缓冲区内的数据全部写入到输出流中 } buf[count++] = (byte)b;//将b存储到缓冲区中 } public synchronized void write(byte b[], int off, int len) throws IOException { if (len >= buf.length) { /* 如果请求的长度超出了输出缓冲区的大小，刷新输出缓冲区然后直接写数据。这样缓冲流可以无损害地流动*/ flushBuffer(); out.write(b, off, len);//直接调用下层输出流的write方法 return; } if (len > buf.length - count) {//要输出的字节长度超过了剩余缓冲区大小则先刷新清空缓冲区 flushBuffer(); } System.arraycopy(b, off, buf, count, len);//将要输出的内容存储到缓冲区 count += len; } flushBuffer这个内部方法的作用是刷新内部缓冲区，调用下层输出流的write方法将缓冲区内的数据全部写入 private void flushBuffer() throws IOException { if (count > 0) {//如果缓冲区内存在有效数据 out.write(buf, 0, count);//将缓冲区内的有效数据全部写入到输出流 count = 0; } } flush刷新这个缓冲输出流，这个命令任何的缓冲区中的输出字节写出到底层的输出流中 public synchronized void flush() throws IOException { flushBuffer();//将缓冲区内的数据全部写入到输出流中 out.flush();//触发下层输出流的刷新操作，对于FileOutputStream的话是什么也不做 } 最后close方法直接调用的父类的close，刷新一次之后直接关闭下层输出流，这里的写法是JDK7之后的try-with-resource的写法，流只在try的代码块内打开，结束后自动关闭。 public void close() throws IOException { try (OutputStream ostream = out) { flush(); } } 在关闭之后，仅对BufferedOutputStream中缓冲区的操作并不会报错，只有涉及到下层输出流才会报错，此时写入到缓冲区的数据是无法写入到文件的 public static void main(String args[]) throws IOException{ BufferedOutputStream bout = new BufferedOutputStream(new FileOutputStream("D:/test/file.txt")); bout.write("1234567890".getBytes());//1234567890 bout.close(); bout.write("1234567890".getBytes());//这里写入的数据在缓冲区的数组中所以不报错 bout.flush();//这里会报错 }
2018-08-31
优秀的个人博客,低调大师

Java FileWriter OutputStreamWriter类源码解析

FileWriter 因为篇幅原因，上一篇直接了字符输入流，今天来分析一下跟FileReader相对应的字符输出流FileWriter。FileWriter是将字符写入文件的通用类，构造函数假定使用默认的字符编码和默认的字节缓冲区大小8K是使用者可以接受的，如果要指定这些值，需要通过一个FileOutputStream来构造FileWriter的父类OutputStreamWriter。 文件是否有效或者是否能够被创建取决于平台，在一些平台上，对于同一个文件同一时间只允许一个FileWriter或者其他文件写入对象打开。在这种情况下，如果一个文件已经被打开，构造函数会抛出异常。 和FileReader类似，FileWriter也是除了构造函数以外全部是继承了父类的方法。先创建一个FileOutputStream，如果不给出append参数或者append为false则清空原文件从头开始写入，否则是从尾部开始扩展文件内容，使用文件描述符创建时必定是从文件头部开始写。然后通过FileOutputStream创建OutputStreamWriter public FileWriter(String fileName) throws IOException { super(new FileOutputStream(fileName)); } public FileWriter(String fileName, boolean append) throws IOException { super(new FileOutputStream(fileName, append)); } public FileWriter(File file) throws IOException { super(new FileOutputStream(file)); } public FileWriter(File file, boolean append) throws IOException { super(new FileOutputStream(file, append)); } public FileWriter(FileDescriptor fd) { super(new FileOutputStream(fd)); } OutputStreamWriter OutputStreamWriter继承了抽象类Writer，跟OutputStreamReader类似，主要重写的方法都是基于StreamEncoder来完成，StreamEncoder在构造函数中通过工厂方法构造。 在构造函数中存在super(out)，作用是构造父类Writer，将OutputStream作为加锁的对象 //指定字符集名字 public OutputStreamWriter(OutputStream out, String charsetName) throws UnsupportedEncodingException { super(out); if (charsetName == null) throw new NullPointerException("charsetName"); se = StreamEncoder.forOutputStreamWriter(out, this, charsetName);//传给StreamEncoder的加锁对象是OutputStreamWriter对象自身 } //使用默认字符集 public OutputStreamWriter(OutputStream out) { super(out); try { se = StreamEncoder.forOutputStreamWriter(out, this, (String)null); } catch (UnsupportedEncodingException e) { throw new Error(e); } } //使用指定的字符集 public OutputStreamWriter(OutputStream out, Charset cs) { super(out); if (cs == null) throw new NullPointerException("charset"); se = StreamEncoder.forOutputStreamWriter(out, this, cs); } //使用指定的CharsetEncoder public OutputStreamWriter(OutputStream out, CharsetEncoder enc) { super(out); if (enc == null) throw new NullPointerException("charset encoder"); se = StreamEncoder.forOutputStreamWriter(out, this, enc); } OutputStreamWriter重写了Writer中的write, flush, close，此外Writer中的append方法最终也是基于write来完成的，这些方法都是直接调用StreamEncoder中的对应方法。 //写入单个字符 public void write(int c) throws IOException { se.write(c); } //写入一部分字符数组 public void write(char cbuf[], int off, int len) throws IOException { se.write(cbuf, off, len); } //写入一部分字符串 public void write(String str, int off, int len) throws IOException { se.write(str, off, len); } //刷新 public void flush() throws IOException { se.flush(); } //关闭输出流 public void close() throws IOException { se.close(); } getEncoding和flushBuffer是OutputStreamWriter相比于Writer新增的两个调用StreamEncoder中的中间接口 //返回字符集的名称 public String getEncoding() { return se.getEncoding(); } //只有PrintStream这样的类可以调用这个方法，将输出缓冲区的内容刷新到字节流中，但是字节流不会刷新到文件中 void flushBuffer() throws IOException { se.flushBuffer(); } StreamEncoder 跟上一篇一样，要分析字符输出流关键还是要分析StreamEncoder，很多方法可以对照StreamDecoder进行比较，它们从设计上除了输入和输出外是相近的。StreamEncoder同样一次至少操作两个字符，避免出现代替对，也就是2个字节码表示一个字符的特殊情况，当然前面提到过，这种通常是一些机器或者数学上的特殊符号，键盘输入是不会出现的。StreamEncoder继承了抽象类Writer，并重写了其中的write、flush、close方法。 StreamEncoder的构造函数也是private函数，外部只能通过工厂方法类调用 private Charset cs; private CharsetEncoder encoder; private ByteBuffer bb; // Exactly one of these is non-null至少有一个不为null private final OutputStream out; private WritableByteChannel ch; // Leftover first char in a surrogate pair代理对中剩下的第一个字符 private boolean haveLeftoverChar = false; private char leftoverChar; private CharBuffer lcb = null; private StreamEncoder(OutputStream out, Object lock, Charset cs) { this(out, lock, cs.newEncoder().onMalformedInput(CodingErrorAction.REPLACE)// 有畸形输入错误时解码器丢弃错误的输入，替换为替代值然后继续后面的操作 .onUnmappableCharacter(CodingErrorAction.REPLACE));// 有不可用图形表示的字符错误出现时解码器丢弃错误的输入，替换为替代值然后继续后面的操作 } private StreamEncoder(OutputStream out, Object lock, CharsetEncoder enc) { super(lock);// lock是OutputStream对象本身 this.out = out; this.ch = null; this.cs = enc.charset(); this.encoder = enc; // 在堆外内存速度更快之前不使用这段代码 if (false && out instanceof FileOutputStream) { ch = ((FileOutputStream) out).getChannel(); if (ch != null) bb = ByteBuffer.allocateDirect(DEFAULT_BYTE_BUFFER_SIZE); } if (ch == null) { bb = ByteBuffer.allocate(DEFAULT_BYTE_BUFFER_SIZE);//分配一个8K的堆内ByteBuffer } } private StreamEncoder(WritableByteChannel ch, CharsetEncoder enc, int mbc) { this.out = null; this.ch = ch; this.cs = enc.charset(); this.encoder = enc; this.bb = ByteBuffer.allocate(mbc < 0 ? DEFAULT_BYTE_BUFFER_SIZE : mbc);//分配一个大小mbc的堆内ByteBuffer } 在看工厂方法，这里传入的lock对象是OutputStreamWriter本身 // java.io.OutputStreamWriter工厂模式 public static StreamEncoder forOutputStreamWriter(OutputStream out, Object lock, String charsetName) throws UnsupportedEncodingException { String csn = charsetName; if (csn == null) csn = Charset.defaultCharset().name(); try { if (Charset.isSupported(csn)) return new StreamEncoder(out, lock, Charset.forName(csn)); } catch (IllegalCharsetNameException x) { } throw new UnsupportedEncodingException(csn); } public static StreamEncoder forOutputStreamWriter(OutputStream out, Object lock, Charset cs) { return new StreamEncoder(out, lock, cs); } public static StreamEncoder forOutputStreamWriter(OutputStream out, Object lock, CharsetEncoder enc) { return new StreamEncoder(out, lock, enc); } // java.nio.channels.Channels.newWriter工厂模式 public static StreamEncoder forEncoder(WritableByteChannel ch, CharsetEncoder enc, int minBufferCap) { return new StreamEncoder(ch, enc, minBufferCap); } 内部属性isOpen标记流当前是否打开，在关闭时会被设置为close，实际操作方法都会先检查流是否开启，否则会抛出异常 private volatile boolean isOpen = true; private void ensureOpen() throws IOException { if (!isOpen) throw new IOException("Stream closed"); } private boolean isOpen() { return isOpen; } getEncoding返回字符集的历史名，没有的话返回官方名，这里的名字可能和构造时传入的有不同 public String getEncoding() { if (isOpen()) return encodingName(); return null; } String encodingName() { return ((cs instanceof HistoricallyNamedCharset) ? ((HistoricallyNamedCharset) cs).historicalName() : cs.name()); } 接下来分析write部分，通过将字符编码为字节放入ByteBuffer中，然后通过文件通道或者输出流进行输出 public void write(int c) throws IOException { char cbuf[] = new char[1]; cbuf[0] = (char) c; write(cbuf, 0, 1); } //这个是实际调用写入的方法 public void write(char cbuf[], int off, int len) throws IOException { synchronized (lock) { ensureOpen(); if ((off < 0) || (off > cbuf.length) || (len < 0) || ((off + len) > cbuf.length) || ((off + len) < 0)) { throw new IndexOutOfBoundsException(); } else if (len == 0) { return; } implWrite(cbuf, off, len); } } public void write(String str, int off, int len) throws IOException { /* 创建字符缓冲区前检查长度 */ if (len < 0) throw new IndexOutOfBoundsException(); char cbuf[] = new char[len]; str.getChars(off, off + len, cbuf, 0);// 将str中的value从off开始长度len的内容复制到cbuf中 write(cbuf, 0, len); } void implWrite(char cbuf[], int off, int len) throws IOException { CharBuffer cb = CharBuffer.wrap(cbuf, off, len);// 将字符数组组装成一个堆内CharBuffer，数组中的内容不存在复制 if (haveLeftoverChar) flushLeftoverChar(cb, false);//如果有的话，将leftoverChar写入输出流 while (cb.hasRemaining()) { CoderResult cr = encoder.encode(cb, bb, false);//将字符编码为二进制字节直到ByteBuffer满或者CharBuffer中没有更多内容 if (cr.isUnderflow()) {//ByteBuffer没有满，说明CharBuffer内的内容全部编码完成 assert (cb.remaining() <= 1) : cb.remaining(); if (cb.remaining() == 1) { //如果当前缓冲区仅剩一个字符，保存到leftoverChar并修改haveLeftoverChar状态，结束输出 haveLeftoverChar = true; leftoverChar = cb.get(); } break; } if (cr.isOverflow()) {//ByteBuffer满了 assert bb.position() > 0; writeBytes();//将ByteBuffer的内容写入到输出流里面 continue; } cr.throwException(); } } private void writeBytes() throws IOException { bb.flip();//ByteBuffer准备输出当前内容，将limit设为当前位置，pos设为0 int lim = bb.limit(); int pos = bb.position(); assert (pos <= lim); int rem = (pos <= lim ? lim - pos : 0); if (rem > 0) {//输出ByteBuffer中全部内容 if (ch != null) { if (ch.write(bb) != rem) assert false : rem; } else { out.write(bb.array(), bb.arrayOffset() + pos, rem); } } bb.clear();//清空ByteBuffer } implWrite调用了内部方法flushLeftoverChar，作用是将缓存的字符写入输出流，这个方法同时在close时也被调用，因为leftoverChar只有一个字符，而一次输出至少是两个字符，所以还要从cb中读取字符，保证写入的是2个字符 private void flushLeftoverChar(CharBuffer cb, boolean endOfInput) throws IOException { if (!haveLeftoverChar && !endOfInput) return; if (lcb == null)//lcb内一开始是空的 lcb = CharBuffer.allocate(2); else lcb.clear(); if (haveLeftoverChar) lcb.put(leftoverChar);//将leftoverChar放入lcb中 if ((cb != null) && cb.hasRemaining()) lcb.put(cb.get());//cb的内容复制到lcb中 lcb.flip();//将limit设为当前位置，pos设为0，所以现在lcb要输出的内容就是刚才从leftoverChar（如果有的话）和cb里读入的 while (lcb.hasRemaining() || endOfInput) { CoderResult cr = encoder.encode(lcb, bb, endOfInput);//将lcb的内容编码为字节尽可能多的放入到ByteBuffer中 if (cr.isUnderflow()) {//cr未溢出，ByteBuffer还有剩余的空间 if (lcb.hasRemaining()) {//lcb还有剩余的数据 leftoverChar = lcb.get(); if (cb != null && cb.hasRemaining()) flushLeftoverChar(cb, endOfInput); return; } break; } if (cr.isOverflow()) {//cr溢出，超出了ByteBuffer的上限 assert bb.position() > 0;//ByteBuffer中存在数据 writeBytes();//将ByteBuffer中的数据写入到输出流后清空ByteBuffer continue; } cr.throwException(); } haveLeftoverChar = false; } flushBuffer()和flush()将ByteBuffer中的数据写入输出流，flush()同时还会进行输出流的刷新，具体操作取决于输出流的实现，比如FileOutputStream是什么也不做，因为没有缓冲区。 public void flushBuffer() throws IOException { synchronized (lock) { if (isOpen()) implFlushBuffer(); else throw new IOException("Stream closed"); } } public void flush() throws IOException { synchronized (lock) { ensureOpen(); implFlush(); } } void implFlush() throws IOException { implFlushBuffer(); if (out != null) out.flush();//这里out的刷盘操作取决于子类的具体实现 } void implFlushBuffer() throws IOException { if (bb.position() > 0)// 如果ByteBuffer内还有剩余的数据，将它们写入文件 writeBytes(); } close()同样也只能关闭一次，并且是线程同步方法，在关闭之前需要先将缓冲区的内容全部输入到输入流中 public void close() throws IOException { synchronized (lock) { if (!isOpen) return; implClose(); isOpen = false;//只能关闭一次 } } void implClose() throws IOException { flushLeftoverChar(null, true);//将leftoverChar的内容写入的输出流 try { for (;;) { CoderResult cr = encoder.flush(bb); if (cr.isUnderflow())//cr未溢出说明ByteBuffer中的数据全部写入到输入流了 break; if (cr.isOverflow()) {//cr溢出说明ByteBuffer仍然存在数据 assert bb.position() > 0; writeBytes();//将ByteBuffer中的数据写入输出流 continue; } cr.throwException(); } if (bb.position() > 0) writeBytes(); if (ch != null)//关闭文件通道或者输出流 ch.close(); else out.close(); } catch (IOException x) { encoder.reset(); throw x; } } 总结 基于OutputStreamWriter和StreamEncoder来实现字符输出时，保证字符编码为字节后输入到输入流中的操作是可靠的，并且直到最后一个字符前一次会传递成对的字符来解决代理对的问题，后续的部分由输入流来完成，而OutputStreamWriter在通过FileWriter构建时是基于FileOutputStream，也就是没有缓冲区，有多少内容就写多少。
2018-08-29
优秀的个人博客,低调大师

IDEA 下如何修改 Elasticsearch 源码

以下所有过程以mac操作系统下（macOS 10.13.6）为例进行。 环境准备 安装jdk10 因为es需要高版本jdk进行开发，所以我们直接安装jdk10。 前往 http://www.oracle.com/technetwork/java/javase/downloads/jdk10-downloads-4416644.html ，直接下载安装mac的dmg版本即可。 jdk8和jdk10并存 如果平时开发用的jdk8，为了不影响之后使用，jdk8和jdk10将会同时存在。如果之前没用过java低版本请忽略。在 .bash_profile 中设置如下： export JAVA8_HOME=$(/usr/libexec/java_home -v 1.8) export JAVA10_HOME=$(/usr/libexec/java_ho
2018-08-29
优秀的个人博客,低调大师

Java ByteArrayInputStream ByteArrayOutputStream类源码解析

ByteArrayOutputStream ByteArrayOutputStream继承了抽象类OutputStream，本质是一个存在于堆内存中的可扩展byte数组，因为所有操作都在内存中所以flush()也什么都没做。 该类实现了一个输出流，数据写入一个byte数组，缓冲区随着写入的数据自动增大。数据可以通过toByteArray()和toString()重新取回。 关闭ByteArrayOutputStream没有影响，类中的方法可以在流被关闭后调用而不抛出IOException，因为close()实际上什么也没做。 从下面的代码中可以看出，所有涉及到输出流内容读写的操作，包括write、reset、toByteArray、size和toString都加了synchronized关键字，因为grow操作只改变数组大小不改变有效内容所以不需要加synchronized，因为要避免输出内容产生错乱。 从内部变量中可以看到，底层是一个byte数组，一个统计比特数的计数器 //存储数据的缓冲区 protected byte buf[]; //缓冲区内的有效byte protected int count; 无参构造时数组大小初始设为32，也可以指定size大小，size为负数时抛出IllegalArgumentException //创建一个新的ByteArrayOutputStream，初始缓冲区大小为32 public ByteArrayOutputStream() { this(32); } //创建一个新的ByteArrayOutputStream，初始缓冲区大小为size public ByteArrayOutputStream(int size) { if (size < 0) { throw new IllegalArgumentException("Negative initial size: " + size); } buf = new byte[size]; } 输入前要检查空间大小是否足够，下面一系列函数都是用于检查大小和进行扩大，数组大小存在上限，太大时会出现OOM错误。每次进行扩展时大小变为当前容量*2和需要的最小容量参数中的较大值，除非minCapacity>MAX_ARRAY_SIZE，否则不会突破这个大小，MAX_ARRAY_SIZE //在必要时增加容量来确保足够持有minCapacity表明的最小数目的元素 private void ensureCapacity(int minCapacity) { // overflow-conscious code if (minCapacity - buf.length > 0) grow(minCapacity); } //数组的最大大小，一些虚拟机保留了数组头部的一些位置。试图分配更大的数组会导致OutOfMemoryError：请求的数组大小超过了虚拟机的限制 private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; private void grow(int minCapacity) { // overflow-conscious code int oldCapacity = buf.length;//当前数组大小 int newCapacity = oldCapacity << 1;//新容量=当前容量*2 if (newCapacity - minCapacity < 0) newCapacity = minCapacity;//新容量为当前容量*2和minCapacity中的较大值 if (newCapacity - MAX_ARRAY_SIZE > 0) newCapacity = hugeCapacity(minCapacity);//当前容量*2>MAX_ARRAY_SIZE是要做处理 buf = Arrays.copyOf(buf, newCapacity);//新分配一个数组把数据复制过去 } private static int hugeCapacity(int minCapacity) { if (minCapacity < 0) // 负数代表minCapacity超过正数所能表示的范围了，所以内存溢出 throw new OutOfMemoryError(); return (minCapacity > MAX_ARRAY_SIZE) ? Integer.MAX_VALUE ://minCapacity在Integer.MAX_VALUE到Integer.MAX_VALUE-8范围内，大小为Integer.MAX_VALUE MAX_ARRAY_SIZE;//小于MAX_ARRAY_SIZE时为MAX_ARRAY_SIZE } 每次的写入操作需要先确保数组大小足够，然后将内容复制到数组中 //将特定的byte写入到这个输出流中 public synchronized void write(int b) { ensureCapacity(count + 1);//确保数组还能再多存储一个 buf[count] = (byte) b;//int转为byte保存到数组中 count += 1;//计数器+1 } //将从off开始长度为len的bytes写入到输出流中 public synchronized void write(byte b[], int off, int len) { if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) - b.length > 0)) { throw new IndexOutOfBoundsException(); } ensureCapacity(count + len); System.arraycopy(b, off, buf, count, len);//复制内容到数组中 count += len; } writeTo是将当前输出流中的内容全部写入到参数中的输出流中去，相当于调用out.write(buf, 0, count) public synchronized void writeTo(OutputStream out) throws IOException { out.write(buf, 0, count); } reset将count设为0，当前累计的输出都被丢弃了，这个输出流可以被重新使用，再次利用已经分配好的缓冲区空间，因为没有可以直接更改count的方法，所以相当于旧的内容无法再被访问了 public synchronized void reset() { count = 0; } toByteArray创建一个新的byte数组，大小为当前输出流中的byte大小，将其中的byte内容复制到新的数组中 public synchronized byte toByteArray()[] { return Arrays.copyOf(buf, count); } size返回当前流中的byte大小 public synchronized int size() { return count; } toString将输出流中的内容重新解码为String，可以指定字符集也可以直接默认使用平台字符集 //将缓冲区中的内容通过平台的默认字符集转换成string解码。新的String长度是字符集的函数，所以可能不等于buffer的大小,这个方法总是用平台的默认字符集中的默认替代字符串来替代畸形输入和非图形化表示字符序列 public synchronized String toString() { return new String(buf, 0, count); } //跟上面方法相比指定了字符集 public synchronized String toString(String charsetName) throws UnsupportedEncodingException { return new String(buf, 0, count, charsetName); } close可以看到什么都没做，所以一切照常，原因是该类操作都在堆内存中，不需要关闭一些句柄、套接字之类的东西所以什么都不用做 public void close() throws IOException { } ByteArrayInputStream ByteArrayInputStream是和ByteArrayOutputStream对应的输入流，继承了抽象类InputStream，本质上也是一个存储在堆内存中的byte数组，这个数组必须在构造时传入并直接被使用而不是复制，之后大小无法改变，可以通过直接对数组中的内容进行修改改变流中的内容。可以标记一个位置，通过reset重置回去来进行重复读取。 public static void main(String args[]) throws IOException{ ByteArrayOutputStream out = new ByteArrayOutputStream(); out.write("1234567890".getBytes()); byte buf[] = out.toByteArray(); ByteArrayInputStream in = new ByteArrayInputStream(buf); byte res[] = new byte[10]; in.read(res); System.out.println(new String(res));//1234567890 buf[0] = '4'; in.reset(); in.read(res); System.out.println(new String(res));//4234567890 } ByteArrayInputStream含有一个内部缓冲区包含了可能从流中读取的比特，一个内部计数器用来保持对下一个read方法读取的byte的跟踪。 关闭ByteArrayInputStream没有作用，所有方法都可以继续使用不会出现IOException，因为全部操作都在堆内存中不存在要关闭的句柄等。 从内部变量中可以看出，底层通过buf[]来存储流中的内容 /** * 构造时提供一个数组，buf中的比特是唯一能够从流中读取的，buf[pos]是下一个要被读取的byte */ protected byte buf[]; /** * 下一个要被读取的byte的下标，总是非负数，不能超过count，buf[pos]是下一个要被读取的byte */ protected int pos; /** * 流中当前标记的位置，ByteArrayInputStream在构造时默认标记位置是0。可以通过mark()方法标记缓冲区中的其他位置，能够通过reset()回到当前标记的位置 * 如果没有设置过mark，他的值会是传递给构造器的值，如果没有的话是0 */ protected int mark = 0; /** * 比当前有效字符的index大1，总是非负数，不超过buf数组的长度。比buf数组中最后一个能读取的byte的下标大1 */ protected int count; 构造函数必须要传入buf[]，传入后对于buf的引用无法改变，但可以通过缓存引用来直接修改其中的内容 //创建一个ByteArrayInputStream，使用buf作为缓冲区，缓冲区没有拷贝而是直接引用这个数组。pos初始值为0，count的初始值是buf的数组大小 public ByteArrayInputStream(byte buf[]) { this.buf = buf; this.pos = 0; this.count = buf.length; } //创建一个ByteArrayInputStream，使用buf作为缓冲区，pos初始值是offset，count的初始值是offset+length和buf.length中的较小值。缓冲区没有拷贝，直接引用了buf，mark设为offset public ByteArrayInputStream(byte buf[], int offset, int length) { this.buf = buf; this.pos = offset; this.count = Math.min(offset + length, buf.length); this.mark = offset; } 读取单个byte时，读取输入流中的下一个byte，返回值是0-255的int，已经到达末端没有可读的byte返回-1 public synchronized int read() { return (pos < count) ? (buf[pos++] & 0xff) : -1;//pos==count时到达末端返回-1 } 从流中读取最多len长度的bytes。如果pos==count，返回-1表示已经到达了文件末端，否则返回读取到的bytes数量k，k是len和count-pos中的较小值。如果k是正数，buf[pos]~buf[pos+k-1]将通过ystem.arraycopy被复制到b[off]~b[off+k-1]中，pos增加k，k被返回。 public synchronized int read(byte b[], int off, int len) { if (b == null) { throw new NullPointerException(); } else if (off < 0 || len < 0 || len > b.length - off) { throw new IndexOutOfBoundsException(); } if (pos >= count) { return -1; } int avail = count - pos; if (len > avail) { len = avail;//最多读取到流中数据末端 } if (len <= 0) { return 0; } System.arraycopy(buf, pos, b, off, len);//复制数据到b中 pos += len;//增加pos return len; } skip跳过输入流中的n个bytes，如果到达了末端跳过的bytes会少于n。实际跳过的bytes数量k等于n和count-pos中的较小值，pos增加k，k被返回 public synchronized long skip(long n) { long k = count - pos; if (n < k) { k = n < 0 ? 0 : n; } pos += k; return k; } available返回输入流中剩余可以读取或者跳过的bytes数量，等于count-pos，也就是缓冲区中剩余的可被读取的数量 public synchronized int available() { return count - pos; } markSupported测试这个InputStream是否支持mark/reset操作，ByteArrayInputStream总是支持该操作所以直接返回true public boolean markSupported() { return true; } mark将当前的位置标记，ByteArrayInputStream的标记位再构造时默认值是0，可以通过这个方法标记到缓冲区的其他位置，在构造时也可以通过传入offset设置其他mark，若没有传入则标记0。readAheadLimit这个参数在这里没有实际作用，InputStream中加这个参数是留给其他类中mark使用指出mark位置开始能读取的最大bytes数量。 public void mark(int readAheadLimit) { mark = pos; } reset将位置重置到mark的位置 public synchronized void reset() { pos = mark; } close方法什么都没做，所以依然可以照常使用，不会产生IOException public void close() throws IOException { }
2018-08-21

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2026-10-18 大小: 211.28KB 下载: 56次
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2026-09-24 大小: 189MB 下载: 9次
Spring

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

时间: 2026-08-21 大小: 5MB 下载: 2次
WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源，继承了IntelliJ IDEA强大的JS部分的功能。

时间: 2025-12-13 大小: 249.54MB 下载: 22次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册