最近，很多朋友反映大厂的面试喜欢挖底层知识，像OkHttp这些都是必问的问题。这里就给大家分享一篇非常有帮助的技术文吧。
OkHttp是一个高效的HTTP库：
支持HTTP / 2，允许对同一主机的所有请求共享一个套接字通过连接池可减少请求延迟（如果HTTP / 2不可用）支持GZIP压缩减少数据流量响应缓存可以完全避免 *** 重复请求静默恢复处理常见的连接问题…本文就以请求使用为入口，来深入学习下OkHttp。
一、请求流程分析
1. 同步请求
Okhttp同步GET请求使用：
// 新建一个Okhttp客户端（也可以通过OkHttpClient.Builder来构造） OkHttpClient client = new OkHttpClient(); // 构造一个请求对象 Request request = new Request.Builder().url(url).build(); // 执行同步请求，返回响应 Response response = client.newCall(request).execute(); // 从响应体中获取数据 String str = response.body().string();
先来瞧瞧构建OkhttpClient的源码：
open class OkHttpClient internal constructor( builder: Builder ) : Cloneable, Call.Factory, WebSocket.Factory { //若直接实例化OkHttpClient，则调用主构造函数以默认Builder作为参数 constructor() : this(Builder()) // 通过builder中的值赋值 @get:JvmName("dispatcher") val dispatcher: Dispatcher = builder.dispatcher @get:JvmName("connectionPool") val connectionPool: ConnectionPool = builder.connectionPool ... class Builder constructor() { //分发器 internal var dispatcher: Dispatcher = Dispatcher() //连接池 internal var connectionPool: ConnectionPool = ConnectionPool() //应用拦截器*** internal val interceptors: MutableList<Interceptor> = mutableListOf() // *** 拦截器*** internal val networkInterceptors: MutableList<Interceptor> = mutableListOf() //事件监听工厂 internal var eventListenerFactory: EventListener.Factory = EventListener.NONE.asFactory() //连接失败是否重试 internal var retryOnConnectionFailure = true internal var authenticator: Authenticator = Authenticator.NONE //是否跟随重定向 internal var followRedirects = true internal var followSslRedirects = true //cookie internal var cookieJar: CookieJar = CookieJar.NO_COOKIES //磁盘缓存 internal var cache: Cache? = null //dns internal var dns: Dns = Dns.SYSTEM // *** 设置 internal var proxy: Proxy? = null ... } }
OkhttpClient可以通过构造者配置参数来构建，也可以直接实例化，直接实例化其实也是内部调用构造者，只是传入的是默认builder。
再来看看OkhttpClient的newCall ***
Override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)
发现返回的是RealCall，接下来去RealCall中看后续的execute执行 ***
override fun execute(): Response { //确认call没有执行过并置executed为true，否则抛出异常 check(executed.compareAndSet(false, true)) { "Already Executed" } timeout.enter() callStart() try { //标记执行中 client.dispatcher.executed(this) //通过拦截器链获取 *** 响应 return getResponseWithInterceptorChain() } finally { //标记执行结束 client.dispatcher.finished(this) } }
看起来比较精简，通过拦截器链获取 *** 响应，然后返回响应（拦截器链路在后续拦截器分析）。
2. 异步请求
Okhttp异步GET请求使用：
OkHttpClient client = new OkHttpClient(); Request request = new Request.Builder().url(url).build(); // 执行异步请求，通过回调返回响应 client.newCall(request).enqueue(new Callback() { @Override public void onFailure(@NotNull Call call, @NotNull IOException e) {} @Override public void onResponse(@NotNull Call call, @NotNull Response response) Throws IOException { // 从回调中通过响应体获取数据 String str = response.body().string(); } });
异步请求流程大致和同步请求相似，但是最后的执行 *** 是enqueue，并传入回调对象。
我们来看看源码：
override fun enqueue(responseCallback: Callback) { //确认call没有执行过并置executed为true，否则抛出异常 check(executed.compareAndSet(false, true)) { "Already Executed" } //监听回调 callStart() //调用Dispatcher的enqueue *** ，并传入一个AsyncCall对象 client.dispatcher.enqueue(AsyncCall(responseCallback)) }
内部调用了客户端的分发器的enqueue *** ，并把AsyncCall(responseCallback)作为参数传入，AsyncCall是继承自Runnable，且是RealCall的内部类，我们先看Dispatcher.enqueue() ***
class Dispatcher constructor() { /** '异步准备执行'队列 */ private val readyAsyncCalls = ArrayDeque<AsyncCall>() /** '异步正在执行'队列，包括取消但至今还没结束的 */ private val runningAsyncCalls = ArrayDeque<AsyncCall>() /** ‘同步正在执行’队列*/ private val runningSyncCalls = ArrayDeque<RealCall>() ... internal fun enqueue(call: AsyncCall) { synchronized(this) { //加入准备执行队列 readyAsyncCalls.add(call) ... } // 执行 promoteAndExecute() } 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 //检查请求是否超过一个Host对应的更大请求数 if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue i.remove() asyncCall.callsPerHost.incrementAndGet() executableCalls.add(asyncCall) runningAsyncCalls.add(asyncCall) } isRunning = runningCallsCount() > 0 } for (i in 0 until executableCalls.size) { val asyncCall = executableCalls[i] //调用AsyncCall的executeOn() *** asyncCall.executeOn(executorService) } return isRunning } }
可以从上面代码看出，就是将符合条件的调用从readyAsyncCalls队列提升到runningAsyncCalls，并调用 AsyncCall的executeOn() *** ，把线程池传入。
我们来看看AsyncCall：
inner class AsyncCall( private val responseCallback: Callback ) : Runnable { ... fun executeOn(executorService: ExecutorService) { client.dispatcher.assertThreadDoesntHoldLock() var success = false try { //使用线程池执行自己的run() *** executorService.execute(this) success = true } catch (e: RejectedExecutionException) { ... //失败回调 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) { ... } else { //回调失败 responseCallback.onFailure(this@RealCall, e) } } catch (t: Throwable) { cancel() if (!signalledCallback) { ... //回调失败 responseCallback.onFailure(this@RealCall, canceledException) } throw t } finally { //标记结束 client.dispatcher.finished(this) } } } }
使用线程池来执行自己，接下来就看run() *** ，发现和同步请求一样，通过拦截器链获取 *** 响应，再调用回调对象的回调 *** 返回响应。
二、拦截器分析
请求大致流程知道了，我们来看看重头戏，拦截器链里面做了什么操作。
@Throws(IOException::class) internal fun getResponseWithInterceptorChain(): Response { // 建立一个拦截器列表 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) } } }
我们在看下RealInterceptorChain的proceed *** ：
@Throws(IOException::class) override fun proceed(request: Request): Response { ... // ***一个RealInterceptorChain，用于调用链中的下一个拦截器 val next = copy(index = index + 1, request = request) val interceptor = interceptors[index] @Suppress("USELESS_ELVIS") // 调用下一个拦截器的intercept *** ，获取response返回给上一个拦截器 val response = interceptor.intercept(next) ?: throw NullPointerException( "interceptor $interceptor returned null") ... return response }
接下来我们来具体看看各个拦截器的作用
1. RetryAndFollowUpInterceptor
RetryAndFollowUpInterceptor 处理错误恢复和重定向，它会判断错误是否满足条件进行重试，还有根据返回的响应判断是否需要重定向请求。
class RetryAndFollowUpInterceptor(private val client: OkHttpClient) : Interceptor { @Throws(IOException::class) override fun intercept(chain: Interceptor.Chain): Response { val realChain = chain as RealInterceptorChain var request = chain.request val call = realChain.call var followUpCount = 0 var priorResponse: Response? = null var newExchangeFinder = true var recoveredFailures = listOf<IOException>() while (true) { // 初始化ExchangeFinder(后续ConnectInterceptor会用到ExchangeFinder来查找连接) call.enterNetworkInterceptorExchange(request, newExchangeFinder) var response: Response var closeActiveExchange = true try { if (call.isCanceled()) { throw IOException("Canceled") } try { //执行下一个拦截器，获取响应 response = realChain.proceed(request) newExchangeFinder = true } catch (e: RouteException) { ... // 满足条件则重试 continue } catch (e: IOException) { ... // 满足条件则重试 continue } // 赋上重定向之前的响应（响应体置空） if (priorResponse != null) { response = response.newBuilder() .priorResponse(priorResponse.newBuilder() .body(null) .build()) .build() } val exchange = call.interceptorScopedExchange //判断是否需重定向，若需则返回重定向请求 val followUp = followUpRequest(response, exchange) //不需要重定向则直接返回response if (followUp == null) { if (exchange != null && exchange.isDuplex) { call.timeoutEarlyExit() } closeActiveExchange = false return response } val followUpBody = followUp.body // 若该请求只可传输一次，则返回响应 if (followUpBody != null && followUpBody.isOneShot()) { closeActiveExchange = false return response } response.body?.closeQuietly() //超过重定向更大次数则抛出异常 if (++followUpCount > MAX_FOLLOW_UPS) { throw ProtocolException("Too many follow-up requests: $followUpCount") } //将请求重新赋值为重定向的请求，继续循环，再次发送 request = followUp priorResponse = response } finally { call.exitNetworkInterceptorExchange(closeActiveExchange) } } } ... }
2. BridgeInterceptor
BridgeInterceptor 桥接应用层和 *** 层的代码，对用户的请求进行加工（如对请求头进行设置添加），也对 *** 响应做相应的处理（如解压服务端返回的 gzip 压缩数据）。
class BridgeInterceptor(private val cookieJar: CookieJar) : Interceptor { @Throws(IOException::class) override fun intercept(chain: Interceptor.Chain): Response { // 获取用户请求 val userRequest = chain.request() // 真正发送的 *** 请求的构建者 val requestBuilder = userRequest.newBuilder() // 用户请求的请求体 val body = userRequest.body // 对请求头的设置 ... var transparentGzip = false if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) { transparentGzip = true requestBuilder.header("Accept-Encoding", "gzip") } val cookies = cookieJar.loadForRequest(userRequest.url) if (cookies.isNotEmpty()) { requestBuilder.header("Cookie", cookieHeader(cookies)) } if (userRequest.header("User-Agent") == null) { requestBuilder.header("User-Agent", userAgent) } // 执行下一个拦截器，获取 *** 响应 val networkResponse = chain.proceed(requestBuilder.build()) cookieJar.receiveHeaders(userRequest.url, networkResponse.headers) val responseBuilder = networkResponse.newBuilder() .request(userRequest) // 若因配置问题，服务端返回gzip压缩的数据，则做相应的解压缩 if (transparentGzip && "gzip".equals(networkResponse.header("Content-Encoding"), ignoreCase = true) && networkResponse.promisesBody()) { val responseBody = networkResponse.body if (responseBody != null) { // GzipSource对象，用于解压 val gzipSource = GzipSource(responseBody.source()) val strippedHeaders = networkResponse.headers.newBuilder() .removeAll("Content-Encoding") .removeAll("Content-Length") .build() responseBuilder.headers(strippedHeaders) val contentType = networkResponse.header("Content-Type") responseBuilder.body(RealResponseBody(contentType, -1L, gzipSource.buffer())) } } return responseBuilder.build() } }
3. CacheInterceptor
CacheInterceptor 承担着缓存的查找与保存的职责。根据策略判断是使用缓存还是走 *** 请求，对于返回的响应，满足条件则进行缓存。
class CacheInterceptor(internal val cache: Cache?) : Interceptor { @Throws(IOException::class) override fun intercept(chain: Interceptor.Chain): Response { val call = chain.call() val cacheCandidate = cache?.get(chain.request()) val now = System.currentTimeMillis() // 检查缓存策略 val strategy = CacheStrategy.Factory(now, chain.request(), cacheCandidate).compute() // 若还需发送 *** 请求，则networkRequest不为空 val networkRequest = strategy.networkRequest // 若存在可用缓存，则cacheResponse不为空 val cacheResponse = strategy.cacheResponse ... // 如果我们被禁止使用 *** ，并且无可用缓存，则返回失败 if (networkRequest == null && cacheResponse == null) { return Response.Builder() .request(chain.request()) .protocol(Protocol.HTTP_1_1) .code(HTTP_GATEWAY_TIMEOUT) .message("Unsatisfiable Request (only-if-cached)") .body(EMPTY_RESPONSE) .sentRequestAtMillis(-1L) .receivedResponseAtMillis(System.currentTimeMillis()) .build().also { listener.satisfactionFailure(call, it) } } // 如果不需要 *** 请求，缓存可用，则返回缓存 if (networkRequest == null) { return cacheResponse!!.newBuilder() .cacheResponse(stripBody(cacheResponse)) .build().also { listener.cacheHit(call, it) } } ... var networkResponse: Response? = null try { // 若无缓存可用，则执行下一个拦截器，获取响应 networkResponse = chain.proceed(networkRequest) } finally { if (networkResponse == null && cacheCandidate != null) { cacheCandidate.body?.closeQuietly() } } // 如果我们还有缓存响应，且 *** 响应code为304，则更新缓存响应，并返回 if (cacheResponse != null) { if (networkResponse?.code == HTTP_NOT_MODIFIED) { // 合并响应头、更新为 *** 请求时间和 *** 响应时间等 val response = cacheResponse.newBuilder() .headers(combine(cacheResponse.headers, networkResponse.headers)) .sentRequestAtMillis(networkResponse.sentRequestAtMillis) .receivedResponseAtMillis(networkResponse.receivedResponseAtMillis) .cacheResponse(stripBody(cacheResponse)) .networkResponse(stripBody(networkResponse)) .build() networkResponse.body!!.close() cache!!.trackConditionalCacheHit() // 更新缓存 cache.update(cacheResponse, response) return response.also { listener.cacheHit(call, it) } } else { cacheResponse.body?.closeQuietly() } } // 包装 *** 响应 val response = networkResponse!!.newBuilder() .cacheResponse(stripBody(cacheResponse)) .networkResponse(stripBody(networkResponse)) .build() // 若用户配置了缓存 if (cache != null) { // 判断是否满足缓存条件 if (response.promisesBody() && CacheStrategy.isCacheable(response, networkRequest)) { // 将 *** 响应写入缓存，并返回 val cacheRequest = cache.put(response) return cacheWritingResponse(cacheRequest, response).also { if (cacheResponse != null) { listener.cacheMiss(call) } } } // 根据请求 *** 判断是否为无效请求，是则从缓存移除相对应响应 if (HttpMethod.invalidatesCache(networkRequest.method)) { try { cache.remove(networkRequest) } catch (_: IOException) { // cache无法被写 } } } return response } }
4. ConnectInterceptor
ConnectInterceptor 主要是给 *** 请求提供一个连接，并交给下一个拦截器处理，这里还没有发送请求到服务器获取响应。在获取连接对象的时候，使用了连接池ConnectionPool来复用连接。
object ConnectInterceptor : Interceptor { @Throws(IOException::class) override fun intercept(chain: Interceptor.Chain): Response { val realChain = chain as RealInterceptorChain // 查找新连接或池里的连接以承载即将到来的请求和响应 val exchange = realChain.call.initExchange(chain) val connectedChain = realChain.copy(exchange = exchange) return connectedChain.proceed(realChain.request) } }
ConnectInterceptor 看似代码很少，其实代码都在深处，看下initExchange ***
internal fun initExchange(chain: RealInterceptorChain): Exchange { ... // codec（ExchangeCodec） 是一个连接所用的编码解码器，用于编码HTTP请求和解码HTTP响应 val codec = exchangeFinder.find(client, chain) // result（Exchange）是封装这个编码解码器的一个工具类，用于管理ExchangeCodec，处理实际的 I/O val result = Exchange(this, eventListener, exchangeFinder, codec) ... return result }
ExchangeCodec持有连接，可通过其编码请求到服务端和获取服务端的响应并解码，我们依 *** 进入到最深处，看看是连接是如何获取的（代码已做简化处理）。
private fun findConnection(): RealConnection { // 1、复用当前连接 val callConnection = call.connection if (callConnection != null) { //检查这个连接是否可用和可复用 if (callConnection.noNewExchanges || !sameHostAndPort(callConnection.route().address.url)) { toClose = call.releaseConnectionNoEvents() } return callConnection } //2、从连接池中获取可用连接 if (connectionPool.callAcquirePooledConnection(address, call, null, false)) { val result = call.connection!! eventListener.connectionAcquired(call, result) return result } //3、从连接池中获取可用连接，通过一组路由routes（涉及知识点Http2多路复用） if (connectionPool.callAcquirePooledConnection(address, call, routes, false)) { val result = call.connection!! return result } route = localRouteSelection.next() // 4、创建新连接，进行tcp连接 val newConnection = RealConnection(connectionPool, route) newConnection.connect // 5、再获取一次连接，在新建连接过程中可能有其他竞争连接被创建了，如可用防止浪费 if (connectionPool.callAcquirePooledConnection(address, call, routes, true)) { val result = call.connection!! // 关闭刚刚创建的新连接 newConnection.socket().closeQuietly() return result } //6、还是要使用创建的新连接，放入连接池，并返回 connectionPool.put(newConnection) return newConnection }
5. CallServerInterceptor
CallServerInterceptor 是真正向服务器发起请求并获取响应的，它是拦责任链的最后一个拦截器，拿到响应后返回给上一个拦截器。代码已做简化（省略了很多条件判断和处理）。
class CallServerInterceptor(private val forWebSocket: Boolean) : Interceptor { @Throws(IOException::class) override fun intercept(chain: Interceptor.Chain): Response { val realChain = chain as RealInterceptorChain // ConnectInterceptor获取到的，持有编码解码器 val exchange = realChain.exchange!! val request = realChain.request val requestBody = request.body val sentRequestMillis = System.currentTimeMillis() var responseBuilder: Response.Builder? = null try { // 写入请求头 exchange.writeRequestHeaders(request) if (HttpMethod.permitsRequestBody(request.method) && requestBody != null) { if (...) { // 写入请求体 val bufferedRequestBody = exchange.createRequestBody(request, false).buffer() requestBody.writeTo(bufferedRequestBody) bufferedRequestBody.close() } else { ... } } else { // 无请求体 exchange.noRequestBody() } } catch (e: IOException) {...} try { if (responseBuilder == null) { // 读取响应头 responseBuilder = exchange.readResponseHeaders(expectContinue = false)!! if (invokeStartEvent) { exchange.responseHeadersStart() invokeStartEvent = false } } // 构建响应 var response = responseBuilder .request(request) .handshake(exchange.connection.handshake()) .sentRequestAtMillis(sentRequestMillis) .receivedResponseAtMillis(System.currentTimeMillis()) .build() var code = response.code // 读取响应体 response = if (forWebSocket && code == 101) { response.newBuilder() .body(EMPTY_RESPONSE) .build() } else { response.newBuilder() .body(exchange.openResponseBody(response)) .build() } ... return response } catch (e: IOException) { ... } } }
总结
以上就是对OkHttp的源码解析，可以看出它是一个结构清晰的优质源码库，各个模块通过设计模式解耦。
总结下流程：首先通过OkHttpClient对象调用newCall *** 得到RealCall实例，再通过调用RealCall的execute *** 或enqueue *** ，这两个 *** 最终都会调用到getResponseWithInterceptorChain *** ，运用责任链模式，开始一层层传入各个拦截器，每个拦截器都有着自己的职责，最终在CallServerInterceptor发出请求并获取响应，然后层层返回响应。
