IO系列 | 一文掌握OKHTTP中的OKIO为什么这么OK

前言

本篇是 IO系列 的第4篇，前三篇文章中，我们已经对JAVA经典IO设计、JAVA－NIO内容、操作系统IO架构基础概念、Zero－Copy做了较为系统的回顾。

而绝大部分Android应用中都会涉及到网络模块，Retrofit、Okhttp 几乎是必用的框架， Okio 作为 Okhttp 中的重要模块，原先用于处理网络模块中的IO问题，随着其项目发展，Okio也开始面向其他问题。

这一篇，我们一同对OKIO做一次系统的梳理，搞明白OKIO为什么OK，做到在面试中自如的吹牛批、在日常工作中灵活使用。

编者按：面试吹牛批需要把握尺度，避免远超岗位预期，导致浪费时间

因文章篇幅较长，可结合内容导图阅读：

okio的主旨与架构

在OKIO项目的 wiki 中，对其主旨有如下介绍：

Okio is a library that complements java.io and java.nio to make it much easier to access, store, and process your data. It started as a component of OkHttp, the capable HTTP client included in Android. It’s well-exercised and ready to solve new problems.

简单直译为中文如下：

Okio是一个类库，对 java.io 和 java.nio 进行了补充，使得访问、存储和处理数据变得更加容易。它最初是OkHttp的一个组件，OkHttp是安卓中的一个功能强大的HTTP客户端。它非常健壮，可以解决新问题。

简言之：为了更简单的访问、存储、处理数据,基于 java.io 和 java.nio 进行了功能补充

wiki中，简单介绍了设计中的几个重点角色：

• ByteStrings and Buffers
• Sources and Sinks

分层架构中相对扁平、简单：在应用和Java IO 之间增加了一层，即OKIO本身，包含 数据封装 、 输入输出 、 超时机制

体现在类图上还是比较复杂的：

在库内部，ByteStrings 的使用不多，对 Buffer 数据包装后为上层应用服务，单独拎出。

信息噪声比较多，去掉功能装饰的实现类后较为精简：

与Java的输入输出的对比

Java经典IO中的输入输出定义为Stream，在 系列文章 中进行了介绍。字符流类似，图略

在JDK的IO框架中，使用装饰模式建立了规模庞大、功能丰富的输入输出流。从OKIO的主旨出发，不难理解其设计者希望类库尽可能简单、易扩展、内建部分功能足够完善。因此，OKIO会适当的另起炉灶，不会全面的使用JDK中的Stream。

OKIO中使用了自定义的输入、输出，即 Source 和 Sink ，注意淡黄色、淡粉色部分：

Sink 在计算机领域有特定含义：指程序或者线程，可以接收数据，并且可以处理或者发送这些数据

差异点

在wiki中提到如下内容：

An elegant part of the java.io design is how streams can be layered for transformations like encryption and compression. Okio includes its own stream types called Source and Sink that work like InputStream and OutputStream, but with some key differences:

• Timeouts.
• Easy to implement.
• Easy to use.
• No artificial distinction between byte streams and char streams.
• Easy to test.

简单翻译下， Java IO的设计中有一处非常优雅：可以调整流的分层包装以实现加密、压缩等转换。OKIO包含自有的流类型 Source、Sink，与Java的 InputStream、 OutputStream 功能类似，但是有几点关键的不同：

• 超时机制
• 更容易实现
• 更容易使用
• 字节流、字符流之间没有人为的差异
• 更容易测试

从输入方面看：

在JDK中，InputStream 使用多种层（以及复合层）处理种类繁多的各类数据

• DataInputStream 用于处理基础数据类型
• BufferedInputStream 处理缓冲
• InputStreamReader 处理文本字符串

而OKIO在这些层之上建立了 BufferedSource，Source避免了一些无法实现 available() 方法的困境, 转而由调用者指定它们需要的byte个数

在实现一个Source时，不必操心 read() 方法，它难以有效实现且需从257种值中返回一个 ，注：null & [0x00，0xFF]

从输出方面看：

类似的，在JDK中 OutputStream 使用多种层（以及复合层）处理种类繁多的各类数据，而Sink也非常容易采用分层设计

相同点

• Source 、 Sink 的功能与 InputStream 、 OutputStream 、 Reader 、 Writer 相同
• 使用时可以通过装饰追加功能

对于功能相同，wiki中提到如下内容：

Sources and sinks interoperate with InputStream and OutputStream. You can view any Source as an InputStream, and you can view any InputStream as a Source. Similarly for Sink and OutputStream.

