Okio 源碼解析（一）：數據讀取流程

senntyou 發布于2019-08-15 13:31 / 1694人閱讀

摘要：封裝了和，并且有多個優點提供超時機制不需要人工區分字節流與字符流，易于使用易于測試本文先介紹的基本用法，然后分析源碼中數據讀取的流程。和分別用于提供字節流和接收字節流，對應于和。和則是保存了相應的緩存數據用于高效讀寫。

簡介

Okio 是 square 開發的一個 Java I/O 庫，并且也是 OkHttp 內部使用的一個組件。Okio 封裝了 java.io 和 java.nio，并且有多個優點：

提供超時機制

不需要人工區分字節流與字符流，易于使用

易于測試

本文先介紹 Okio 的基本用法，然后分析源碼中數據讀取的流程。

基本用法

Okio 的用法很簡單，下面是讀取和寫入的示例：

// 讀取
InputStream inputStream = ...
BufferedSource bufferedSource = Okio.buffer(Okio.source(inputStream));
String line = bufferedSource.readUtf8();

// 寫入
OutputStream outputStream = ...
BufferedSink bufferedSink = Okio.buffer(Okio.sink(outputStream));
bufferedSink.writeString("test", Charset.defaultCharset());
bufferedSink.close();

Okio 用 Okio.source 封裝 InputStream，用 Okio.sink 封裝 OutputStream。然后統一交給 Okio.buffer 分別獲得 BufferedSource 和 BufferedSink，這兩個類提供了大量的讀寫數據的方法。BufferedSource 中包含的部分接口如下：

int readInt() throws IOException;
long readLong() throws IOException;
byte readByte() throws IOException;
ByteString readByteString() throws IOException;
String readUtf8() throws IOException;
String readString(Charset charset) throws IOException;

其中既包含了讀取字節流，也包含讀取字符流的方法，BufferedSink 則提供了對應的寫入數據的方法。

基本框架

Okio 中有4個接口，分別是 Source、Sink、 BufferedSource 和 BufferedSink。Source 和 Sink 分別用于提供字節流和接收字節流，對應于 Inpustream 和 OutputStream。BufferedSource 和 BufferedSink 則是保存了相應的緩存數據用于高效讀寫。這幾個接口的繼承關系如下：

從上圖可以看出，Source 和 Sink 提供基本的 read 和 write 方法，而 BufferedSource 和 BufferedSink 則提供了更多的操作數據的方法，但這些都是接口，真正實現的類是 RealBufferedSource 和 RealBufferedSink。

另外還有個類是 Buffer, 它同時實現了 BufferedSource 和 BufferedSink，并且 RealBufferedSource 和 RealbufferedSink 都包含一個 Buffer 對象，真正的數據讀取操作都是交給 Buffer 完成的。

由于 read 和 write 操作類似，下面以 read 的流程對代碼進行分析。

Okio.source

Okio.source 有幾個重載的方法，用于封裝輸入流，最終調用的代碼如下：

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 + ")";
      }
    };
  }

從上面代碼可以看出，Okio.source 接受兩個參數，一個是 InputStream，另一個是 Timeout，返回了一個匿名的 Source 的實現類。這里主要看一下 read 方法，首先是參數為空的判斷，然后是從 in 中讀取數據到類型為 Buffer 的 sink 中，這段代碼中涉及到 Buffer 以及 Segment，下面先看看這兩個東西。

Segment

在 Okio 中，每個 Segment 代表一段數據，多個 Segment 串成一個循環雙向鏈表。下面是 Segment 的成員變量和構造方法：

final class Segment {
  // segment數據的字節數
  static final int SIZE = 8192;
  // 共享的Segment的最低的數據大小
  static final int SHARE_MINIMUM = 1024;
  // 實際保存的數據
  final byte[] data;
  // 下一個可讀的位置
  int pos;
  // 下一個可寫的位置
  int limit;
  // 保存的數據是否是共享的
  boolean shared;
  // 保存的數據是否是獨占的
  boolean owner;
  // 鏈表中下一個節點
  Segment next;
  // 鏈表中上一個節點
  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;
  }
  ...
}

變量的含義已經寫在了注釋中，可以看出 Segment 中的數據保存在一個字節數組中，并提供了一些變量標識讀與寫的位置。Segment 既然是鏈表中的節點，下面看一下插入與刪除的方法：

