OkHttp3 kotlin版本源码分析

okhttp 责任链.png

基于okhttp4.7.2版本，该版本源码使用kotlin语言编写

1 OkHttp使用

//1.构造OkHttpClient，主要设置超时时间，cookie，自定义拦截器等，OkHttpClient一般全局单例
val okHttpClient = OkHttpClient.Builder()
    .connectTimeout(30, TimeUnit.SECONDS)//设置连接超时时间
    .readTimeout(30, TimeUnit.SECONDS)//设置读取超时时间
    .writeTimeout(30, TimeUnit.SECONDS)//设置写超时时间
    .cookieJar(CookieJar.NO_COOKIES)//设置cookie的读写
    .addInterceptor(LogInterceptor())//设置自定义日志拦截器
    .addInterceptor(ErrorInterceptor())//设置自定义错误拦截器
    .build()
  
//2.构造Request
val request = Request.Builder()
    .url("www.xxx.com/login")
    .addHeader("Content-Type", "application/x-www-form-urlencoded; charset=UTF-8")
    .get()//如果是post请求，替换为post(RequestBody)，需要传入body
    .build()
  
//3.发起请求
//同步请求
val response = okHttpClient.newCall(request).execute()
//异步请求
okHttpClient.newCall(request).enqueue(object : Callback {
    override fun onFailure(call: Call, e: IOException) {
    }
    override fun onResponse(call: Call, response: Response) {
    }
})

OkHttp的使用可以说非常简单，仅需三步：

  1. 构造个客户端；
  2. 构造个请求；
  3. 客户端发起请求；

2 源码分析

2.1 OkHttpClient

主要通过Builder模式构建OkHttpClient，里面一大堆关于Http协议的设置，不再贴代码，这里主要看一下拦截器的构造，拦截器是OkHttp的精髓。网络的连接，数据的读写，重定向等都是通过拦截器实现的。

open class OkHttpClient internal constructor(
  builder: Builder
) : Cloneable, Call.Factory, WebSocket.Factory {

  @get:JvmName("interceptors") val interceptors: List<Interceptor> =
      builder.interceptors.toImmutableList()

  class Builder constructor() {
    //...
    internal val interceptors: MutableList<Interceptor> = mutableListOf()
    
    fun addInterceptor(interceptor: Interceptor) = apply {
      interceptors += interceptor
    }    
}

2.2 Request

Http的请求报文主要包含：

  • 请求行：URL（统一资源定位符）、协议版本、请求方法（EGT、POST、DELETE、PUT等）

  • 首部行（头部）

  • 请求体（body）

Request无非就是封装了以上些信息。

也是通过Builder模式构造，大体比较简单，不再贴代码。

2.3 同步请求原理

同步请求代码如下

val response = okHttpClient.newCall(request).execute()

直接拿到response，这是一个耗时操作，须在子线程中完成。

2.3.1 OkHttpClient#newCall()

/** Prepares the [request] to be executed at some point in the future. */
override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)

newCall()方法很简单，就是返回一个RealCall实例，那么不用说，请求的逻辑都在execute()里面了。

2.3.2 RealCall#execute()

override fun execute(): Response {
  synchronized(this) {
    // 检查请求是否已执行，如果已执行就会抛出异常
    check(!executed) { "Already Executed" }
    // 标志请求开始执行
    executed = true
  }
  
  //开始超时时间定时
  timeout.enter()
  
  callStart()
  try {
    // 将当前请求加入到请求正在请求队列中
    client.dispatcher.executed(this)
      
    // 这个方法才是真正的获取Http响应
    return getResponseWithInterceptorChain()
  } finally {
    client.dispatcher.finished(this)
  }
}

/** Used by [Call.execute] to signal it is in-flight. */
@Synchronized internal fun executed(call: RealCall) {
  runningSyncCalls.add(call)
}

其中，第18行，getResponseWithInterceptorChain()里面执行了网络连接、写请求行、写首部行、写body、读响应并转化为Response类型等等。这个方法我们在后面具体分析，先看一下异步请求原理。

2.4 异步请求

异步请求代码如下：

okHttpClient.newCall(request).enqueue(object : Callback {
    override fun onFailure(call: Call, e: IOException) {
    }
    override fun onResponse(call: Call, response: Response) {
    }
})

2.4.1 RealCall#enqueue()

