一、前言

时间过得可真是快，距离上一次更新已经将近 2 个月了，这之间竟然没有任何的文章更新。原因呢，无非就是工作与生活变得紧张了，渐渐的就把这个给落下了。趁着 Flag 还没倒下，咱得把它扶起来不是。
Okio深入分析——基础使用部分主要了解了Okio这个库的特性，以及最基本的使用。但正如最后所说，我们还不知道它为什么好，为什么快。但是在分析之前，我们先过一遍其主体类的框架图，以便我们对其整体有一个粗略的认识。

OkioFramework.jpg

这个图看上去是不是有点简单啊，都不好意思说是一个框架。确实如此，所以说它是短小精悍呀。言归正传，我们还是来展开一下。
(1) Sink 和 Source 分别定义了写入和读取。
(2) BufferedSink 和 BufferedSource 分别定义了各种写入方法和读取方法，同时还定义了 buffer() 方法应该返回一个实际的 Buffer 用于操作写入与读取。
(3) Buffer 兼具实现了读和写的功能，Buffer 中又进一步将读写单元划分成了 Segment。各个 Segment 之间通过 prev 节点以及 next 节点构成一个双向链表。SegmentPool 是避免为了过多的分配 Segment 对象，用以循环利用 Segment。
(4) RealBufferSink 和 RealBufferSource 是唯一返回给用户的写入与读取实例。其分别由 Sink + Buffer 和 Source + Buffer 的组合模式而成，用以完成真正的写入与读取。

二、写入

案例分析

在分析源码之前，我们还是得先来个 demo 的案例分析。这次我们的 demo 得先做一个对比，使其看起来更有意思些。
通过 Okio 的 BufferSink 进行的写入代码