// 在當前Segment后面插入一個Segment
public Segment push(Segment segment) {
    segment.prev = this;
    segment.next = next;
    next.prev = segment;
    next = segment;
    return segment;
  }
// 從鏈表中刪除當前Segment，并返回其后繼節點
public @Nullable Segment pop() {
    Segment result = next != this ? next : null;
    prev.next = next;
    next.prev = prev;
    next = null;
    prev = null;
    return result;
  }

插入與刪除的代碼其實就是數據結構中鏈表的操作。

Buffer

下面看看 Buffer 是如何使用 Segment 的。Buffer 中有兩個重要變量：

@Nullable Segment head;
long size;

一個是 head，表示這個 Buffer 保存的 Segment 鏈表的頭結點。還有一個 size，用于記錄 Buffer 當前的字節數。

在上面 Okio.source 中生成的匿名的 Source 的 read 方法中，要讀取數據到 Buffer 中，首次是調用了 writableSegment，這個方法是獲取一個可寫的 Segment，代碼如下所示：

  Segment writableSegment(int minimumCapacity) {
    if (minimumCapacity < 1 || minimumCapacity > Segment.SIZE) throw new IllegalArgumentException();

    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;
  }

獲取 Segment 的邏輯是先判斷 Buffer 是否有了 Segment 節點，沒有就先去 SegmentPool 中取一個，并且將首尾相連，形成循環鏈表。如果已經有了，找到末尾的 Segment，判斷其剩余空間是否滿足，不滿足就再從 SegmentPool 中獲取一個新的 Segment 添加到末尾。最后，返回末尾的 Segment 用于寫入。

SegmentPool 用于保存廢棄的 Segment，其中有兩個方法，take 從中獲取，recycle 用于回收。

上面 Okio.buffer(Okio.source(in)) 最終得到的是 RealBufferedSource，這個類中持有一個 Buffer 對象和一個 Source 對象，真正的讀取操作由這兩個對象合作完成。下面是 readString 的代碼：

@Override public String readString(long byteCount, Charset charset) throws IOException {
    require(byteCount);
    if (charset == null) throw new IllegalArgumentException("charset == null");
    return buffer.readString(byteCount, charset);
  }
@Override public void require(long byteCount) throws IOException {
    if (!request(byteCount)) throw new EOFException();
  }
// 從source中讀取數據到buffer中
@Override public boolean request(long byteCount) throws IOException {
    if (byteCount < 0) throw new IllegalArgumentException("byteCount < 0: " + byteCount);
    if (closed) throw new IllegalStateException("closed");
    while (buffer.size < byteCount) {
      if (source.read(buffer, Segment.SIZE) == -1) return false;
    }
    return true;
  }

首先是從 Source 中讀取數據到 Buffer 中，然后調用 buffer.readstring 方法得到最終的字符串。下面是 readString 的代碼：

@Override public String readString(long byteCount, Charset charset) throws EOFException {
    checkOffsetAndCount(size, 0, byteCount);
    if (charset == null) throw new IllegalArgumentException("charset == null");
    if (byteCount > Integer.MAX_VALUE) {
      throw new IllegalArgumentException("byteCount > Integer.MAX_VALUE: " + byteCount);
    }
    if (byteCount == 0) return "";

    Segment s = head;
    if (s.pos + byteCount > s.limit) {
      // 如果string跨多個Segment，委托給readByteArray去讀
      return new String(readByteArray(byteCount), charset);
    }
    // 將字節序列轉換成String
    String result = new String(s.data, s.pos, (int) byteCount, charset);
    s.pos += byteCount;
    size -= byteCount;

    // 如果pos==limit，回收這個Segment
    if (s.pos == s.limit) {
      head = s.pop();
      SegmentPool.recycle(s);
    }

    return result;
}

在上面的代碼中，便是從 Buffer 的 Segment 鏈表中讀取數據。如果 String 跨多個 Segment，那么調用 readByteArray 循環讀取字節序列。最終將字節序列轉換為 String 對象。如果 Segment 的 pos 等于 limit，說明這個 Segment 的數據已經全部讀取完畢，可以回收，放入 SegmentPool。

Okio 讀取數據的時候統一將輸入流看成是字節序列，讀入 Buffer 后在用到的時候再轉換，例如上面讀取 String 時將字節序列進行了轉換。其它還有很多類型，如下面是 readInt 的代碼：