下文的Source、Sink详解中，解析他们与IOStream 为何 “等价”、如何“互操作”

Source、Sink 详解

Source 体系

抛开功能类（压缩、哈希、加密、装饰等），主要关注：

• Source
• BufferedSource
• Buffer
• RealBufferedSource

Source的定义中规中矩：

interface Source : Closeable {

  @Throws(IOException::class)
  fun read(sink: Buffer, byteCount: Long): Long

  fun timeout(): Timeout

  @Throws(IOException::class)
  override fun close()
}

其中的 timeout 将在下文超时机制章节中展开。

BufferedSource接口约定使用 Buffer 承接实际数据，并且定义了一系列方便使用的接口，如：

• 读取Int
• 读取Short
• 读取字符串
• 内容选择 等，不做罗列

RealBufferedSource 实现了 BufferedSource 接口，从本质上可以认为是 Buffer 类的代理，增加了边界校验

Buffer 实现了 BufferedSource 接口，包含 读 的具体实现

Sink 体系

抛开功能类（压缩、哈希、加密、装饰等），主要关注：

• Sink
• BufferedSink
• Buffer
• RealBufferedSink

interface Sink : Closeable {
  @Throws(IOException::class)
  fun write(source: Buffer, byteCount: Long)

  @Throws(IOException::class)
  fun flush()

  fun timeout(): Timeout

  @Throws(IOException::class)
  override fun close()
}

同样，Sink接口的定义也中规中矩。

类似的，BufferedSink 接口继承了 Sink 接口，约定了使用 Buffer 承接实际数据, RealBufferedSink 是具体实现，从本质上是 Buffer 作为 Sink 时的代理，进行了边界校验。

Buffer 实现了 BufferedSink 接口，包含 写 的具体实现

Source、Sink与I/O-Stream的互操作

作者按：请仔细思考一下 互操作 ，其本质是：使用一种实例对象的API去操作另一种对象实例的API, 请留意直接操作和间接操作，国内程序员更习惯使用 “转”、“转换” ，着眼点是从一种实例对象获得另一种实例对象。 从转换角度思考时，容易陷入误区，枚举出没必要地转换情况、忽略掉必要的间接转换。

首先明确一点：在基于I/O-Stream进行读写时，InputStreamSource、 OutputStreamSink 是 I/O-Stream的读写代理； I/O-Stream 是 InputStreamSource、 OutputStreamSink 的读写委托。

依靠转换API：

fun InputStream.source(): Source = InputStreamSource(this, Timeout())
fun OutputStream.sink(): Sink = OutputStreamSink(this, Timeout())

排除此 “基本情况1” ，还有一种 基本情况2 ，即我们希望使用I/O-Stream的API去实质操作OKIO。不难想象，此时必然存在用OKIO实现的定制业务，即数据的实质处理，OKIO选择了 BufferedSink 和 BufferedSource 而非 Sink 、 Source，此时 BufferedSink 和 BufferedSource 是 I/O-Stream 的业务委托。

API如下：

interface BufferedSink : Sink, WritableByteChannel {
   //... ignore

  /** Returns an output stream that writes to this sink. */
  fun outputStream(): OutputStream
}

interface BufferedSource : Source, ReadableByteChannel {
  //... ignore

  /** Returns an input stream that reads from this source. */
  fun inputStream(): InputStream
}

更复杂的情况 -- 基于基本情况加各类业务的组合

排除掉这些基本情况，还需要需要互操作时，意味着存在两套模块，一套使用Okio的Source、Sink，一套使用Java的InputStream、OutputStream。假定使用Sink、Source的模块为A，使用I/O-Stream的模块为B。

以写为例，进行分析，有两种可能：

• 操作模块A，数据从A流向B，进而写入 (一般来说，B存在业务定制，否则是基本情况1)
• 操作模块B，数据从B流向A，进而写入（A最终使用了基本情况1，此情况属于基本情况2的复杂版）

1. 数据从A流向B，进而写入

关注A模块的数据出口，A模块可能的设计

• A1:接受一个Sink实例，或者接受一些参数构建Sink实例
• A2:对外提供一个Source实例，由外界自行消费 -- 这种设计思路比较奇葩
• 接受回调函数，提供实际数据 -- 不属于实际讨论范畴，忽略

关注B模块的数据入口，B模块可能的设计

• B1:接受一个InputStream实例，或者接受一些参数构建InputStream实例，消费其数据 -- 这种设计思路比较奇葩
• B2:对外提供一个OutputStream实例，由外界控制实际写
• 暴露使用ByteArray等写入API -- 不属于实际讨论范畴，忽略