private void testSink(File file) {
        if(!file.exists()) {
            try {
                file.createNewFile();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try {
            BufferedSink bufferedSink = Okio.buffer(Okio.sink(file));
            StringBuilder stringBuilder = new StringBuilder();
            for(int k = 0;k < 1000;k++) {
                stringBuilder.append("Buffer Sink Test Sample");
            }
            int i = 0;
            System.out.println("Okio sink test start");
            long ts = System.currentTimeMillis();
            long nts = System.nanoTime();
            while(i++ < 100000) {
                bufferedSink.writeUtf8(stringBuilder.toString());
            }
            bufferedSink.close();
            System.out.println("Okio sink ms time " + (System.currentTimeMillis() - ts));
            System.out.println("Okio sink nano time " + (System.nanoTime() - nts));
            System.out.println("Okio sink test end");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

通过 Java 的 BufferedWriter 进行的写入代码

private void testFileOut(File file) {
        if(!file.exists()) {
            try {
                file.createNewFile();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try {
            FileWriter fileWriter = new FileWriter(file);
            BufferedWriter bufferedWriter = new BufferedWriter(fileWriter);
            StringBuilder stringBuilder = new StringBuilder();
            for(int k = 0;k < 1000;k++) {
                stringBuilder.append("Buffer Sink Test Sample");
            }
            int i = 0;
            System.out.println("Java buffered write test start");
            long ts = System.currentTimeMillis();
            long nts = System.nanoTime();
            while(i++ < 100000) {
                bufferedWriter.write(stringBuilder.toString());
            }

            bufferedWriter.close();
            System.out.println("Java buffered write ms time " + (System.currentTimeMillis() - ts));
            System.out.println("Java buffered write time " + (System.nanoTime() - nts));
            System.out.println("Java buffered write test end");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

这段程序运行之后有 2 个数据是需要我们关注的。
数据大小：通过下面的图，我们可以看到，这个写入数据大小为2.3GB

image.png
写入所花的时间： 如下 BufferSink 与 BufferedWriter 的时间对比，分别进行 10 次读写，应该能很明显的看到 BufferSink 的整体时间分布是要优于 BufferedWriter 的。

次序	BufferSink	BufferedWriter
1	4637	4944
2	4228	4299
3	4088	4334
4	4383	4440
5	4467	4375
6	4201	4117
7	4385	4453
8	4149	4439
9	4266	4222
10	4363	4433

上面的结果是PC上跑出来的结果，再来看看手机上的结果。也是基本得到一样的结果。就是 OKIO 的结果基本趋于稳定下降。当然在手机上写10000次太慢了，所以改成了 5000 次。

次序	BufferSink	BufferedWriter
1	3530	3139
2	3085	2943
3	2572	2137
4	2501	3192
5	2462	3108
6	3107	4081
7	2047	2377
8	2044	2131
9	2042	3557
10	2055	2099

总结下来就是，Okio 的写入并不总是比 Java 原生 I/O 快。而是在多次反复写入的情况下，Okio 则速度较快，且比较稳定。

源码分析

前面的案例分析，更加增强了我们对 Okio 的认知，接下来就是我们的重点，源码分析与理解了。按照 demo ，画出时序图，并逐层进行分解并深入。所以首先我们还是先来看看时序图。

OkioSink.jpg

构造Sink —— Okio#sink(file)

  /** Returns a sink that writes to {@code file}. */
  // 当传入的是一个文件时所调用的 Sink 构造方法
  public static Sink sink(File file) throws FileNotFoundException {
    if (file == null) throw new IllegalArgumentException("file == null");
    return sink(new FileOutputStream(file));
  }
 
  /** Returns a sink that writes to {@code out}. */
  // 当传入的是一个 OutputStream 时所调用的 Sink 构建方法
  public static Sink sink(OutputStream out) {
    return sink(out, new Timeout());
  }
  // 这个构建方法是私有的，最终的调用都会来到这里
  private static Sink sink(final OutputStream out, final Timeout timeout) {
    if (out == null) throw new IllegalArgumentException("out == null");
    if (timeout == null) throw new IllegalArgumentException("timeout == null");

    return new Sink() {
      @Override public void write(Buffer source, long byteCount) throws IOException {
        checkOffsetAndCount(source.size, 0, byteCount);
        while (byteCount > 0) {
          timeout.throwIfReached();
          Segment head = source.head;
          int toCopy = (int) Math.min(byteCount, head.limit - head.pos);
          out.write(head.data, head.pos, toCopy);

          head.pos += toCopy;
          byteCount -= toCopy;
          source.size -= toCopy;

          if (head.pos == head.limit) {
            source.head = head.pop();
            SegmentPool.recycle(head);
          }
        }
      }

      @Override public void flush() throws IOException {
        out.flush();
      }

      @Override public void close() throws IOException {
        out.close();
      }

      @Override public Timeout timeout() {
        return timeout;
      }

      @Override public String toString() {
        return "sink(" + out + ")";
      }
    };
  }

上面这段代码中，是构建 Sink 的 3 个重载方法，是Okio 的静态方法，而且他们也是依次调用。第一个告诉了我们其底层的写入也是基于 Java 原生的 I/O ，如 FileOutputStream。第二个加入了 Timeout ，这就使其具备了超时机制。第三个方法是最为关键的，这里才真正生成了 Sink 实例。而在这个实例中，也直接定义出了数据如何写入的，如何关闭流，以及返回哪一个 Timeout 实例。
这一波调用总的来说确定了实际的 Sink 对象，以及 Sink 是如何写入的和其应该返回的超时机制。
以上是 5 个Sink 构造方法中的其中 3 个，还有另外 2 个也一起来看看吧。

  /** Returns a sink that writes to {@code path}. */
  // 通过文件路径构建 Sink 实例
  @IgnoreJRERequirement // Should only be invoked on Java 7+.
  public static Sink sink(Path path, OpenOption... options) throws IOException {
    if (path == null) throw new IllegalArgumentException("path == null");
    return sink(Files.newOutputStream(path, options));
  }
  /**
   * Returns a sink that writes to {@code socket}. Prefer this over {@link
   * #sink(OutputStream)} because this method honors timeouts. When the socket
   * write times out, the socket is asynchronously closed by a watchdog thread.
   */
  // 通过  Socket 进行构建
  public static Sink sink(Socket socket) throws IOException {
    if (socket == null) throw new IllegalArgumentException("socket == null");
    AsyncTimeout timeout = timeout(socket);
    Sink sink = sink(socket.getOutputStream(), timeout);
    return timeout.sink(sink);
  }

上面这 2 个 Sink 构造方法，第一个是以文件路径来进行构建，而第二个则是以 Socket 进行构建，并且对于 Socket 的构建中，其 Timeout 是 AsyncTimeout。这里说明了 2 个其本特性。第一、除了对文件进行支持，同时还支持 Socket，第二、超时机制有同步的 Timeout 以及异步的 AsyncTimeout。关于超时机制，后面还会详细来讲，这里先有个概念即可。

构造BufferedSink——Okio#buffer(sink)

/**
   * Returns a new sink that buffers writes to {@code sink}. The returned sink
   * will batch writes to {@code sink}. Use this wherever you write to a sink to
   * get an ergonomic and efficient access to data.
   */
  public static BufferedSink buffer(Sink sink) {
    return new RealBufferedSink(sink);
  }

该方法就是构建了一个 RealBufferedSink 实例。那就来看看这个类。

final class RealBufferedSink implements BufferedSink {
  public final Buffer buffer = new Buffer();
  public final Sink sink;
  boolean closed;

  RealBufferedSink(Sink sink) {
    if (sink == null) throw new NullPointerException("sink == null");
    this.sink = sink;
  }
......
}

这个类中两个重要的成员变量一个 sink ，由外部传入，一个是 Buffer 由内部直构建。这里的 sink 是前面通过 Okio.sink()方法最终构建的那个匿名类。那么再来看看 Buffer 这个类。

/**
 * A collection of bytes in memory.
 *
 * <p><strong>Moving data from one buffer to another is fast.</strong> Instead
 * of copying bytes from one place in memory to another, this class just changes
 * ownership of the underlying byte arrays.
 *
 * <p><strong>This buffer grows with your data.</strong> Just like ArrayList,
 * each buffer starts small. It consumes only the memory it needs to.
 *
 * <p><strong>This buffer pools its byte arrays.</strong> When you allocate a
 * byte array in Java, the runtime must zero-fill the requested array before
 * returning it to you. Even if you're going to write over that space anyway.
 * This class avoids zero-fill and GC churn by pooling byte arrays.
 */
public final class Buffer implements BufferedSource, BufferedSink, Cloneable {
  private static final byte[] DIGITS =
      { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
  static final int REPLACEMENT_CHARACTER = '\ufffd';

  @Nullable Segment head;
  long size;

  public Buffer() {
  }

Buffer 类中最重要的成员变量 head 是一个 Segment 类型，Buffer 中的数据结构也正是由 Segment 所构成的一个双向链表。除此之外，关于 Buffer 类，这里的解释也极其重要。大概的意思是说：
(1) 这是一个在内存中的字节集合。
(2)将一个 Buffer 中的数据移到另一个 Buffer 中去是非常快的。原因是它并不是通过复制来实现，而是从底层字节数组去改变与 Buffer 这个类的所属关系。
(3)它就是 ArrayList 一样的，容量自动增长的。一开始是很小的，且仅分配它所需要的内存大小。
(4)Buffer池化了它的字节数组。如果直接从Java中分配一个字节数组，使用其必须为其填充 0。而 Buffer 通过池化字节数组从而避免了 0 填充和GC 抖动的情况。
上面详细的说明了 Buffer 的特性，当然也是 Okio 的特性。在后面的章节中我们还会继续分析其特性的实现细节。而关于 Buffer 就先了解到这里，再进一步看看Segment。

package okio;

import javax.annotation.Nullable;

/**
 * A segment of a buffer.
 *
 * <p>Each segment in a buffer is a circularly-linked list node referencing the following and
 * preceding segments in the buffer.
 *
 * <p>Each segment in the pool is a singly-linked list node referencing the rest of segments in the
 * pool.
 *
 * <p>The underlying byte arrays of segments may be shared between buffers and byte strings. When a
 * segment's byte array is shared the segment may not be recycled, nor may its byte data be changed.
 * The lone exception is that the owner segment is allowed to append to the segment, writing data at
 * {@code limit} and beyond. There is a single owning segment for each byte array. Positions,
 * limits, prev, and next references are not shared.
 */
final class Segment {
  /** The size of all segments in bytes. */
// 每个 segment 共 8 K字节
  static final int SIZE = 8192;
  /** Segments will be shared when doing so avoids {@code arraycopy()} of this many bytes. */
// 是否共享底层数据的阈值
  static final int SHARE_MINIMUM = 1024;
// 存放字节的数组
  final byte[] data;
  /** The next byte of application data byte to read in this segment. */
 // 下一个要读的字节的位置
  int pos;
  /** The first byte of available data ready to be written to. */
 // 写入的起始位置
  int limit;
  /** True if other segments or byte strings use the same byte array. */
 // 表明此 segment 是共享的。那什么时候情况下会被共享呢？
  boolean shared;
  /** True if this segment owns the byte array and can append to it, extending {@code limit}. */
// 表明这个段是可以追加数据的，而追回的位置是基于 limit。一般情况下都是为 true 的
  boolean owner;
  /** Next segment in a linked or circularly-linked list. */
// 双向环形链表的下一个节点
  Segment next;
  /** Previous segment in a circularly-linked list. */
// 双向环形链表的前一个节点
  Segment prev;

  Segment() {
    this.data = new byte[SIZE];
    this.owner = true;
    this.shared = false;
  }

  Segment(Segment shareFrom) {
    this(shareFrom.data, shareFrom.pos, shareFrom.limit);
    shareFrom.shared = true;
  }

  Segment(byte[] data, int pos, int limit) {
    this.data = data;
    this.pos = pos;
    this.limit = limit;
    this.owner = false;
    this.shared = true;
  }

  /**
   * Removes this segment of a circularly-linked list and returns its successor.
   * Returns null if the list is now empty.
   */
// 将自己弹出链表
  public @Nullable Segment pop() {
    ......
  }

  /**
   * Appends {@code segment} after this segment in the circularly-linked list.
   * Returns the pushed segment.
   */
// 压入到当前链表
  public Segment push(Segment segment) {
   ......
  }

  /**
   * Splits this head of a circularly-linked list into two segments. The first
   * segment contains the data in {@code [pos..pos+byteCount)}. The second
   * segment contains the data in {@code [pos+byteCount..limit)}. This can be
   * useful when moving partial segments from one buffer to another.
   *
   * <p>Returns the new head of the circularly-linked list.
   */
// 分割，把数据以 [pos,pos + byteCount] 和 [pos + byteCount,limit] 这两段来进行分割。分割出去的数据依据共享阈值决定数据是否被共享
  public Segment split(int byteCount) {
    ......
  }

  /**
   * Call this when the tail and its predecessor may both be less than half
   * full. This will copy data so that segments can be recycled.
   */
// 压缩存储，当处于尾节点和它的前一节点的数据都少于一半时，把数据进行压缩存储到一个，以便可以回收调一个 Segment。
  public void compact() {
    ......
  }
// 移动 byteCount 个字节数据到另一个段中
  /** Moves {@code byteCount} bytes from this segment to {@code sink}. */
  public void writeTo(Segment sink, int byteCount) {
    ......
  }
}

来解读一下 Segment 的特性：
(1) segment 是 Buffer 中的一个组织单元
(2) segment 在 Buffer 中以环形双向链表的数据结构来存储
(3) segment 在 SegmentPool 中以单链接表数据结构来存储
(4) segment 可以在 Buffer 和 ByteStrings 之间进行共享。对于共享的 segment 不可以回收，也不可以修改其中的数据。但例外的是，段的拥有者可以在 limit 之后追回数据。对于这个段来说，底层的字节数组是共享的，但position,limits,prev以及next这些是不共享的。
此外，注释里有对每一个属性的详细解释以及个人的理解。为了方便更容易理解，简单画了一个 Segment 的数据模型图来帮助我们。如下所示。

image.png

关于 Segment 就先了解到这里，其中的方法 split(),compact()以及 writeTo() 后面还会详细说明。接下来继续看看 SegmentPool。

package okio;

import javax.annotation.Nullable;

/**
 * A collection of unused segments, necessary to avoid GC churn and zero-fill.
 * This pool is a thread-safe static singleton.
 */
// 主要是用于收集未使用的 Segments，避免了内存抖动以及0填充。同时，这个 Pool 也是一个线程安全的静态单例类。
final class SegmentPool {
  /** The maximum number of bytes to pool. */
  // TODO: Is 64 KiB a good maximum size? Do we ever have that many idle segments?
// 一个池子最大存储 64 KB 字节
  static final long MAX_SIZE = 64 * 1024; // 64 KiB.

  /** Singly-linked list of segments. */
// 表明其是一个单向链表
  static @Nullable Segment next;

  /** Total bytes in this pool. */
// 当前的总数据大小
  static long byteCount;

  private SegmentPool() {
  }

  static Segment take() {
    synchronized (SegmentPool.class) {
      if (next != null) {
        Segment result = next;
        next = result.next;
        result.next = null;
        byteCount -= Segment.SIZE;
        return result;
      }
    }
    return new Segment(); // Pool is empty. Don't zero-fill while holding a lock.
  }

  static void recycle(Segment segment) {
    if (segment.next != null || segment.prev != null) throw new IllegalArgumentException();
    if (segment.shared) return; // This segment cannot be recycled.
    synchronized (SegmentPool.class) {
      if (byteCount + Segment.SIZE > MAX_SIZE) return; // Pool is full.
      byteCount += Segment.SIZE;
      segment.next = next;
      segment.pos = segment.limit = 0;
      next = segment;
    }
  }
}

这个类其实挺简单的，take() 方法中就是获取一个 Segment ，如果池子中有就取一个出来，如果没有就给 new 一个。而 recycle() 方法，如果池子还没涨到 64 KB 的限制，且 Segment 不是共享的，那就把它加到链接里面去。

小结
通过对 Sink 以及 BufferedSink 的构造，可以说是对 Okio 整个使用环境的一个初始化。同时，在这一段也详细介绍了其相关的核心概念，Buffer,Segment,SegmentPool 等。有了这些基本的核心概念，我们就可以进一步分析字节的写入了。

写入数据——RealBufferedSink#writeUtf8()

  @Override public BufferedSink writeUtf8(String string) throws IOException {
    if (closed) throw new IllegalStateException("closed");
    buffer.writeUtf8(string);
    return emitCompleteSegments();
  }

进一步调用了 Buffer 的 writeUtf8() 和 emitCompleteSegments()，那么就往下看Buffer#writeUft8() 方法。代码很长，但关键的是 2 个调用，writableSegment(1) 和 writeByte()，其他的都是 utf-8 编码相关。

  @Override public Buffer writeUtf8(String string) {
    return writeUtf8(string, 0, string.length());
  }

  @Override public Buffer writeUtf8(String string, int beginIndex, int endIndex) {
    ......

    // Transcode a UTF-16 Java String to UTF-8 bytes.
    for (int i = beginIndex; i < endIndex;) {
      int c = string.charAt(i);
      if (c < 0x80) {
        Segment tail = writableSegment(1);
        byte[] data = tail.data;
        int segmentOffset = tail.limit - i;
        int runLimit = Math.min(endIndex, Segment.SIZE - segmentOffset);

        // Emit a 7-bit character with 1 byte.
        data[segmentOffset + i++] = (byte) c; // 0xxxxxxx

        // Fast-path contiguous runs of ASCII characters. This is ugly, but yields a ~4x performance
        // improvement over independent calls to writeByte().
        while (i < runLimit) {
          c = string.charAt(i);
          if (c >= 0x80) break;
          data[segmentOffset + i++] = (byte) c; // 0xxxxxxx
        }

        int runSize = i + segmentOffset - tail.limit; // Equivalent to i - (previous i).
        tail.limit += runSize;
        size += runSize;

      } else if (c < 0x800) {
        // Emit a 11-bit character with 2 bytes.
        writeByte(c >>  6        | 0xc0); // 110xxxxx
        writeByte(c       & 0x3f | 0x80); // 10xxxxxx
        i++;

      } else if (c < 0xd800 || c > 0xdfff) {
        // Emit a 16-bit character with 3 bytes.
        writeByte(c >> 12        | 0xe0); // 1110xxxx
        writeByte(c >>  6 & 0x3f | 0x80); // 10xxxxxx
        writeByte(c       & 0x3f | 0x80); // 10xxxxxx
        i++;

      } else {
        // c is a surrogate. Make sure it is a high surrogate & that its successor is a low
        // surrogate. If not, the UTF-16 is invalid, in which case we emit a replacement character.
        int low = i + 1 < endIndex ? string.charAt(i + 1) : 0;
        if (c > 0xdbff || low < 0xdc00 || low > 0xdfff) {
          writeByte('?');
          i++;
          continue;
        }

        // UTF-16 high surrogate: 110110xxxxxxxxxx (10 bits)
        // UTF-16 low surrogate:  110111yyyyyyyyyy (10 bits)
        // Unicode code point:    00010000000000000000 + xxxxxxxxxxyyyyyyyyyy (21 bits)
        int codePoint = 0x010000 + ((c & ~0xd800) << 10 | low & ~0xdc00);

        // Emit a 21-bit character with 4 bytes.
        writeByte(codePoint >> 18        | 0xf0); // 11110xxx
        writeByte(codePoint >> 12 & 0x3f | 0x80); // 10xxxxxx
        writeByte(codePoint >>  6 & 0x3f | 0x80); // 10xxyyyy
        writeByte(codePoint       & 0x3f | 0x80); // 10yyyyyy
        i += 2;
      }
    }

    return this;
  }

Buffer有两个重载版本，第一个调用了第二个，主要就是确定了起始位置为0，而写入的数据为全部数据。而第二个是具体的写入，这段代码比较长，其实不应该贴出来的，但我看到它对 Utf8 的编码十分的优秀，所以还是贴出来，有时间的可以仔细读一读。当然，在这里我们只关注其中最关键的几句即可。如前面所说，这里面有两个进一步调用的方法分别是 writableSegment(1) 和 writeByte()，而 writeByte() 里面也有对 writableSegment() 的调用。那我们先来看看 writableSegment() 方法的实现。

/**
   * Returns a tail segment that we can write at least {@code minimumCapacity}
   * bytes to, creating it if necessary.
   */
  Segment writableSegment(int minimumCapacity) {
    ......
    if (head == null) {
      head = SegmentPool.take(); // Acquire a first segment.
      return head.next = head.prev = head;
    }
    Segment tail = head.prev;
    if (tail.limit + minimumCapacity > Segment.SIZE || !tail.owner) {
      tail = tail.push(SegmentPool.take()); // Append a new empty segment to fill up.
    }
    return tail;
  }

这个方法的主要含义就是如果head为空，则说明当前的链接还是空的，那么就新建一个节点，此时头尾相等。如果不为空就取出尾节点，即头节点的前向节点。而尾节点的空间不能再装下当前要写入的字节数时，就会创建一个新的节点，并把新建的节点作为尾节点，然后继续写入。再来看看 writeByte()。

  @Override public Buffer writeByte(int b) {
    Segment tail = writableSegment(1);
    tail.data[tail.limit++] = (byte) b;
    size += 1;
    return this;
  }

这里就是把数据写入到 Segment 的 data[] 中。对于一次 writeUtf8() 的调用来看，目前为止是把数据全部都写入到了 Segment 中，同时也是内存中，还没有发生实际的 I/O 的。接下来继续看 RealBufferedSink#emitCompleteSegments()方法。

  @Override public BufferedSink emitCompleteSegments() throws IOException {
    if (closed) throw new IllegalStateException("closed");
    long byteCount = buffer.completeSegmentByteCount();
    if (byteCount > 0) sink.write(buffer, byteCount);
    return this;
  }

这里先是调用了 Buffer#completeSegmentByteCount()确定需要从 buffer 中实际取多少字节写入到实际的输出流中。

  /**
   * Returns the number of bytes in segments that are not writable. This is the
   * number of bytes that can be flushed immediately to an underlying sink
   * without harming throughput.
   */
  public long completeSegmentByteCount() {
    long result = size;
    if (result == 0) return 0;

    // Omit the tail if it's still writable.
    Segment tail = head.prev;
    if (tail.limit < Segment.SIZE && tail.owner) {
      result -= tail.limit - tail.pos;
    }

    return result;
  }

上面代码的意思是用当前总的 size 大小减去尾节点已写入的字节数，也就是除尾节点，其他的都会被立即写入到底层的输出流中。这说明了不足 8K 的数据写入，在 close() 之前是不会发生实际 I/O 的。接下来，进一步调用了 sink#write()。这个 sink 就是我们在前面通过 Okio#sink() 方法所创建出来的匿名类。为了连贯性，我们还是把写入相关的代码再贴一下以便于进一步的阅读与分析。

@Override public void write(Buffer source, long byteCount) throws IOException {
        ......
        while (byteCount > 0) { // 循环，直到数据全部写入
          timeout.throwIfReached(); // 超时判断
          Segment head = source.head; // 总是取头结点
         // 确定当次需要写入的字节数，当前 segment 中可写入的数据与剩余字节数，取最小的那个
          int toCopy = (int) Math.min(byteCount, head.limit - head.pos);
          out.write(head.data, head.pos, toCopy);//写入数据到输出流
         
         // 更新各位置
          head.pos += toCopy; // 读指针往前移 toCopy 个
          byteCount -= toCopy; // 写入字节数减 toCopy 个
          source.size -= toCopy; // Buffer 中大小减 toCopy 个
          // 如果当前 segment 的数据已经全部写入到输出流，那么将其弹出双向循环链接，并将其回收到 SegmentPool 中
          if (head.pos == head.limit) {
            source.head = head.pop();
            SegmentPool.recycle(head);
          }
        }

贴出来的代码中，已经作了非常详细的说明，因为这里的每一句都很重要，都需要我们去理解。当然，总的来说其实也只有 2 个关键点。其一，超时机制。

/**
   * Throws an {@link InterruptedIOException} if the deadline has been reached or if the current
   * thread has been interrupted. This method doesn't detect timeouts; that should be implemented to
   * asynchronously abort an in-progress operation.
   */
  public void throwIfReached() throws IOException {
    if (Thread.interrupted()) {
      throw new InterruptedIOException("thread interrupted");
    }

    if (hasDeadline && deadlineNanoTime - System.nanoTime() <= 0) {
      throw new InterruptedIOException("deadline reached");
    }
  }

根据这里的注释，以及分析 hasDeadLine 和 deadlineNanoTime 的赋值情况来看，超时的检测只应该发生在异步的情况下。纳尼！！！也就是说同步超时机制是不检查的。其二，就是数据的写入以及各位位置标记的置位和Segment的回收。

关闭——RealBufferedSink#close()

@Override public void close() throws IOException {
    if (closed) return;

    // Emit buffered data to the underlying sink. If this fails, we still need
    // to close the sink; otherwise we risk leaking resources.
    Throwable thrown = null;
    try {
      if (buffer.size > 0) {
        sink.write(buffer, buffer.size);
      }
    } catch (Throwable e) {
      thrown = e;
    }

    try {
      sink.close();
    } catch (Throwable e) {
      if (thrown == null) thrown = e;
    }
    closed = true;

    if (thrown != null) Util.sneakyRethrow(thrown);
  }

close() 也就是将 buffer 中剩余的数据都写入到底层输出流中，最后再调用 sink#close()关闭底层的输出流。至此，写入数据可以说大致分析完了。因为与 writeUtf8() 同级别的其他 write方法，其基本原理都是类似的。通过分析我们知道，其是先将数据都写入到 Buffer 中，然后才将数据写入到实际的 I/O 中，以此减少实际的 I/O。而分析到这里，我们应该也注意到，前面所说的 Segment 的 splite,compact以及share 这些优化内存的方法都没有被提及过。这就接下来要分析的读取与另一不同级别的写入。

三、读取

案例分析

关于读取，我们同样也从案例分析开始。先来看看读取的代码。
通过 Okio 的 BufferSource 进行的读取代码

private void testSource(File file) {
        try {
            BufferedSource bufferedSource = Okio.buffer(Okio.source(file));

            System.out.println("Okio source test start");
            long ts = System.currentTimeMillis();
            byte[] buffer = new byte[8 * 1024];

            while(bufferedSource.read(buffer) > 0) {

            }

            bufferedSource.close();
            System.out.println("Okio source test end");
            System.out.println(" source test " + (System.currentTimeMillis() - ts));

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

通过 Java 的 BufferedReader 进行的读取代码

private void testJava(File file) {

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(file));

            System.out.println("java test start");
            long ts = System.currentTimeMillis();

            char buffer[] = new char[8 * 1024];

            while(bufferedReader.read(buffer) > 0) {
            }
            System.out.println("java test end");
            System.out.println("java test " + (System.currentTimeMillis() - ts));
            bufferedReader.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

同样来看看测试结果，不过这里偷了个懒，只测试了PC端的，没有测试手机端。因为平台不影响结果。文件就是上面写入的文件，大小 2.3 G。

次序	BufferSource	BufferedReader
1	1117	2391
2	986	2458
3	979	2252
4	980	2232
5	976	2225
6	977	2202
7	983	2252
8	968	2218
9	979	2223
10	977	2221

对比结果很明显，Okio#BufferSource 的读取速度明显大于 Java 原生的2 倍多。但是，为什么呢？另外有一个细节，上面的 buffer 大小只设置了 8 * 1024 ，也就是 8KiB，这是因为 Okio 的 segment 为 8 KiB，其一次读取最大也只读取 8 KiB

源码分析

有了前面案例分析的基本认知，接下来就是我们的源码分析与理解了。同样，先画出时序图，再逐层进行分解并深入。

OkioSource.jpg

从时序图上可以看到，前面几个步骤都是相同的，如 Buffer,Segment,Timeout 这些概念和构造都是一样的。就不重复介绍了。主要看看 Source 和 RealBufferSource 的构造吧。

构造Source——Okio#source(file)

  /** Returns a source that reads from {@code file}. */
  public static Source source(File file) throws FileNotFoundException {
    if (file == null) throw new IllegalArgumentException("file == null");
    return source(new FileInputStream(file));
  }
  /** Returns a source that reads from {@code in}. */
  public static Source source(InputStream in) {
    return source(in, new Timeout());
  }
private static Source source(final InputStream in, final Timeout timeout) {
    if (in == null) throw new IllegalArgumentException("in == null");
    if (timeout == null) throw new IllegalArgumentException("timeout == null");

    return new Source() {
      @Override public long read(Buffer sink, long byteCount) throws IOException {
        if (byteCount < 0) throw new IllegalArgumentException("byteCount < 0: " + byteCount);
        if (byteCount == 0) return 0;
        try {
          timeout.throwIfReached();
          Segment tail = sink.writableSegment(1);
          int maxToCopy = (int) Math.min(byteCount, Segment.SIZE - tail.limit);
          int bytesRead = in.read(tail.data, tail.limit, maxToCopy);
          if (bytesRead == -1) return -1;
          tail.limit += bytesRead;
          sink.size += bytesRead;
          return bytesRead;
        } catch (AssertionError e) {
          if (isAndroidGetsocknameError(e)) throw new IOException(e);
          throw e;
        }
      }

      @Override public void close() throws IOException {
        in.close();
      }

      @Override public Timeout timeout() {
        return timeout;
      }

      @Override public String toString() {
        return "source(" + in + ")";
      }
    };
  }

上面的代码看着多，但和 Sink 的构造几乎一样。这里的重点也构造了一个匿名的 Source 接口的实例，以完成实际从输入中的读取。同样的，Source 也是支持 Socket 的，顺便也一起来看看吧。

/**
   * Returns a source that reads from {@code socket}. Prefer this over {@link
   * #source(InputStream)} because this method honors timeouts. When the socket
   * read times out, the socket is asynchronously closed by a watchdog thread.
   */
  public static Source source(Socket socket) throws IOException {
    if (socket == null) throw new IllegalArgumentException("socket == null");
    AsyncTimeout timeout = timeout(socket);
    Source source = source(socket.getInputStream(), timeout);
    return timeout.source(source);
  }

支持 Socket 也是比较简单的，即从 socket 中获取输入流以便读取数据，同时这里也设定了 Timeout 为 AsyncTimeout。关于 AsyncTimeout 就放到后面专门的分节里来讲解。

构造 BufferdSource —— Okio#buffer(Source)

  /**
   * Returns a new source that buffers reads from {@code source}. The returned
   * source will perform bulk reads into its in-memory buffer. Use this wherever
   * you read a source to get an ergonomic and efficient access to data.
   */
  public static BufferedSource buffer(Source source) {
    return new RealBufferedSource(source);
  }

构造 RealBufferedSource 实例，继续看。

final class RealBufferedSource implements BufferedSource {
  public final Buffer buffer = new Buffer();
  public final Source source;
  boolean closed;

  RealBufferedSource(Source source) {
    if (source == null) throw new NullPointerException("source == null");
    this.source = source;
  }
......

同样是组合了 Source 和 Buffer 这两个实例。Source 就是前面通过 Okio.source()方法最终构建的那个匿名类。而 Buffer 和其内部聚合的 Segment 都已经介绍过了。所以直接来看 read() 方法吧。

读取数据——RealBufferedSource#read()

  @Override public int read(byte[] sink) throws IOException {
    return read(sink, 0, sink.length);
  }

 @Override public int read(byte[] sink, int offset, int byteCount) throws IOException {
    checkOffsetAndCount(sink.length, offset, byteCount);

    if (buffer.size == 0) {
      long read = source.read(buffer, Segment.SIZE);
      if (read == -1) return -1;
    }

    int toRead = (int) Math.min(byteCount, buffer.size);
    return buffer.read(sink, offset, toRead);
  }

read() 是个重载方法，我们调用的是第一个，然后它会进一步调用指定参数 offset 和 byteCount 的 read() 方法。
在这个方法里，buffer.size 为 0 说明当前 buffer 中为没有数据，即数据已经被从buffer 中读出来了。如果不为0，说明 buffer 中还有数据那么就进一步读取剩下的或者剩余的数据。这里最初肯定是要走为 0 的情况。这里调用的是 source.read()，也即前面匿名类 Source 的。注意，这里传入的参数中除了 buffer ，还有 Segment.SIZE，即 8KiB。
这里为了方便分析，也重复贴一下 read() 方法的相关代码。

@Override public long read(Buffer sink, long byteCount) throws IOException {
       .....
       ......
        try {
          timeout.throwIfReached();
          Segment tail = sink.writableSegment(1);
          int maxToCopy = (int) Math.min(byteCount, Segment.SIZE - tail.limit);
          int bytesRead = in.read(tail.data, tail.limit, maxToCopy);
          if (bytesRead == -1) return -1;
          tail.limit += bytesRead;
          sink.size += bytesRead;
          return bytesRead;
        } catch (AssertionError e) {
          if (isAndroidGetsocknameError(e)) throw new IOException(e);
          throw e;
        }
      }

同步 Timeout.throwIfReached() 前面已经分析过了，是不起作用的，其实现在来看看也能明白过来，不管是在读还是写的时候，它都没有设置起始时间，那它又如何能计算出所消耗的时间呢？而 Buffer#writeableSegment() 在前面也分析过了，主要是返回一个 Segment。然后就是实际从文件读出数据并只保存在 Segment 里面。

在 RealBufferedSource#read() 中，走了不为 0 的情况后，就是将Buffer中的数据按需要读到内存中返回给调用者。

有了前面写入的分析，再来看读取显然要简单的多了。至此，读取和写入的最基本流程都分析完了。Okio 为打包了很多的读取和写入的方法，这里只分析了其中最基本的，当然有了这些最基本的认知后，再来理解基本的就容易许多了。

三、读取到写入，写入自读取

除了基本的读取与写入，还可以将读取与写入直接串起来。也即读取出来数据可以直接送进写入，而写入的内容也可以是来自 Source 。先来看看写入自读取吧。

@Override public long writeAll(Source source) throws IOException {
    if (source == null) throw new IllegalArgumentException("source == null");
    long totalBytesRead = 0;
    for (long readCount; (readCount = source.read(buffer, Segment.SIZE)) != -1; ) {
      totalBytesRead += readCount;
      emitCompleteSegments();
    }
    return totalBytesRead;
  }

  @Override public BufferedSink write(Source source, long byteCount) throws IOException {
    while (byteCount > 0) {
      long read = source.read(buffer, byteCount);
      if (read == -1) throw new EOFException();
      byteCount -= read;
      emitCompleteSegments();
    }
    return this;
  }

上面两段代码就是将数据从 Source 读出，再写入到Buffer 中去。这里需要关注的有两个事情：
source实例 ：这里的 Source 是 RealBufferdSource 实例。
关键调用 : 这里的关键调用是 RealBufferdSource#read(Buffer,byteCount)。来看看其实现。

@Override public long read(Buffer sink, long byteCount) throws IOException {
    ......
    return buffer.read(sink, toRead);
  }

调用了 Buffer#read(Buffer,toRead)

  @Override public long read(Buffer sink, long byteCount) {
   ......
    sink.write(this, byteCount);
    return byteCount;
  }

看到这里可能会有点晕，所以强调一下，这里的 sink 是 RealBufferdSink 中的 Buffer，而 this 是 RealBufferedSource 中的 buffer。意思就是将 RealBufferdSource 中的 buffer 的数据写入到 RealBufferdSink 中的 buffer 中去。下面来看看具体的实现，其实代码并不长，注释就占了一大半，为了完整性，我并没有将其删除去。

@Override public void write(Buffer source, long byteCount) {
    // Move bytes from the head of the source buffer to the tail of this buffer
    // while balancing two conflicting goals: don't waste CPU and don't waste
    // memory.
    //
    //
    // Don't waste CPU (ie. don't copy data around).
    //
    // Copying large amounts of data is expensive. Instead, we prefer to
    // reassign entire segments from one buffer to the other.
    //
    //
    // Don't waste memory.
    //
    // As an invariant, adjacent pairs of segments in a buffer should be at
    // least 50% full, except for the head segment and the tail segment.
    //
    // The head segment cannot maintain the invariant because the application is
    // consuming bytes from this segment, decreasing its level.
    //
    // The tail segment cannot maintain the invariant because the application is
    // producing bytes, which may require new nearly-empty tail segments to be
    // appended.
    //
    //
    // Moving segments between buffers
    //
    // When writing one buffer to another, we prefer to reassign entire segments
    // over copying bytes into their most compact form. Suppose we have a buffer
    // with these segment levels [91%, 61%]. If we append a buffer with a
    // single [72%] segment, that yields [91%, 61%, 72%]. No bytes are copied.
    //
    // Or suppose we have a buffer with these segment levels: [100%, 2%], and we
    // want to append it to a buffer with these segment levels [99%, 3%]. This
    // operation will yield the following segments: [100%, 2%, 99%, 3%]. That
    // is, we do not spend time copying bytes around to achieve more efficient
    // memory use like [100%, 100%, 4%].
    //
    // When combining buffers, we will compact adjacent buffers when their
    // combined level doesn't exceed 100%. For example, when we start with
    // [100%, 40%] and append [30%, 80%], the result is [100%, 70%, 80%].
    //
    //
    // Splitting segments
    //
    // Occasionally we write only part of a source buffer to a sink buffer. For
    // example, given a sink [51%, 91%], we may want to write the first 30% of
    // a source [92%, 82%] to it. To simplify, we first transform the source to
    // an equivalent buffer [30%, 62%, 82%] and then move the head segment,
    // yielding sink [51%, 91%, 30%] and source [62%, 82%].

    if (source == null) throw new IllegalArgumentException("source == null");
    if (source == this) throw new IllegalArgumentException("source == this");
    checkOffsetAndCount(source.size, 0, byteCount);

    while (byteCount > 0) {
      // Is a prefix of the source's head segment all that we need to move?
      if (byteCount < (source.head.limit - source.head.pos)) {
        Segment tail = head != null ? head.prev : null;
        if (tail != null && tail.owner
            && (byteCount + tail.limit - (tail.shared ? 0 : tail.pos) <= Segment.SIZE)) {
          // Our existing segments are sufficient. Move bytes from source's head to our tail.
          source.head.writeTo(tail, (int) byteCount);
          source.size -= byteCount;
          size += byteCount;
          return;
        } else {
          // We're going to need another segment. Split the source's head
          // segment in two, then move the first of those two to this buffer.
          source.head = source.head.split((int) byteCount);
        }
      }

      // Remove the source's head segment and append it to our tail.
      Segment segmentToMove = source.head;
      long movedByteCount = segmentToMove.limit - segmentToMove.pos;
      source.head = segmentToMove.pop();
      if (head == null) {
        head = segmentToMove;
        head.next = head.prev = head;
      } else {
        Segment tail = head.prev;
        tail = tail.push(segmentToMove);
        tail.compact();
      }
      source.size -= movedByteCount;
      size += movedByteCount;
      byteCount -= movedByteCount;
    }
  }

通过注释，可以明白，这个方法的核心是在讲：
(1) 这个方法主要完成的是从 Source 的首部将数据移入到 Sink 的尾部
(2) 移动过程中需要平衡两个重要的指标，优化 CPU 资源，优化内存资源。
(3) 移动数据是通过改变引用指向，而不是数据的复制
(4) Segment 合并，是说前一个 Segment 与当前的 Segment 数据大小之和如果没有超过 100%，那么就会将当前的 Segment 合并到前一个中去。看看下面的代码，你可能会更加的清楚。

/**
   * Call this when the tail and its predecessor may both be less than half
   * full. This will copy data so that segments can be recycled.
   */
  public void compact() {
    if (prev == this) throw new IllegalStateException();
    if (!prev.owner) return; // Cannot compact: prev isn't writable.
    int byteCount = limit - pos;
    int availableByteCount = SIZE - prev.limit + (prev.shared ? 0 : prev.pos);
    if (byteCount > availableByteCount) return; // Cannot compact: not enough writable space.
    writeTo(prev, byteCount);
    pop();
    SegmentPool.recycle(this);
  }

(5) 分割 Segment，对于有一些 Segment ，如果只是部分被读取出来，那么可以通过将其分割成 2 个 Segment ，然后取需要的那个 Segment 加入到当前的 Buffer 中。

/**
   * Splits this head of a circularly-linked list into two segments. The first
   * segment contains the data in {@code [pos..pos+byteCount)}. The second
   * segment contains the data in {@code [pos+byteCount..limit)}. This can be
   * useful when moving partial segments from one buffer to another.
   *
   * <p>Returns the new head of the circularly-linked list.
   */
  public Segment split(int byteCount) {
    if (byteCount <= 0 || byteCount > limit - pos) throw new IllegalArgumentException();
    Segment prefix;

    // We have two competing performance goals:
    //  - Avoid copying data. We accomplish this by sharing segments.
    //  - Avoid short shared segments. These are bad for performance because they are readonly and
    //    may lead to long chains of short segments.
    // To balance these goals we only share segments when the copy will be large.
    if (byteCount >= SHARE_MINIMUM) {
      prefix = new Segment(this);
    } else {
      prefix = SegmentPool.take();
      System.arraycopy(data, pos, prefix.data, 0, byteCount);
    }

    prefix.limit = prefix.pos + byteCount;
    pos += byteCount;
    prev.push(prefix);
    return prefix;
  }

明白了这个方法的主要意图，再来分析具体的代码。分两类情况：
其一，当要读取的数据大小小于 source.head 中的字节数时，如果当前 sink.head 不为 null，则将 source.head 的数据都写入到 sink.tail 中去，当然也是其最后的要读取的数据了，即 soruce.head.writeTo()；而当 sink.head 为空时，则先将 source.head 再 byteCount 大小进行数据的分割，再进入到第二种情况。而在分割时，如果需要的数据大于分割的阈值，那么被分割出来的 Segment 就会被共享成同一个 Segment。
其二，当要读取的数据大小大于等于 source.head 字节时，如果当前 sink.head 为 null ，那么就让 source.head 从 source.buffer 中取出，并让 sink.head 直接指向 source.head 即可。而如果当前 sink.head 不为空，则将取出来的 segment 并到当前 sink.buffer 的尾部。根据前面的分析，如果尾部的数据与其前一个数据大小之和没有达到 100% ，那就可以进行合并。

以上便是 Segment 的合并与分割了。Okio 在这里同时考虑到了 CPU 和内存的优化，并且实现的非常微妙。

同时上面有部分也提到了 Segment 的共享，从整个源码以及分析来看，共享主要的场景就是发生在复制时以及可能的分割时。当然，最重要的是分割时了。而关于复制时的共享，其实也是非常符合逻辑了。即使被标记成分享的 Segment 不能再写入，但是对于内存的利用来上讲也是非常优化的。

四、异步的超时机制

异步超时机制主要只是用于 Socket 通信，这里也重复贴一下 Source 的代码，Sink 差不多。

  /**
   * Returns a source that reads from {@code socket}. Prefer this over {@link
   * #source(InputStream)} because this method honors timeouts. When the socket
   * read times out, the socket is asynchronously closed by a watchdog thread.
   */
  public static Source source(Socket socket) throws IOException {
    if (socket == null) throw new IllegalArgumentException("socket == null");
    AsyncTimeout timeout = timeout(socket);
    Source source = source(socket.getInputStream(), timeout);
    return timeout.source(source);
  }

如前面所说，Socket 通信时，主要是拿出了其 InputStream，以便于读取数据。而这里的 Timeout 是其子类 AsyncTimeout。当通过重载方法 source() 构造出了 Source 之后，进一步送进了 AsyncTimeout#source() 方法中

/**
   * Returns a new source that delegates to {@code source}, using this to implement timeouts. This
   * works best if {@link #timedOut} is overridden to interrupt {@code sink}'s current operation.
   */
  public final Source source(final Source source) {
    return new Source() {
      @Override public long read(Buffer sink, long byteCount) throws IOException {
        boolean throwOnTimeout = false;
        enter();
        try {
          long result = source.read(sink, byteCount);
          throwOnTimeout = true;
          return result;
        } catch (IOException e) {
          throw exit(e);
        } finally {
          exit(throwOnTimeout);
        }
      }

      @Override public void close() throws IOException {
        boolean throwOnTimeout = false;
        try {
          source.close();
          throwOnTimeout = true;
        } catch (IOException e) {
          throw exit(e);
        } finally {
          exit(throwOnTimeout);
        }
      }

      @Override public Timeout timeout() {
        return AsyncTimeout.this;
      }

      @Override public String toString() {
        return "AsyncTimeout.source(" + source + ")";
      }
    };
  }

Okio#source()主要是定义了如何读取数据，而AsyncTimeout#source()定义了读取数据的超时机制。在 read() 方法中，主要体现在 enter() 以及 exit() 这两个方法上。

思考一下当前的场景，当前线程正在读取数据，那管理超时必然需要另一个线程。在 AsyncTimeout 中就定义了 WatchDog 线程，来监测超时的读写。

WatchDog 线程，俗名 “看门狗”，其在嵌入式方面应用非常多。而Android系统中的SystemServer是否存在耗时操作的监测也是由一个 WatchDog 线程来实现的。此处的 WatchDog 线程是由第一次 enter() 进入所触发的，起用后便被设置成守护线程运行，并且线程以链表的形式来管理。与此同时， enter() 也会添加一个新的 AsyncTimeout 节点到 WatchDog 线程中。而 exit() 则会检查节点是否还在链表中，否则时间到了不在链表中就会被认为是超时状态了，从而抛出超时的异常。具体我们来看看代码的实现。

public final void enter() {
    if (inQueue) throw new IllegalStateException("Unbalanced enter/exit");
    long timeoutNanos = timeoutNanos();
    boolean hasDeadline = hasDeadline();
    if (timeoutNanos == 0 && !hasDeadline) {
      return; // No timeout and no deadline? Don't bother with the queue.
    }
    inQueue = true;
    scheduleTimeout(this, timeoutNanos, hasDeadline);
  }

同 enter() 调用 scheduleTimeout() ，调度起一个 Timeout。

private static synchronized void scheduleTimeout(
      AsyncTimeout node, long timeoutNanos, boolean hasDeadline) {
    // Start the watchdog thread and create the head node when the first timeout is scheduled.
// 链表为空说明 Watchdog 还没有启动起来，这里添加了头结点，并且启动了 watchdog 线程。
    if (head == null) {
      head = new AsyncTimeout();
      new Watchdog().start();
    }

// 分不同的条件讲计算终止时间
    long now = System.nanoTime();
    if (timeoutNanos != 0 && hasDeadline) {
      // Compute the earliest event; either timeout or deadline. Because nanoTime can wrap around,
      // Math.min() is undefined for absolute values, but meaningful for relative ones.
      node.timeoutAt = now + Math.min(timeoutNanos, node.deadlineNanoTime() - now);
    } else if (timeoutNanos != 0) {
      node.timeoutAt = now + timeoutNanos;
    } else if (hasDeadline) {
      node.timeoutAt = node.deadlineNanoTime();
    } else {
      throw new AssertionError();
    }

// 按剩余时间升序排序
    // Insert the node in sorted order.
    long remainingNanos = node.remainingNanos(now);
    for (AsyncTimeout prev = head; true; prev = prev.next) {
      if (prev.next == null || remainingNanos < prev.next.remainingNanos(now)) {
        node.next = prev.next;
        prev.next = node;
        if (prev == head) {
          AsyncTimeout.class.notify(); // Wake up the watchdog when inserting at the front.
        }
        break;
      }
    }
  }

以上便是插入了一个 AsyncTimeout 的节点。再来看看 Watchdog 线程是如何监测超时的。

private static final class Watchdog extends Thread {
    Watchdog() {
      super("Okio Watchdog");
      setDaemon(true);
    }

    public void run() {
      while (true) {
        try {
          AsyncTimeout timedOut;
          synchronized (AsyncTimeout.class) {
            timedOut = awaitTimeout();

            // Didn't find a node to interrupt. Try again.
            if (timedOut == null) continue;

            // The queue is completely empty. Let this thread exit and let another watchdog thread
            // get created on the next call to scheduleTimeout().
            if (timedOut == head) {
              head = null;
              return;
            }
          }

          // Close the timed out node.
          timedOut.timedOut();
        } catch (InterruptedException ignored) {
        }
      }
    }
  }

重点在 awaitTimeout ()，即等待一下，看看是否有 AsyncTimeout 超时了。

/**
   * Removes and returns the node at the head of the list, waiting for it to time out if necessary.
   * This returns {@link #head} if there was no node at the head of the list when starting, and
   * there continues to be no node after waiting {@code IDLE_TIMEOUT_NANOS}. It returns null if a
   * new node was inserted while waiting. Otherwise this returns the node being waited on that has
   * been removed.
   */
  static @Nullable AsyncTimeout awaitTimeout() throws InterruptedException {
    // Get the next eligible node.
    AsyncTimeout node = head.next;

    // The queue is empty. Wait until either something is enqueued or the idle timeout elapses.
    if (node == null) {
      long startNanos = System.nanoTime();
      AsyncTimeout.class.wait(IDLE_TIMEOUT_MILLIS);
      return head.next == null && (System.nanoTime() - startNanos) >= IDLE_TIMEOUT_NANOS
          ? head  // The idle timeout elapsed.
          : null; // The situation has changed.
    }

    long waitNanos = node.remainingNanos(System.nanoTime());

    // The head of the queue hasn't timed out yet. Await that.
    if (waitNanos > 0) {
      // Waiting is made complicated by the fact that we work in nanoseconds,
      // but the API wants (millis, nanos) in two arguments.
      long waitMillis = waitNanos / 1000000L;
      waitNanos -= (waitMillis * 1000000L);
      AsyncTimeout.class.wait(waitMillis, (int) waitNanos);
      return null;
    }

    // The head of the queue has timed out. Remove it.
    head.next = node.next;
    node.next = null;
    return node;
  }

首先找到剩余时间最小的那个节点，按照前面升序排序，也就是 head.next 节点。然后再检查剩余时间是否大于 0 ，说明还没有超时，这时便将等待到其剩余时间，并且返回空，表示并未超时。这里提一下的是 class.wait(time) 方法，如果时间没有到，收到对应 class 的 notify()/notifyAll() 也是会提前被唤醒的。假设这里并没有超时，最后再来看看 exit()。

/**
   * Throws an IOException if {@code throwOnTimeout} is {@code true} and a timeout occurred. See
   * {@link #newTimeoutException(java.io.IOException)} for the type of exception thrown.
   */
  final void exit(boolean throwOnTimeout) throws IOException {
    boolean timedOut = exit();
    if (timedOut && throwOnTimeout) throw newTimeoutException(null);
  }

  /**
   * Returns either {@code cause} or an IOException that's caused by {@code cause} if a timeout
   * occurred. See {@link #newTimeoutException(java.io.IOException)} for the type of exception
   * returned.
   */
  final IOException exit(IOException cause) throws IOException {
    if (!exit()) return cause;
    return newTimeoutException(cause);
  }

  /** Returns true if the timeout occurred. */
  public final boolean exit() {
    if (!inQueue) return false;
    inQueue = false;
    return cancelScheduledTimeout(this);
  }

exit() 方法又重载了 3 个，很显然，前面 2 个主要是调用第 3 个来判断是否超时以便进一步的抛出超异常。这里主要来看看第 3 个。其又进一步调用了 cancelScheduledTimeout() 方法，这是与 scheduledTimeout()相对应的方法。

/** Returns true if the timeout occurred. */
  private static synchronized boolean cancelScheduledTimeout(AsyncTimeout node) {
    // Remove the node from the linked list.
    for (AsyncTimeout prev = head; prev != null; prev = prev.next) {
      if (prev.next == node) {
        prev.next = node.next;
        node.next = null;
        return false;
      }
    }

    // The node wasn't found in the linked list: it must have timed out!
    return true;
  }

从链表中寻找当前结点，如果找到则从链接中移除并表示没有超时。如果没有找到，则表示已经超时了。

至此，异步超时机制分析完毕。如果看前面的分析还有点晕，不防来看看总结：
(1) 管理超时机制的数据结构是一个升序链表，而管理这个链表的是守护线程 Watchdog ，它会每次选择一个节点，并通过剩余时间的等待来等待当前 AsyncTimeout 的超时与否。
(2) 通过 enter() 方法，再经过 scheduleTimeout() 方法添加一个 AsyncTimeout节点
(3) 通过 exit() 方法，再经过 cancleScheduleTimeut() 方法将未超时的 AsyncTimeout 从链表中移除。移除后，当 Watchdog 进入下一次遍历时就不会再遍历到被它等待超时的那个节点了。

五、总结

感谢你能读到并读完这篇文章，对于 Okio 的全部分析到此为止了。文章确实贴了不少的源码，但我仍然觉得是有必要的，它能使我们读起来更加的通畅以及完整。当然，也许还有更好的表述方式。本文并没有分析到了 Okio 所有的方方面面，但我相信能够认真读完并且理解其中大部分的意思，再去分析其他部分应该也非常简单了。最后，由于本人水平有限，在分析过程中难免会有错误，表述上也会存在不尽人意的地方。对于一切问题，欢迎下方留言讨论，不甚感激。

Okio深入分析—源码分析部分

一、前言

二、写入

案例分析

源码分析

构造Sink —— Okio#sink(file)

构造BufferedSink——Okio#buffer(sink)

写入数据——RealBufferedSink#writeUtf8()

关闭——RealBufferedSink#close()

三、读取

案例分析

源码分析

构造Source——Okio#source(file)

构造 BufferdSource —— Okio#buffer(Source)

读取数据——RealBufferedSource#read()

三、读取到写入，写入自读取

四、异步的超时机制

五、总结