Android IO 框架 Okio 的實現原理，到底哪裡 OK？

本文已收錄到 AndroidFamily，技術和職場問題，請關注公眾號 [彭旭銳] 提問。

前言

大家好，我是小彭。

今天，我們來討論一個 Square 開源的 I/O 框架 Okio，我們最開始接觸到 Okio 框架還是源於 Square 家的 OkHttp 網路框架。那麼，OkHttp 為什麼要使用 Okio，它相比於 Java 原生 IO 有什麼區別和優勢？今天我們就圍繞這些問題展開。

本文原始碼基於 Okio v3.2.0。

---

思維導圖

---

1. 說一下 Okio 的優勢？

相比於 Java 原生 IO 框架，我認為 Okio 的優勢主要體現在 3 個方面：

• 1、精簡且全面的 API： 原生 IO 使用裝飾模式，例如使用 BufferedInputStream 裝飾 FileInputStream 檔案輸入流，可以增強流的緩衝功能。但是原生 IO 的裝飾器過於龐大，需要區分位元組、字元流、位元組陣列、字元陣列、緩衝等多種裝飾器，而這些恰恰又是最常用的基礎裝飾器。相較之下，Okio 直接在 BufferedSource 和 BufferedSink 中聚合了原生 IO 中所有基礎的裝飾器，使得框架更加精簡；

• 2、基於共用的緩衝區設計： 由於 IO 系統呼叫存在上下文切換的效能損耗，為了減少系統呼叫次數，應用層往往會採用緩衝區策略。但是緩衝區又會存在副作用，當資料從一個緩衝區轉移到另一個緩衝區時需要拷貝資料，這種記憶體中的拷貝顯得沒有必要。而 Okio 採用了基於共用的緩衝區設計，在緩衝區間轉移資料只是共用 Segment 的參照，而減少了記憶體拷貝。同時 Segment 也採用了物件池設計，減少了記憶體分配和回收的開銷；

• 3、超時機制： Okio 彌補了部分 IO 操作不支援超時檢測的缺陷，而且 Okio 不僅支援單次 IO 操作的超時檢測，還支援包含多次 IO 操作的複合任務超時檢測。

下面，我們將從這三個優勢展開分析：

---

2. 精簡的 Okio 框架

先用一個表格總結 Okio 框架中主要的型別：

型別	描述
Source	輸入流
Sink	輸出流
BufferedSource	快取輸入流介面，實現類是 RealBufferedSource
BufferedSink	緩衝輸出流介面，實現類是 RealBufferedSink
Buffer	緩衝區，由 Segment 連結串列組成
Segment	資料片段，多個片段組成邏輯上連續資料
ByteString	String 類
Timeout	超時控制

2.1 Source 輸入流 與 Sink 輸出流

在 Java 原生 IO 中有四個基礎介面，分別是：

• 位元組流： InputStream 輸入流和 OutputStream 輸出流；
• 字元流： Reader 輸入流和 Writer 輸出流。

而在 Okio 更加精簡，只有兩個基礎介面，分別是：

• 流： Source 輸入流和 Sink 輸出流。

Source.kt

interface Source : Closeable {

    // 從輸入流讀取資料到 Buffer 中（Buffer 等價於 byte[] 位元組陣列）
    // 返回值：-1：輸入內容結束
    @Throws(IOException::class)
    fun read(sink: Buffer, byteCount: Long): Long

    // 超時控制（詳細分析見後續文章）
    fun timeout(): Timeout

    // 關閉流
    @Throws(IOException::class)
    override fun close()
}

Sink.java

actual interface Sink : Closeable, Flushable {

    // 將 Buffer 的資料寫入到輸出流中（Buffer 等價於 byte[] 位元組陣列）
    @Throws(IOException::class)
    actual fun write(source: Buffer, byteCount: Long)

    // 清空輸出緩衝區
    @Throws(IOException::class)
    actual override fun flush()

    // 超時控制（詳細分析見後續文章）
    actual fun timeout(): Timeout

    // 關閉流
    @Throws(IOException::class)
    actual override fun close()
}

2.2 InputStream / OutputStream 與 Source / Sink 互轉