A1+B1 情况的伪代码

A1:接受一个Sink实例，或者接受一些参数构建Sink实例

B1:接受一个InputStream实例，或者接受一些参数构建InputStream实例，消费其数据

class Client {
    fun demo() {
        //使用Sink、Source的模块A
        val moduleA = A()

        //使用I/O-Stream的模块B
        val moduleB = B()

        val buffer = Buffer()

        //设置模块A的数据出口
        moduleA.setSink(buffer)
        //设置模块B的数据入口
        moduleB.setInputStream(buffer.inputStream())

        //假定最终写入file：
        moduleB.setOutputFile(File("/target.txt"))

        //调用模块A开始写入
        moduleA.write("some thing")
    }
}

值得注意的是，需要在Buffer区的数据被消费后，进行清理，以避免内存占用越来越多，而因为B模块的奇葩设计，往往带入多线程问题，编程难度较大

A1+B2 情况的伪代码

A1:接受一个Sink实例，或者接受一些参数构建Sink实例

B2:对外提供一个OutputStream实例，由外界控制实际写

class Client {
    fun demo() {
        //使用Sink、Source的模块A
        val moduleA = A()

        //使用I/O-Stream的模块B 
        val moduleB = B()
        //假定其写入方式为：moduleB.writer().write("xxx")， moduleB.writer()获得OutputStream实例
        //假定最终写入file：
        moduleB.setOutputFile(File("/target.txt"))

        //使用B的入口，套接到模块A的数据出口
        moduleA.setSink(moduleB.writer().sink())

        //调用模块A开始写入
        moduleA.write("some thing")
    }
}

A2+B1 情况的伪代码

A2:对外提供一个Source实例，由外界自行消费

B1:接受一个InputStream实例，或者接受一些参数构建InputStream实例，消费其数据

class Client {
    fun demo() {
        //使用Sink、Source的模块A
        val moduleA = A()

        //使用I/O-Stream的模块B 
        val moduleB = B()
        //假定最终写入file：
        moduleB.setOutputFile(File("/target.txt"))

        //此情况需要moduleA.getSource() 提供 BufferedSource实例，
        //如果实现了Sink而并未实现BufferedSource, 需要模块提供者自己考虑接口系统的转换
        moduleB.setInputStream(moduleA.getSource().inputStream())

        //调用模块A开始写入
        moduleA.write("some thing")
    }
}

A2+B2 情况的伪代码

A2:对外提供一个Source实例，由外界自行消费

B2:对外提供一个OutputStream实例，由外界控制实际写

class Client {
    fun demo() {
        //使用Sink、Source的模块A
        val moduleA = A()

        //使用I/O-Stream的模块B 
        val moduleB = B()
        //假定最终写入file：
        moduleB.setOutputFile(File("/target.txt"))

        val resultFromA = moduleA.getSource()

        val buffer = Buffer()
        while (resultFromA.read(buffer, count) != -1) {
            buffer.writeTo(moduleB.writer())
            buffer.clear()
        }

        //调用模块A开始写入
        moduleA.write("some thing")
    }
}

2. 数据从B流向A，进而写入

关注A模块的数据入口，A模块可能的设计

• A1:接受一个Source实例，或者接受一些参数构建Source实例 -- 这种设计思路比较奇葩，在主动拥抱复杂
• A2:对外提供一个Sink实例，由外界自行控制写入
• 暴露使用ByteArray等写入API -- 不属于实际讨论范畴，忽略

关注B模块的数据出口，B模块可能的设计

• B1:接受一个OutputStream实例，或者接受一些参数构建OutputStream实例
• B2:对外提供一个InputStream实例，由外界自行消费数据 -- 这种设计思路比较奇葩
• 接受回调函数，提供实际数据 -- 不属于实际讨论范畴，忽略

简单归纳伪代码如下

//A1 + B1 同样会有内存释放、多线程编程难度问题
val buffer = Buffer()
moduleA.setSource(buffer)
moduleB.setWriter(buffer.outputStream())

//A1 + B2
// moduleB.getResult() 返回InputStream实例
moduleA.setSource(moduleB.getResult().source())

//A2 + B1
moduleB.setWriter(moduleA.getSink().outputStream())

//A2 + B2
val resultFromB:InputStream = moduleB.getResult()
val buffer = ByteArray(1024)
while ((len = in.read(buffer))!=-1) {
    moduleA.sink().write(buffer,0,len)
}