@Override public int readInt() {
    if (size < 4) throw new IllegalStateException("size < 4: " + size);

    Segment segment = head;
    int pos = segment.pos;
    int limit = segment.limit;

    // If the int is split across multiple segments, delegate to readByte().
    if (limit - pos < 4) {
      return (readByte() & 0xff) << 24
          |  (readByte() & 0xff) << 16
          |  (readByte() & 0xff) <<  8
          |  (readByte() & 0xff);
    }

    byte[] data = segment.data;
    int i = (data[pos++] & 0xff) << 24
        |   (data[pos++] & 0xff) << 16
        |   (data[pos++] & 0xff) <<  8
        |   (data[pos++] & 0xff);
    size -= 4;

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

    return i;
}

Buffer 使用 Segment 鏈表保存數據，有個好處是在不同的 Buffer 之間移動數據只需要轉移其字節序列的擁有權，如 copyTo(Buffer out, long offset, long byteCount) 代碼所示：

public Buffer copyTo(Buffer out, long offset, long byteCount) {
    if (out == null) throw new IllegalArgumentException("out == null");
    checkOffsetAndCount(size, offset, byteCount);
    if (byteCount == 0) return this;

    out.size += byteCount;

    // Skip segments that we aren"t copying from.
    Segment s = head;
    for (; offset >= (s.limit - s.pos); s = s.next) {
      offset -= (s.limit - s.pos);
    }

    // Copy one segment at a time.
    for (; byteCount > 0; s = s.next) {
      Segment copy = new Segment(s);
      copy.pos += offset;
      copy.limit = Math.min(copy.pos + (int) byteCount, copy.limit);
      if (out.head == null) {
        out.head = copy.next = copy.prev = copy;
      } else {
        out.head.prev.push(copy);
      }
      byteCount -= copy.limit - copy.pos;
      offset = 0;
    }

    return this;
}

其中并沒有拷貝字節數據，只是鏈表的相關操作。

總結

Okio 讀取數據的流程基本就如本文所分析的，寫入操作與讀取是類似的。Okio 通過 Source 與 Sink 標識輸入流與輸出流。在 Buffer 中使用 Segment 鏈表的方式保存字節數據，并且通過 Segment 擁有權的共享避免了數據的拷貝，通過 SegmentPool 避免了廢棄數據的GC，使得 Okio 成為一個高效的 I/O 庫。Okio 還有一個優點是超時機制，具體內容可進入下一篇：Okio 源碼解析（二）：超時機制

如果我的文章對您有幫助，不妨點個贊支持一下(^_^)

GPU云服務器云服務器 asp讀取數據庫源碼 linux串口讀取一幀數據包 Okio 解析流程

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/68152.html

Okio 源碼解析（二）：超時機制

摘要：內部會新開一個叫做的線程，根據超時時間依次處理鏈表的節點。總結通過以及分別提供了同步超時和異步超時功能，同步超時是在每次讀取數據前判斷是否超時，異步超時則是將組成有序鏈表，并且開啟一個線程來監控，到達超時則觸發相關操作。簡介上一篇文章（Okio 源碼解析（一）：數據讀取流程）分析了 Okio 數據讀取的流程，從中可以看出 Okio 的便捷與高效。Okio 的另外一個優點是提供了超時...

honmaple 2019-08-15 13:41 評論0 收藏0
從設計者的角度理解Java IO流

摘要：通過多個裝飾類實現責任鏈模式，它將對一個輸入流的不同處理分散到不同的中去。 1、基本概念 1.1、InputStream 最基本的字節輸入流，抽象類，定義了讀取原始字節的所有基本方法1.1.1、public abstract int read() throws IOException 讀取一個字節的方法，最基礎的方法1.1.2、public int read(byte b[], in...

Flink_China 2019-08-14 17:14 評論0 收藏0
Android網絡編程6之OkHttp3用法全解析

摘要：使用前準備配置添加網絡權限異步請求慣例，請求百度可以省略，默認是請求請求成功與版本并沒有什么不同，比較郁悶的是回調仍然不在線程。前言上一篇介紹了OkHttp2.x的用法，這一篇文章我們來對照OkHttp2.x版本來看看，OkHttp3使用起來有那些變化。當然，看這篇文章前建議看一下前一篇文章Android網絡編程（五）OkHttp2.x用法全解析。 1.使用前準備 Android ...

Amos 2019-08-16 10:51 評論0 收藏0