override fun enqueue(responseCallback: Callback) {
  synchronized(this) {
    // 和同步请求一样，检查是否已执行了请求
    check(!executed) { "Already Executed" }
    executed = true
  }
  callStart()
    
  // 调用Dispatcher的enqueue()方法
  client.dispatcher.enqueue(AsyncCall(responseCallback))
}

最后一行代码首选创建一个AsyncCall实例，再调用Dispatcher的enqueue(AsyncCall)方法。

AsyncCall是RealCall的内部类（它拥有外部类的引用）

internal inner class AsyncCall(
  private val responseCallback: Callback
) : Runnable {
  @Volatile var callsPerHost = AtomicInteger(0)
    private set
  fun reuseCallsPerHostFrom(other: AsyncCall) {
    this.callsPerHost = other.callsPerHost
  }
  val host: String
    get() = originalRequest.url.host
  val request: Request
      get() = originalRequest
  val call: RealCall
      get() = this@RealCall
  /**
   * Attempt to enqueue this async call on [executorService]. This will attempt to clean up
   * if the executor has been shut down by reporting the call as failed.
   */
  fun executeOn(executorService: ExecutorService) {
    client.dispatcher.assertThreadDoesntHoldLock()
    var success = false
    try {
      executorService.execute(this)
      success = true
    } catch (e: RejectedExecutionException) {
      val ioException = InterruptedIOException("executor rejected")
      ioException.initCause(e)
      noMoreExchanges(ioException)
      responseCallback.onFailure(this@RealCall, ioException)
    } finally {
      if (!success) {
        client.dispatcher.finished(this) // This call is no longer running!
      }
    }
  }
  
  override fun run() {
    threadName("OkHttp ${redactedUrl()}") {
      var signalledCallback = false
      timeout.enter()
      try {
        val response = getResponseWithInterceptorChain()
        signalledCallback = true
        responseCallback.onResponse(this@RealCall, response)
      } catch (e: IOException) {
        if (signalledCallback) {
          // Do not signal the callback twice!
          Platform.get().log("Callback failure for ${toLoggableString()}", Platform.INFO, e)
        } else {
          responseCallback.onFailure(this@RealCall, e)
        }
      } catch (t: Throwable) {
        cancel()
        if (!signalledCallback) {
          val canceledException = IOException("canceled due to $t")
          canceledException.addSuppressed(t)
          responseCallback.onFailure(this@RealCall, canceledException)
        }
        throw t
      } finally {
        client.dispatcher.finished(this)
      }
    }
  }
}

AsyncCall实现了Runnable接口，我们可以大胆的猜测，最终是通过调用的是它的run()方法获取Response，我们来分析一下Dispatcher#enqueue()，看看是不是这样的。

class Dispatcher constructor() {

  // 最大并发数
  @get:Synchronized var maxRequests = 64
    set(maxRequests) {
      require(maxRequests >= 1) { "max < 1: $maxRequests" }
      synchronized(this) {
        field = maxRequests
      }
      promoteAndExecute()
    }

  // 可能为null的ExecutorService
  private var executorServiceOrNull: ExecutorService? = null

  // 不为null的ExecutorService
  @get:Synchronized
  @get:JvmName("executorService") val executorService: ExecutorService
    get() {
      if (executorServiceOrNull == null) {
        executorServiceOrNull = ThreadPoolExecutor(0, Int.MAX_VALUE, 60, TimeUnit.SECONDS,
            SynchronousQueue(), threadFactory("$okHttpName Dispatcher", false))
      }
      return executorServiceOrNull!!
    }

  /** Ready async calls in the order they'll be run. */
  private val readyAsyncCalls = ArrayDeque<AsyncCall>()

  /** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
  private val runningAsyncCalls = ArrayDeque<AsyncCall>()

  /** Running synchronous calls. Includes canceled calls that haven't finished yet. */
  private val runningSyncCalls = ArrayDeque<RealCall>()

  constructor(executorService: ExecutorService) : this() {
    this.executorServiceOrNull = executorService
  }

  internal fun enqueue(call: AsyncCall) {
    synchronized(this) {
      readyAsyncCalls.add(call)

      if (!call.call.forWebSocket) {
        val existingCall = findExistingCallWithHost(call.host)
        if (existingCall != null) call.reuseCallsPerHostFrom(existingCall)
      }
    }
    promoteAndExecute()
  }