很显然，OKIO提供的转化方式，能够满足正常的设计，而剩余的奇葩设计，自然需要设计者自行处理内存、多线程问题。

而读的例子也是类似的，考虑到篇幅已经很长，读者诸君可以自行梳理。

在读和写都能够完成两套系统的互操作时，即可随心随意地构建出更加复杂的层叠layer，亦不再展开

Buffer 详解

顾名思义，OKIO 中的 Buffer 是特意设计的缓冲区。它在 数据处理 和 数据读写 之间进行缓冲

class Buffer : BufferedSource, BufferedSink, Cloneable, ByteChannel

它的设计意图可以概括为三个方面，这三方面并不孤立互斥：

• 内存中的读写缓冲区
• 更方便的API
• 和 Java IO 互操作

更方便的API：系列文章中提到过，数据的实质内容可以编码成ByteArray，Buffer提供了更方便的编解码

和 Java IO 互操作：除了上文中已经提及的内容，还包含 readFrom(input: InputStream) , fun writeTo(out: OutputStream, byteCount: Long = size) 等API，这些也可以算为更方便的API

Buffer实现了ByteChannel接口，可以适应NIO的设计体系，当然此时它又是内存中的读写缓冲区

API方面，读者诸君可自行研读代码，内容比较简单。让我们将精力放在 缓冲区 上，看一看它的实现原理。

核心实现

移除掉干扰代码后, 可以发现它的重点为 head: Segment ，代码简单扫一眼有个印象即可：

class Buffer : /*BufferedSource, BufferedSink,*/ Cloneable, ByteChannel {
  internal var head: Segment? = null

  var size: Long = 0L
    internal set

  @Throws(EOFException::class)
  override fun readByte(): Byte = commonReadByte()
  //和下面的read类似，仅作为Source的API示例

  @Throws(IOException::class)
  override fun read(sink: ByteBuffer): Int {
    val s = head ?: return -1

    val toCopy = minOf(sink.remaining(), s.limit - s.pos)
    sink.put(s.data, s.pos, toCopy)

    s.pos += toCopy
    size -= toCopy.toLong()

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

    return toCopy
  }

  override fun write(source: ByteArray): Buffer = commonWrite(source)
  //也用到了writableSegment(1)，可类比下面的write代码，仅留作Sink的API示例

  @Throws(IOException::class)
  override fun write(source: ByteBuffer): Int {
    val byteCount = source.remaining()
    var remaining = byteCount
    while (remaining > 0) {
      val tail = writableSegment(1)

      val toCopy = minOf(remaining, Segment.SIZE - tail.limit)
      source.get(tail.data, tail.limit, toCopy)

      remaining -= toCopy
      tail.limit += toCopy
    }

    size += byteCount.toLong()
    return byteCount
  }
}

而Segment是什么呢？是一个链表数据结构：

internal class Segment {
  @JvmField val data: ByteArray
  @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

  //ignore
}

包含了三种内容：

• 实际数据 data
• 标记 pos+limit和读写有关，shared、owner和数据保护有关
• 上下游节点，链表的本质

而 SegmentPool 是很典型的池化设计，毕竟 ByteArray 需要分配内存空间，使用池化可以很好地减少无效内存管理（频繁分配回收）

不难得出总结：

• Segment 是实现了方便使用的API的ByteArray链表
• Buffer 是使用 Segment 包装而成的数据缓冲区，实现了方便使用的API，实现了和JAVA IO间的互操作
• Buffer 既可以作为上游 Source 角色，也可以作为下游 Sink 角色
• 在设计 Buffer 和 Segment 时，结合了日常使用场景进行了特定优化，例如通过转移、分享数据而非拷贝实质数据

ByteString 概述

在Okio中，ByteString也是一个重要的、方便使用的设计，但比Buffer简单的多。

它的命名也非常有趣，计算机领域中，String一词对应字符串，它本身拥有一个更宽泛的含义就是“一串”。 在不足够严谨的讨论场景下，我们可以认为 String 就是指定了编码的CharArray或ByteArray。而CharSequence又太过于抽象，设计者似乎仅希望将一些特定的具体情况进行封装，因此制造了ByteString。

open class ByteString
internal actual constructor(
  internal actual val data: ByteArray,
) : Serializable, Comparable<ByteString> {}

作者按：我在开发一些蓝牙应用时，数据传输层和应用协议层有一些特定的数据操作，例如：信息摘要计算、CRC校验、AES加解密、Hex-String转换用作日志输出。在很多年前才接触Android时，使用HttpUrlConnection，对body也有类似的处理。属实枯燥繁琐