在功能上，InputStream - Source 和 OutputStream - Sink 分別是等價的，而且是相互相容的。結合 Kotlin 擴充套件函數，兩種介面之間的轉換會非常方便：

• source()： InputStream 轉 Source，實現類是 InputStreamSource；
• sink()： OutputStream 轉 Sink，實現類是 OutputStreamSink；

比較不理解的是： Okio 沒有提供 InputStreamSource 和 OutputStreamSink 轉回 InputStream 和 OutputStream 的方法，而是需要先轉換為 BufferSource 與 BufferSink，再轉回 InputStream 和 OutputStream。

• buffer()： Source 轉 BufferedSource，Sink 轉 BufferedSink，實現類分別是 RealBufferedSource 和 RealBufferedSink。

範例程式碼

// 原生 IO -> Okio
val source = FileInputStream(File("")).source()
val bufferSource = FileInputStream(File("")).source().buffer()

val sink = FileOutputStream(File("")).sink()
val bufferSink = FileOutputStream(File("")).sink().buffer()

// Okio -> 原生 IO
val inputStream = bufferSource.inputStream()
val outputStream = bufferSink.outputStream()

JvmOkio.kt

// InputStream -> Source
fun InputStream.source(): Source = InputStreamSource(this, Timeout())

// OutputStream -> Sink
fun OutputStream.sink(): Sink = OutputStreamSink(this, Timeout())

private class InputStreamSource(
    private val input: InputStream,
    private val timeout: Timeout
) : Source {

    override fun read(sink: Buffer, byteCount: Long): Long {
        if (byteCount == 0L) return 0
        require(byteCount >= 0) { "byteCount < 0: $byteCount" }
        try {
            // 同步超時監控（詳細分析見後續文章）
            timeout.throwIfReached()
            // 讀入 Buffer
            val tail = sink.writableSegment(1)
            val maxToCopy = minOf(byteCount, Segment.SIZE - tail.limit).toInt()
            val bytesRead = input.read(tail.data, tail.limit, maxToCopy)
            if (bytesRead == -1) {
                if (tail.pos == tail.limit) {
                    // We allocated a tail segment, but didn't end up needing it. Recycle!
                    sink.head = tail.pop()
                    SegmentPool.recycle(tail)
                }
                return -1
            }
            tail.limit += bytesRead
            sink.size += bytesRead
            return bytesRead.toLong()
        } catch (e: AssertionError) {
            if (e.isAndroidGetsocknameError) throw IOException(e)
            throw e
        }
  }

  override fun close() = input.close()

  override fun timeout() = timeout

  override fun toString() = "source($input)"
}

private class OutputStreamSink(
    private val out: OutputStream,
    private val timeout: Timeout
) : Sink {

    override fun write(source: Buffer, byteCount: Long) {
        checkOffsetAndCount(source.size, 0, byteCount)
        var remaining = byteCount
        // 寫出 Buffer
        while (remaining > 0) {
            // 同步超時監控（詳細分析見後續文章）
            timeout.throwIfReached()
            // 取有效資料量和剩餘輸出量的較小值
            val head = source.head!!
            val toCopy = minOf(remaining, head.limit - head.pos).toInt()
            out.write(head.data, head.pos, toCopy)

            head.pos += toCopy
            remaining -= toCopy
            source.size -= toCopy

            // 指向下一個 Segment
            if (head.pos == head.limit) {
                source.head = head.pop()
                SegmentPool.recycle(head)
            }
        }
    }

    override fun flush() = out.flush()

    override fun close() = out.close()

    override fun timeout() = timeout

    override fun toString() = "sink($out)"
}

Okio.kt

// Source -> BufferedSource
fun Source.buffer(): BufferedSource = RealBufferedSource(this)

// Sink -> BufferedSink
fun Sink.buffer(): BufferedSink = RealBufferedSink(this)

2.3 BufferSource 與 BufferSink

在 Java 原生 IO 中，為了減少系統呼叫次數，我們一般不會直接呼叫 InputStream 和 OutputStream，而是會使用 BufferedInputStream 和 BufferedOutputStream 包裝類增加緩衝功能。

例如，我們希望採用帶緩衝的方式讀取字元格式的檔案，則需要先將檔案輸入流包裝為字元流，再包裝為緩衝流：

Java 原生 IO 範例