  /**
   * Cancel all calls currently enqueued or executing. Includes calls executed both
   * [synchronously][Call.execute] and [asynchronously][Call.enqueue].
   */
  @Synchronized fun cancelAll() {
    for (call in readyAsyncCalls) {
      call.call.cancel()
    }
    for (call in runningAsyncCalls) {
      call.call.cancel()
    }
    for (call in runningSyncCalls) {
      call.cancel()
    }
  }

  // 将符合条件的Call从readyAsyncCalls队列添加到runningAsyncCalls队列，
  // 并且通过线程池执行这些符合条件的Call
  private fun promoteAndExecute(): Boolean {
    this.assertThreadDoesntHoldLock()

    val executableCalls = mutableListOf<AsyncCall>()
    val isRunning: Boolean
    synchronized(this) {
      val i = readyAsyncCalls.iterator()
      // 遍历已准备就绪的任务队列
      while (i.hasNext()) {
        val asyncCall = i.next()

        // 正在执行的任务已经大于等于最大并发量，则跳出循环
        if (runningAsyncCalls.size >= this.maxRequests) break // Max capacity.
        if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue

        // 从runningAsyncCalls移除该call
        i.remove()
        asyncCall.callsPerHost.incrementAndGet()
        // 将该call添加到executableCalls和runningAsyncCalls中
        executableCalls.add(asyncCall)
        runningAsyncCalls.add(asyncCall)
      }
      isRunning = runningCallsCount() > 0
    }

    for (i in 0 until executableCalls.size) {
      val asyncCall = executableCalls[i]
      asyncCall.executeOn(executorService)
    }

    return isRunning
  }

Dispatcher是异步请求的管理类，内部包含了线程池，最大并发量等。这个类整体还是比较简单的，有一点需要说明下

executorServiceOrNull和executorService的关系

这是惰性初始化的方式：

惰性初始化是一种常见的模式，直到第一次访问该属性的时候，才根据需要创建对象的一部分，当初始化时需要消耗大量的资源并且在使用对象时并不总是需要数据时，这个非常有用。

这里使用了所谓的支持属性技术，你有一个属性，executorServiceOrNull，用来存储这个值，而另一个executorService，用来提供对属性的读取访问，你需要使用两个属性，因为属性具有不同的类型，executorServiceOrNull可以为空，executorService为非空。这种技术经常会使用到，值得熟练掌握。

可以看到enqueue()方法中，首先是将AsyncCall添加到readyAsyncCalls队列，其次，调用promoteAndExecute()方法

promoteAndExecute()方法也很简单：