它是一个内容不可变byte序列，这一点可以通过观测API发现，它并不提供修改内容的相关API。但它封装了一系列好用的API，例如：

• utf8(): String
• string(charset: Charset)
• fun base64()
• fun md5() = digest("MD5")
• fun sha1() = digest("SHA-1")
• digest(algorithm: String): ByteString 等等

相比于定义与使用Utility类，代码可读性更强。

值得注意 ：虽然它在设计意图上是内容不可变的，但注意它的构造函数，它只保留了引用，并没有对内容进行拷贝，这意味实质内容可以在外部篡改

它实现了 Comparable 接口，值得一提的是，它按照无符号数大小进行 “字典序” 比对。

• "字典序" 比对，即按照从头到尾的顺序，依次比对，脑补一下英文词典。
• Byte使用8bit表示，0xFF（补码）如果视作符号数为255，排在0x00后面，如果视作有符号数，则为-1，排在0x00前面

超时机制

简单思考一下，你的BOSS是如何按照Deadline来检查你的工作的。

如果你没有提前告知已完成，

• 最理想的BOSS会在到期时查你
• 宽松一点的BOSS会在Deadline当天或提前一天过问一下，到点再查一下
• 焦虑一点的BOSS会频繁一点
• 有毛病的BOSS会一天到晚盯着你

显然，需要先约定一个超时的信息：

class Timeout {
  private var hasDeadline = false
  private var deadlineNanoTime = 0L
  private var timeoutNanos = 0L
}

假定有一项具体工作，当你和BOSS约定好时间，他会记录这一信息，得到一个timeout，当然，并非所有事情都会有Deadline

此时，你去执行这一事项：

timeout.withTimeout {
    //具体的事项
}

而你的BOSS，则会根据是否有真实Deadline，决定是否记录到他的检查单上。

很显然，你的BOSS需要跟踪的事项进度比较多，他按照到期时间先后顺序对检查单内容进行整理，这样他就省事了，他只需要盯着第一个到期时间进行追踪即可。

当发现超时时，他会将这一项移除，调整他的检查单，并通报此项已经超时...

不难想象，如果他的检查单上没有追踪项，他不会给自己来一个遥遥无期的休假，否则有后续事项没被跟踪，他就惨了，但一直盯着有没有新事项产生会很累，所以他每两小时就会看一下，是否有事项需要写入检查单。即便写入检查单时此事已经延期，但一个2小时内就会到Deadline的事情，稍微拖延了一会去追查，也没啥毛病。

将BOSS的这部分工作写成代码如下：

再回到你这边，如果你的工作完成了，你去会找BOSS汇报工作，让他处理检查单：

inline fun <T> withTimeout(block: () -> T): T {
    var throwOnTimeout = false
    enter()
    try {
        val result = block()
        throwOnTimeout = true
        return result
    } catch (e: IOException) {
        throw if (!exit()) e else `access$newTimeoutException`(e)
    } finally {
        // 找BOSS汇报工作，让他处理检查单
        val timedOut = exit()
        if (timedOut && throwOnTimeout) throw `access$newTimeoutException`(null)
    }
}

此时轮到BOSS来处理，如果这个事项并没有真实Deadline，他并不会额外做什么。否则他会重新维护检查单内容，但如果没有在检查单中发现这一项，则说明该项在之前通报过已经超时。

fun exit(): Boolean {
    return cancelScheduledTimeout(this)
}

private fun cancelScheduledTimeout(node: AsyncTimeout): Boolean {
    AsyncTimeout.lock.withLock {
        if (!node.inQueue) return false
        node.inQueue = false

        // Remove the node from the linked list.
        var prev = head
        while (prev != null) {
            if (prev.next === node) {
                prev.next = node.next
                node.next = null
                return false
            }
            prev = prev.next
        }

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

我们通过一个故事演示了一种异步的超时检测机制，在Okio中，对应了 AsyncTimeout 。当然，实际场景中还有一些更复杂的，例如两个事项合并。

甚至，我们可以直接借用此机制：

演示代码：

以一个效率低下的递归计算斐波那契数列进行演示(如果你的机器性能异常的好，可以适当调大入参)：

Demo 代码 :

结语

至此，IO系列告一段落。

按照惯例，再絮叨几句。在去年放缓节奏后，读了一些书、想了一些事、观了一些人。近期于《孟子》中得一句：

或劳心，或劳力；劳心者治人，劳力者治于人；治于人者食人，治人者食于人：天下之通义

下个系列还在构思斟酌中，下个月再见。