// 第一層包裝
FileInputStream fis = new FileInputStream(file);
// 第二層包裝
InputStreamReader isr = new InputStreamReader(new FileInputStream(file), "UTF-8");
// 第三層包裝
BufferedReader br = new BufferedReader(isr);
String line;
while ((line = br.readLine()) != null) {
    ...
}
// 省略 close

同理，我們在 Okio 中一般也不會直接呼叫 Source 和 Sink，而是會使用 BufferedSource 和 BufferedSink 包裝類增加緩衝功能：

Okio 範例

val bufferedSource = file.source()/*第一層包裝*/.buffer()/*第二層包裝*/
while (!bufferedSource.exhausted()) {
    val line = bufferedSource.readUtf8Line();
    ...
}
// 省略 close

網上有資料說 Okio 沒有使用裝飾器模式，所以類結構更簡單。 這麼說其實不太準確，裝飾器模式本身並不是缺點，而且從 BufferedSource 和 BufferSink 可以看出 Okio 也使用了裝飾器模式。 嚴格來說是原生 IO 的裝飾器過於龐大，而 Okio 的裝飾器更加精簡。

比如原生 IO 常用的流就有這麼多：

• 原始流： FileInputStream / FileOutputStream 與 SocketInputStream / SocketOutputStream；

• 基礎介面（區分位元組流和字元流）： InputStream / OutputStream 與 Reader / Writer；

• 快取流： BufferedInputStream / BufferedOutputStream 與 BufferedReader / BufferedWriter；

• 基本型別： DataInputStream / DataOutputStream；

• 位元組陣列和字元陣列： ByteArrayInputStream / ByteArrayOutputStream 與 CharArrayReader / CharArrayWriter；

• 此處省略一萬個字。

原生 IO 框架

而這麼多種流在 Okio 裡還剩下多少呢？

• 原始流： FileInputStream / FileOutputStream 與 SocketInputStream / SocketOutputStream；
• 基礎介面： Source / Sink；
• 快取流： BufferedSource / BufferedSink。

Okio 框架

就問你服不服？

而且你看哈，這些都是平時業務開發中最常見的基本型別，原生 IO 把它們都拆分開了，讓問題複雜化了。反觀 Okio 直接在 BufferedSource 和 BufferedSink 中聚合了原生 IO 中基本的功能，而不再需要區分位元組、字元、位元組陣列、字元陣列、基礎型別等等裝飾器，確實讓框架更加精簡。

BufferedSource.kt

actual interface BufferedSource : Source, ReadableByteChannel {

    actual val buffer: Buffer

    // 讀取 Int
    @Throws(IOException::class)
    actual fun readInt(): Int

    // 讀取 String
    @Throws(IOException::class)
    fun readString(charset: Charset): String

    ...

    fun inputStream(): InputStream
}

BufferedSink.kt

actual interface BufferedSink : Sink, WritableByteChannel {

    actual val buffer: Buffer

    // 寫入 Int
    @Throws(IOException::class)
    actual fun writeInt(i: Int): BufferedSink

    // 寫入 String
    @Throws(IOException::class)
    fun writeString(string: String, charset: Charset): BufferedSink

    ...

    fun outputStream(): OutputStream
}

2.4 RealBufferedSink 與 RealBufferedSource

BufferedSource 和 BufferedSink 還是介面，它們的真正的實現類是 RealBufferedSource 和 RealBufferedSink。可以看到，在實現類中會建立一個 Buffer 緩衝區，在輸入和輸出的時候，都會藉助 「Buffer 緩衝區」 減少系統呼叫次數。

RealBufferedSource.kt

internal actual class RealBufferedSource actual constructor(
    // 裝飾器模式
    @JvmField actual val source: Source
) : BufferedSource {

    // 建立輸入緩衝區
    @JvmField val bufferField = Buffer()

    // 帶緩衝地讀取（全部資料）
    override fun readString(charset: Charset): String {
        buffer.writeAll(source)
        return buffer.readString(charset)
    }

    // 帶緩衝地讀取（byteCount）
    override fun readString(byteCount: Long, charset: Charset): String {
        require(byteCount)
        return buffer.readString(byteCount, charset)
    }
}

RealBufferedSink.kt