  1. 遍历readyAsyncCalls，从readyAsyncCalls队列中拿出call放入runningAsyncCalls队列中，如果runningAsyncCalls队列正在执行的任务超过最大并发数，则跳出遍历；
  2. 调用每个新加入到runningAsyncCalls中的AsyncCall的executeOn方法。

第2步就回到了RealCall的内部类AsyncCall中，看看executeOn方法。

fun executeOn(executorService: ExecutorService) {
  client.dispatcher.assertThreadDoesntHoldLock()
  var success = false
  try {
    executorService.execute(this)
    success = true
  } catch (e: RejectedExecutionException) {
    val ioException = InterruptedIOException("executor rejected")
    ioException.initCause(e)
    noMoreExchanges(ioException)
    responseCallback.onFailure(this@RealCall, ioException)
  } finally {
    if (!success) {
      client.dispatcher.finished(this) // This call is no longer running!
    }
  }
}

这个方法看起来挺长的，但是值得关注的就一行executorService.execute(this)，这是线程池执行Runnable的用法，内部会调用到当前对象的run()方法。

可见我们的猜测是正确的！

最后，再来看一下run()方法

override fun run() {
  threadName("OkHttp ${redactedUrl()}") {
    var signalledCallback = false
    timeout.enter()
    try {
      val response = getResponseWithInterceptorChain()
      signalledCallback = true
      //responseCallback是我们异步传入的回调
      responseCallback.onResponse(this@RealCall, response)
    } catch (e: IOException) {
      if (signalledCallback) {
        // Do not signal the callback twice!
        Platform.get().log("Callback failure for ${toLoggableString()}", Platform.INFO, e)
      } else {
        responseCallback.onFailure(this@RealCall, e)
      }
    } catch (t: Throwable) {
      cancel()
      if (!signalledCallback) {
        val canceledException = IOException("canceled due to $t")
        canceledException.addSuppressed(t)
        responseCallback.onFailure(this@RealCall, canceledException)
      }
      throw t
    } finally {
      client.dispatcher.finished(this)
    }
  }
}

同样，这个方法看起来挺长的，值得关注的也就是try中的代码，咦，val response = getResponseWithInterceptorChain()是不是很熟悉？同步请求也是通过这个方法获取Response的。

总结一句话：无论同步or异步请求，最终都是通过getResponseWithInterceptorChain()返回响应结果。

2.5 getResponseWithInterceptorChain()

理解这个方法是理解OkHttp原理的基础

这个方法返回Response，它里面一定做了很多事情，看看它的代码：

@Throws(IOException::class)
internal fun getResponseWithInterceptorChain(): Response {
  // Build a full stack of interceptors.
  val interceptors = mutableListOf<Interceptor>()
  interceptors += client.interceptors
  interceptors += RetryAndFollowUpInterceptor(client)
  interceptors += BridgeInterceptor(client.cookieJar)
  interceptors += CacheInterceptor(client.cache)
  interceptors += ConnectInterceptor
  if (!forWebSocket) {
    interceptors += client.networkInterceptors
  }
  interceptors += CallServerInterceptor(forWebSocket)
  val chain = RealInterceptorChain(
      call = this,
      interceptors = interceptors,
      index = 0,
      exchange = null,
      request = originalRequest,
      connectTimeoutMillis = client.connectTimeoutMillis,
      readTimeoutMillis = client.readTimeoutMillis,
      writeTimeoutMillis = client.writeTimeoutMillis
  )
  var calledNoMoreExchanges = false
  try {
    val response = chain.proceed(originalRequest)
    if (isCanceled()) {
      response.closeQuietly()
      throw IOException("Canceled")
    }
    return response
  } catch (e: IOException) {
    calledNoMoreExchanges = true
    throw noMoreExchanges(e) as Throwable
  } finally {
    if (!calledNoMoreExchanges) {
      noMoreExchanges(null)
    }
  }
}

kotlin中，MutableListOf的+=实际就是add()/addAll()，Ctrl+鼠标左键点击+号，可以看到operator方法的定义哦

4~13行：

第5行， client.interceptors实际就是我们通过OkHttpClient.addInterceptor()添加的拦截器。

然后又添加了几个OkHttp实现的Interceptor，所有的interceptor都保存在集合interceptors中。

todo 他们的顺序不能乱。

14行，构造了一个RealInterceptorChain，这个参数我们先只关注index这个参数，初始的index是0，通过后面第26行和第31行知道，最终返回的response是通过chain.proceed(originalRequest)方法返回的，继续跟进这个方法

@Throws(IOException::class)
override fun proceed(request: Request): Response {
  ...省略部分代码
  // Call the next interceptor in the chain.
  val next = copy(index = index + 1, request = request)
  val interceptor = interceptors[index]
  @Suppress("USELESS_ELVIS")
  val response = interceptor.intercept(next) ?: throw NullPointerException(
      "interceptor $interceptor returned null")
  ...省略部分代码
  return response
}

这个方法中我们去掉了检查抛出异常的相关代码让结构看起来更清晰。

proceed()方法中，会构造一个index=index+1的RealInterceptorChain next，并调用interceptors[index].intercept(next)，Interceptor中又会调用RealInterceptorChain.proceed()（除了最后一个interceptor，最后一个interceptor是CallServerInterceptor，后面单独讲解），这样一直顺序调用，直到调用到最后一个interceptor的intercept()方法，这个过程用下面这个图可以说明更加清晰。

okhttp 责任链.png

图中实箭头代表方法调用方向，虚箭头代表方法返回方向。

getResponseWithInterceptorChain()通过chain0获取Response，chain0通过interceptors[0]获取，interceptors[0]又通过chain1获取.....，这是责任链模式。

为什么最后一个interceptor（CallServerInterceptor）的处理和前面不一样呢？因为前面的interceptor的intercept()方法返回的Response是依赖下一个RealInterceptorChain.proceed()的返回，而RealInterceptorChain.proceed()依赖于Interceptor.intercept()的返回，最后一个intercept没有下一个了，所以它必须担任起Response的构造，后面对CallServerInterceptor专门分析。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 203,772评论 6赞 477
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,458评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,610评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,640评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,657评论 5赞 365
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,590评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,962评论 3赞 395
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,631评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,870评论 1赞 297
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,611评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,704评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,386评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,969评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,944评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,179评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,742评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,440评论 2赞 342