internal actual class RealBufferedSink actual constructor(
    // 裝飾器模式
    @JvmField actual val sink: Sink
) : BufferedSink {

    // 建立輸出緩衝區
    @JvmField val bufferField = Buffer()

    // 帶緩衝地寫入（全部資料）
    override fun writeString(string: String, charset: Charset): BufferedSink {
        buffer.writeString(string, charset)
        return emitCompleteSegments()
    }

    // 帶緩衝地寫入（beginIndex - endIndex)
    override fun writeString(
        string: String,
        beginIndex: Int,
        endIndex: Int,
        charset: Charset
    ): BufferedSink {
        buffer.writeString(string, beginIndex, endIndex, charset)
        return emitCompleteSegments()
    }
}

至此，Okio 基本框架分析結束，用一張圖總結：

Okio 框架

---

3. Okio 的緩衝區設計

3.1 使用緩衝區減少系統呼叫次數

在作業系統中，存取磁碟和網路卡等 IO 操作需要通過系統呼叫來執行。系統呼叫本質上是一種軟中斷，程序會從使用者態陷入核心態執行中斷處理程式，完成 IO 操作後再從核心態切換回使用者態。

可以看到，系統呼叫存在上下文切換的效能損耗。為了減少系統呼叫次數，應用層往往會採用緩衝區策略：

以 Java 原生 IO BufferedInputStream 為例，會通過一個 byte[] 陣列作為資料來源的輸入緩衝，每次讀取資料時會讀取更多資料到緩衝區中：

• 如果緩衝區中存在有效資料，則直接從緩衝區資料讀取；
• 如果緩衝區不存在有效資料，則先執行系統呼叫填充緩衝區（fill），再從緩衝區讀取資料；
• 如果要讀取的資料量大於緩衝區容量，就會跳過緩衝區直接執行系統呼叫。

輸出流 BufferedOutputStream 也類似，輸出資料時會優先寫到緩衝區，當緩衝區滿或者手動呼叫 flush() 時，再執行系統呼叫寫出資料。

虛擬碼

// 1. 輸入
fun read(byte[] dst, int len) : Int {
    // 緩衝區有效資料量
    int avail = count - pos
    if(avail <= 0) {
        if(len >= 緩衝區容量) {
            // 直接從輸入流讀取
            read(輸入流 in, dst, len)
        }
        // 填充緩衝區
        fill(資料來源 in, 緩衝區)
    }
    // 本次讀取資料量，不超過可用容量
    int cnt = (avail < len) ? avail : len?
    read(緩衝區, dst, cnt)
    // 更新緩衝區索引
    pos += cnt
    return cnt
}

// 2. 輸出
fun write(byte[] src, len) {
    if(len > 緩衝區容量) {
        // 先將緩衝區寫出
        flush(緩衝區)
        // 直接寫出資料
        write(輸出流 out, src, len)
    }
    // 緩衝區剩餘容量
    int left = 緩衝區容量 - count
    if(len > 緩衝區剩餘容量) {
        // 先將緩衝區寫出
        flush(緩衝區)
    }
    // 將資料寫入緩衝區
    write(緩衝區, src, len)
    // 更新緩衝區已新增資料容量
    count += len
}

3.2 緩衝區的副作用

的確，緩衝區策略能有效地減少系統呼叫次數，不至於讀取一個位元組都需要執行一次系統呼叫，大多數情況下表現良好。 但考慮一種 「雙流操作」 場景，即從一個輸入流讀取，再寫入到一個輸出流。回顧剛才講的快取策略，此時的資料轉移過程為：

• 1、從輸入流讀取到緩衝區；
• 2、從輸入流緩衝區拷貝到 byte[]（拷貝）
• 3、將 byte[] copy 到輸出流緩衝區（拷貝）；
• 4、將輸出流緩衝區寫入到輸出流。

如果這兩個流都使用了緩衝區設計，那麼資料在這兩個記憶體緩衝區之間相互拷貝，就顯得沒有必要。

3.3 Okio 的 Buffer 緩衝區

Okio 當然也有緩衝區策略，如果沒有就會存在頻繁系統呼叫的問題。

Buffer 是 RealBufferedSource 和 RealBufferedSink 的資料緩衝區。雖然在實現上與原生 BufferedInputStream 和 BufferedOutputStream 不一樣，但在功能上是一樣的。區別在於：

• 1、BufferedInputStream 中的緩衝區是 「一個固定長度的位元組陣列」 ，資料從一個緩衝區轉移到另一個緩衝區需要拷貝；

• 2、Buffer 中的緩衝區是 「一個 Segment 雙向迴圈連結串列」 ，每個 Segment 物件是一小段位元組陣列，依靠 Segment 連結串列的順序組成邏輯上的連續資料。這個 Segment 片段是 Okio 高效的關鍵。

Buffer.kt

actual class Buffer : BufferedSource, BufferedSink, Cloneable, ByteChannel {

    // 緩衝區（Segment 雙向連結串列）
    @JvmField internal actual var head: Segment? = null

    // 緩衝區資料量
    @get:JvmName("size")
    actual var size: Long = 0L
        internal set

    override fun buffer() = this

    actual override val buffer get() = this
}

對比 BufferedInputStream：

BufferedInputStream.java

public class BufferedInputStream extends FilterInputStream {

    // 緩衝區的預設大小（8KB）
    private static int DEFAULT_BUFFER_SIZE = 8192;

    // 輸入緩衝區（固定長度的陣列）
    protected volatile byte buf[];

    // 有效資料起始位，也是讀資料的起始位
    protected int pos;

    // 有效資料量，pos + count 是寫資料的起始位
    protected int count;

    ...
}

3.4 Segment 片段與 SegmentPool 物件池

Segment 中的位元組陣列是可以 「共用」 的，當資料從一個緩衝區轉移到另一個緩衝區時，可以共用資料參照，而不一定需要拷貝資料。

Segment.kt

internal class Segment {

    companion object {
        // 片段的預設大小（8KB）
        const val SIZE = 8192
        // 最小共用閾值，超過 1KB 的資料才會共用
        const val SHARE_MINIMUM = 1024
    }

    // 底層陣列
    @JvmField val data: ByteArra
    // 有效資料的起始位，也是讀資料的起始位
    @JvmField var pos: Int = 0
    // 有效資料的結束位，也是寫資料的起始位
    @JvmField var limit: Int = 0
    // 共用標記位
    @JvmField var shared: Boolean = false
    // 宿主標記位
    @JvmField var owner: Boolean = false
    // 後續指標
    @JvmField var next: Segment? = null
    // 前驅指標
    @JvmField var prev: Segment? = null

    constructor() {
        // 預設構造 8KB 陣列（為什麼預設長度是 8KB）
        this.data = ByteArray(SIZE)
        // 宿主標記位
        this.owner = true
        // 共用標記位
        this.shared = false
    }
}

另外，Segment 還使用了物件池設計，被回收的 Segment 物件會快取在 SegmentPool 中。SegmentPool 內部維護了一個被回收的 Segment 物件單連結串列，快取容量的最大值是 MAX_SIZE = 64 * 1024，也就相當於 8 個預設 Segment 的長度：

SegmentPool.kt

// object：全域性單例
internal actual object SegmentPool {

    // 快取容量
    actual val MAX_SIZE = 64 * 1024

    // 頭節點
    private val LOCK = Segment(ByteArray(0), pos = 0, limit = 0, shared = false, owner = false)

    ...
}

Segment 示意圖

---

4. 總結

• 1、Okio 將原生 IO 多種基礎裝飾器聚合在 BufferedSource 和 BufferedSink，使得框架更加精簡；
• 2、為了減少系統呼叫次數的同時，應用層 IO 框架會使用快取區設計。而 Okio 使用了基於共用 Segment 的緩衝區設計，減少了在緩衝區間轉移資料的記憶體拷貝；
• 3、Okio 彌補了部分 IO 操作不支援超時檢測的缺陷，而且 Okio 不僅支援單次 IO 操作的超時檢測，還支援包含多次 IO 操作的複合任務超時檢測。

關於 Okio 超時機制的詳細分析，我們在 下一篇文章 裡討論。請關注。

---

參考資料

• Github · Okio
• Okio 官網
• Okio 原始碼學習分析 —— 川峰 著
• Okio 好在哪？—— MxsQ 著