OKHttp介绍
由square公司贡献的一个处理网络请求的开源项目，是目前Android使用最广泛的网络框架。从Android4.4开始HttpURLConnection的底层实现采用的是OkHttp。
- 支持HTTP/2并允许对同一主机的所有请求共享一个套接字;
- 如果非HTTP/2，则通过连接池,减少了请求延迟:
- 默认请求Gzip压缩数据:
- 响应缓存，避免了重复请求的网络;
一个简单的OKHttp的使用示例：
var okHttpClient = OkHttpClient.Builder().connectionPool(ConnectionPool())
.eventListener() // 配置各种监听器。
.build()
var request = Request.Builder().url("https://www.baidu.com")
.cacheControl(CacheControl.FORCE_CACHE)
.build()
var call = okHttpClient.newCall(request)
val result = call.execute()
调用流程
OkHttp请求过程中最少只需要接触OkHttpClient、Request、Call、Response，但是框架内部进行大量的逻辑处理，所有的逻辑大部分集中在拦截器中，但是在进入拦截器之前还需要依靠分发器来调配请求任务。
- 分发器:内部维护队列与线程池，完成请求调配
- 拦截器:完成整个请求过程
Dispatcher负责任务的分发，Intercepter真正的完成了请求的过程。
源码分析
call包装
OKHttp的请求载体是Call对象，我们先看看Call的生成okHttpClient.newCall(request)：
override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)
我们发出的call会被包装成RealCall。
异步请求
异步请求需要将每一个请求入队。
override fun enqueue(responseCallback: Callback) {
check(executed.compareAndSet(false, true)) { "Already Executed" }
callStart()
// 包装成AsyncCall然后入队。这里的Dispatcher我们可以自定义，不过一般我们也不需要这么做。
client.dispatcher.enqueue(AsyncCall(responseCallback))
}
private fun callStart() {
this.callStackTrace = Platform.get().getStackTraceForCloseable("response.body().close()")
eventListener.callStart(this)
}
可见OKHttp为我们提供了很多监听器。
AsyncCall就是是一个runnable
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!
}
}
}
新入对的请求会被放入readyAsyncCalls中。Dispatcher总共有3个队列。
- 准备执行的异步请求 readyAsyncCalls = ArrayDeque()
- 正在执行的异步请求 runningAsyncCalls = ArrayDeque()
- 正在执行的同步请求 runningSyncCalls = ArrayDeque()
internal fun enqueue(call: AsyncCall) {
synchronized(this) {
readyAsyncCalls.add(call)
// Mutate the AsyncCall so that it shares the AtomicInteger of an existing running call to
// the same host.
if (!call.call.forWebSocket) {
// 查找是否有一个已经完全一样的host请求，防止浪费性能。
val existingCall = findExistingCallWithHost(call.host)
//将这两个一样的host 的callsPerhost变成完全一样的，让相同的host的async中的AtomicInteger callsPerHost是同一个对象，有多少个相同host的正在执行的请求数量。
if (existingCall != null) call.reuseCallsPerHostFrom(existingCall)
}
}
// 分发执行。
promoteAndExecute()
}
在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()
// 正在执行异步请求的任务数 不能大于 64个
if (runningAsyncCalls.size >= this.maxRequests) break // Max capacity.
// 同一个host的请求数 不能大于5，大于5个就去拿下一个任务，该任务会被搁置。
if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue // Host max capacity.
// 任务要执行了，就需要移除掉。
i.remove()
asyncCall.callsPerHost.incrementAndGet()
// 需要开始执行的任务集合，记录到一个临时的集合中。
executableCalls.add(asyncCall)
// 加入running队列。
runningAsyncCalls.add(asyncCall)
}
isRunning = runningCallsCount() > 0
}
// 遍历执行任务集合。
for (i in 0 until executableCalls.size) {
val asyncCall = executableCalls[i]
asyncCall.executeOn(executorService)
}
return isRunning
}
这里就是开始执行任务了。
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 execute(): Response {
// 每一个call都只能执行一次。
check(executed.compareAndSet(false, true)) { "Already Executed" }
timeout.enter()
// 这里边其实就是配置了一个listener.我们在client创建的时候，可以自己配置一个eventListener。
callStart()
try {
client.dispatcher.executed(this)
return getResponseWithInterceptorChain()
} finally {
client.dispatcher.finished(this)
}
}
根据上述的逻辑，我们总结出异步请求工作流程：
okhttp3.internal.connection.RealCall.AsyncCall#run 任务的最终执行是在这个run方法中实现的。
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)
}
}
}
Dispatcher使用的线程池使用的是 SynchronousQueue，SynchronousQueue是一个没有数据缓冲的BlockingQueue，因此提交任务之后，一定会走创建新线程的流程去执行任务，这是为了让我们提交的任务得到及时的执行，满足我们对任务的高效执行的需求：
@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!!
}
同步请求
直接放到同步请求队列，开始执行。
@Synchronized internal fun executed(call: RealCall) {
runningSyncCalls.add(call)
}
最终会走到这里
internal fun finished(call: AsyncCall) {
call.callsPerHost.decrementAndGet()
finished(runningAsyncCalls, call)
}
需要注意的是，同步请求最终也会触发promoteAndExecute()。
拦截器链
无论是同步请求还是异步请求，都会走getResponseWithInterceptorChain()来完成数据获取的。采用的是责任链模式，这是一种对象行为型模式，为请求创建了一个接收者对象的链，在处理请求的时候执行过滤(各司其职)。责任链上的处理者负责处理请求，客户只需要将请求发送到责任链即可，无须关心请求的处理细节和请求的传递，所以职责链将请求的发送者和请求的处理者解耦了。执行流程：
除了这几个拦截器，我们还可以自定义拦截器来处理数据，例如添加用户名之类的，有两种方式，但放置的位置不同，但一定要注意，自定义的拦截器，一定要调用proceed方法。 否则链条会中断。
// 这个拦截器是放在了所有拦截器之前。
.addInterceptor(object : Interceptor {
override fun intercept(chain: Interceptor.Chain): Response {
chain.request()
}
})
// 这个拦截器是放在了CallServerInterceptor之前，ConnectionInterceptor之后。 只在app请求有用，在websocket无效,这个拦截器可以打印出来OKHttp真正发出的请求。可以帮助我们打印请求的数据和响应的数据
.addNetworkInterceptor {
}
源码在这里：
拦截器链的集合：
public class Chain {
// 将所有的拦截器放到这个列表中。
private List<Interceptor> interceptors;
private int index;
public String request;
public Chain(List<Interceptor> interceptors, int index, String request) {
this.interceptors = interceptors;
this.index = index;
this.request = request;
}
public Chain(List<Interceptor> interceptors, int index) {
this.interceptors = interceptors;
this.index = index;
}
// 接受请求，并且交给拦截器去处理。
public String processd(String request) {
if (index >= interceptors.size()) {
throw new AssertionError();
}
// 创建一个新的链，用来将责任转给下一个拦截器。
Chain chain = new Chain(interceptors, index + 1, request);
Interceptor interceptor = interceptors.get(index);
// 将数据返回给上一个拦截器。整个过程是一个U型传递。
return interceptor.intercept(chain);
}
}
各个拦截器的作用：
- 重试重定向拦截器在交出(交给下一个拦截器)之前，负责判断用户是否取消了请求;在获得了结果之后，会根据响应码判断是否需要重定向，如果满足条件那么就会重启执行所有拦截器。
- 桥接拦截器在交出之前，负责将HTTP协议必备的请求头加入其中(如:Host)并添加一些默认的行为(如:GZIP压缩);在获得了结果后，调用保存cookie接口并解析GZIP数据。
- 缓存拦截器顾名思义，交出之前读取并判断是否使用缓存;获得结果后判断是否缓存。
- 连接拦截器在交出之前，负责找到或者新建一个连接，并获得对应的socket流;在获得结果后不进行额外的处理。
- 请求服务器拦截器进行真正的与服务器的通信，向服务器发送数据，解析读取的响应数据
RetryAndFollowUpInterceptor
/**
* Report and attempt to recover from a failure to communicate with a server. Returns true if
* `e` is recoverable, or false if the failure is permanent. Requests with a body can only
* be recovered if the body is buffered or if the failure occurred before the request has been
* sent.
*/
private fun recover(
e: IOException,
call: RealCall,
userRequest: Request,
requestSendStarted: Boolean
): Boolean {
// The application layer has forbidden retries.
//okhttpclient配置不重试
if (!client.retryOnConnectionFailure) return false
// We can't send the request body again.
// 不重试：
// 1、如果是IO异常（非http2中断异常）表示请求可能发出
// 2、如果请求体只能被使用一次（默认为false）
if (requestSendStarted && requestIsOneShot(e, userRequest)) return false
// This exception is fatal.
// 异常不重试：协议异常、IO中断异常（除Socket读写超时之外），ssl认证异常
if (!isRecoverable(e, requestSendStarted)) return false
// No more routes to attempt.
//是否有更多的路线
if (!call.retryAfterFailure()) return false
// For failure recovery, use the same route selector with a new connection.
return true
}
private fun requestIsOneShot(e: IOException, userRequest: Request): Boolean {
val requestBody = userRequest.body
// 第一个条件默认为false
// 第二个条件，比如上传文件，但是本地上传的文件不存在
return (requestBody != null && requestBody.isOneShot()) ||
e is FileNotFoundException
}
private fun isRecoverable(e: IOException, requestSendStarted: Boolean): Boolean {
// If there was a protocol problem, don't recover.
// 协议异常 不重试
if (e is ProtocolException) {
return false
}
// If there was an interruption don't recover, but if there was a timeout connecting to a route
// we should try the next route (if there is one).
// 如果发生中断 不重试，但如果连接到路由时超时可以重试
if (e is InterruptedIOException) {
return e is SocketTimeoutException && !requestSendStarted
}
// Look for known client-side or negotiation errors that are unlikely to be fixed by trying
// again with a different route.
// 证书有问题 不重试
if (e is SSLHandshakeException) {
// If the problem was a CertificateException from the X509TrustManager,
// do not retry.
if (e.cause is CertificateException) {
return false
}
}
// 证书验证失败
if (e is SSLPeerUnverifiedException) {
// e.g. a certificate pinning error.
return false
}
// An example of one we might want to retry with a different route is a problem connecting to a
// proxy and would manifest as a standard IOException. Unless it is one we know we should not
// retry, we return true and try a new route.
return true
}
408:服务端给到retryafter:0，或者没给,；503服务端必须要给到 retryafter:0
BridgeInterceptor
CacheInterceptor
Http的缓存我们可以按照行为将他们分为:强缓存和协商缓存。命中强缓存时，浏览器并不会将请求发送给服务器。强缓存是利用http的返回头中的Expires(一般用这个)或者Cache-Control两个字段来控制的，用来表示资源的缓存时间;若未命中强缓存，则浏览器会将请求发送至服务器。服务器根据http头信息中的Last-Modify/lf-Modify-Since或Etag/lf-None-Match来判断是否命中协商缓存。如果命中，则http返回码为304，客户端从缓存中加载资源。
ConnectInterceptor
一个对象池，用来做链接复用。内部实现使用的一 个RealConnectionPool来实现对象的保存。所有的链接都放到这里边了。这里边很重要的一个操作就是：
fun put(connection: RealConnection) {
connection.assertThreadHoldsLock()
connections.add(connection) // 保存链接对象。
cleanupQueue.schedule(cleanupTask)
}
同时，内部还会启动一个周期性的任务，用来清理无效的链接。
```js
private val cleanupTask = object : Task("$okHttpName ConnectionPool") {
override fun runOnce() = cleanup(System.nanoTime())
}
RealConnection代表是连接socket链路，RealConnection对象意味着我们已经跟服务端有了一条通信链路了。很多朋友这时候会想到，有通信链路了，是不是与意味着在这个类实现的三次握手，你们猜对了，的确是在这个类里面实现的三次握手。
RealConnectionPool 每一个链接，通过构造方法可以确定， 默认的最大空闲数为5，最大允许空闲时间为5分钟。也就是如果这个链接超过5分钟没有下一个请求使用，就会被弃用。弃用的方式采用的LRUCache.
/**
* maxIdleConnections 连接池最大允许的空闲连接数
* keepAliveDuration 连接最大允许的空闲时间5分钟
*/
constructor() : this(5, 5, TimeUnit.MINUTES)
连接能够复用，这下边的所有变量都需要相同，同时host也要一致。
object ConnectInterceptor : Interceptor {
@Throws(IOException::class)
override fun intercept(chain: Interceptor.Chain): Response {
val realChain = chain as RealInterceptorChain
// 获取连接 Exchange：数据交换（封装了连接）
val exchange = realChain.call.initExchange(chain)
val connectedChain = realChain.copy(exchange = exchange)
return connectedChain.proceed(realChain.request)
}
}
internal fun initExchange(chain: RealInterceptorChain): Exchange {
synchronized(this) {
check(expectMoreExchanges) { "released" }
check(!responseBodyOpen)
check(!requestBodyOpen)
}
val exchangeFinder = this.exchangeFinder!!
// ExchangeCodec: 编解码器 find：查找连接Realconnection
val codec = exchangeFinder.find(client, chain)
// Exchange：数据交换器 包含了exchangecodec与Realconnection
val result = Exchange(this, eventListener, exchangeFinder, codec)
this.interceptorScopedExchange = result
this.exchange = result
synchronized(this) {
this.requestBodyOpen = true
this.responseBodyOpen = true
}
if (canceled) throw IOException("Canceled")
return result
}
Socket与Http的区别：socket能够完成TCP/IP所有协议，HTTP协议只能完成HTTP协议。参考：blog.csdn.net/mccand1234/…
连接池工作流程梳理：
新建的链接如果是Keep-alive链接，就放入连接池。放入后还要同时启动清理任务，清理限制链接的事件取决于目前连接池里边最长闲置链接，如果连接池里边的闲置链接最长是1分钟，那么下次清理的事件就是4分钟之后了。
连接池清理流程：
fun `cleanup`(now: Long): Long {
var inUseConnectionCount = 0
var idleConnectionCount = 0
var longestIdleConnection: RealConnection? = null
var longestIdleDurationNs = Long.MIN_VALUE
// Find either a connection to evict, or the time that the next eviction is due.
for (connection in connections) {
synchronized(connection) {
// If the connection is in use, keep searching.
// 记录正在使用与已经闲置的连接数
if (pruneAndGetAllocationCount(connection, now) > 0) {
inUseConnectionCount++
} else {
idleConnectionCount++
// 记录最长闲置时间的连接longestIdleConnection
// If the connection is ready to be evicted, we're done.
val idleDurationNs = now - connection.idleAtNs
if (idleDurationNs > longestIdleDurationNs) {
longestIdleDurationNs = idleDurationNs
longestIdleConnection = connection
} else {
Unit
}
}
}
}
when {
//最长闲置时间的连接超过了允许闲置时间 或者 闲置数量超过允许数量，清理此连接
longestIdleDurationNs >= this.keepAliveDurationNs
|| idleConnectionCount > this.maxIdleConnections -> {
// We've chosen a connection to evict. Confirm it's still okay to be evict, then close it.
val connection = longestIdleConnection!!
synchronized(connection) {
if (connection.calls.isNotEmpty()) return 0L // No longer idle.
if (connection.idleAtNs + longestIdleDurationNs != now) return 0L // No longer oldest.
connection.noNewExchanges = true
connections.remove(longestIdleConnection)
}
connection.socket().closeQuietly()
if (connections.isEmpty()) cleanupQueue.cancelAll()
// Clean up again immediately.
return 0L
}
// 存在闲置连接，下次执行清理任务在 允许闲置时间-已经闲置时候后
idleConnectionCount > 0 -> {
// A connection will be ready to evict soon.
return keepAliveDurationNs - longestIdleDurationNs
}
// 存在使用中的连接，下次清理在 允许闲置时间后
inUseConnectionCount > 0 -> {
// All connections are in use. It'll be at least the keep alive duration 'til we run
// again.
return keepAliveDurationNs // 5分钟之后开始清理。
}
else -> {
// No connections, idle or in use. 立刻执行下一次的清理流程。
return -1
}
}
}
CallServerInterceptor
class CallServerInterceptor(private val forWebSocket: Boolean) : Interceptor {
@Throws(IOException::class)
override fun intercept(chain: Interceptor.Chain): Response {
val realChain = chain as RealInterceptorChain
val exchange = realChain.exchange!!
val request = realChain.request
val requestBody = request.body
val sentRequestMillis = System.currentTimeMillis()
// 如果是获取数据的接口，走这里就够了。
exchange.writeRequestHeaders(request)
// 如果携带者数据上送，需要做以下处理。
if (HttpMethod.permitsRequestBody(request.method) && requestBody != null) {
// If there's a "Expect: 100-continue" header on the request, wait for a "HTTP/1.1 100
// Continue" response before transmitting the request body. If we don't get that, return
// what we did get (such as a 4xx response) without ever transmitting the request body.
// 如果要向服务器发送比较大的数据，先要问服务器能否接受这么大的数据。
if ("100-continue".equals(request.header("Expect"), ignoreCase = true)) {
exchange.flushRequest()
responseBuilder = exchange.readResponseHeaders(expectContinue = true)
exchange.responseHeadersStart()
invokeStartEvent = false
}
// 最终是通过socket向服务器发送请求头请求行。
fun writeRequest(headers: Headers, requestLine: String) {
check(state == STATE_IDLE) { "state: $state" }
sink.writeUtf8(requestLine).writeUtf8("\r\n")
for (i in 0 until headers.size) {
sink.writeUtf8(headers.name(i))
.writeUtf8(": ")
.writeUtf8(headers.value(i))
.writeUtf8("\r\n")
}
sink.writeUtf8("\r\n")
state = STATE_OPEN_REQUEST_BODY
}
拦截器总结
1.重试与重定向拦截器：重定向的次数是20次，重试机制比较苛刻，请求失败是否需要重试。重定向回来会把上一次的请求结果记录到这一次重定向请求的结果中。根据响应码确定是否需要重定向。
2.桥接拦截器：补全请求头，根据返回结果决定是否进行GZip解压，是否保存Cookie，Gzip，请求体的解析方式等等；
3.缓存拦截器：Http的缓存分为强缓存与协商缓存。命中强缓存时，直接返回缓存，而不再发送请求了，浏览器并不会将请求发送给服务器。强缓存是利用http的返回头中的Expires或者Cache-Control两个字段来控制的，用表示资源的缓存时间；若未命中强缓存，则浏览器会将请求发送至服务器。服务器根据http头信息中的Last-Modify/If-Modify- Since或Etag/If-None-Match来判断是否命中协商缓存。如命中，则http返回码为304，客户端从缓存中加载资源。要缓存（FORCE_CACHE）没有缓存，会返回504。no-store，资源不能被缓存。
4.链接拦截器：ExchangeCoder，代理分为SOcket和HTTP，Socket可以完成TCP/IP等的的协议，包含HTTP代理的一些事情，Http只能完成Http层的协议。HTTPs通信的时候，建立的是HTTP的隧道代理（无脑转发），而不是普通代理（可以处理请求）。Socket是链接的真实服务地址，HTTP链接的是代理地址，隧道请求需要先发送一个connet请求，这个请求需要包含真实的服务器地址。链接确定之后，怎么确定发送选择HTTP1.1还是HTTP2.0？ALPN协议。TLS的一个扩展协议。连接池可以自己配置。maxIdleconnections, 空闲连接数尝试保存在最大为5，每一个链接最长空闲时间5分钟，之后就被清理掉。连接池中存在大量的空闲的连接对象。把闲置时间最长的链接一个一个的清理掉，利用的是LRU思想。Connetion-Kepalive,放入连接池。同时启动清理任务。这是一个周期性的任务。
5.请求服务拦截器：拿到已经处理好的请求和链接，利用Socket完成向服务器发送消息。并解析相应数据。
整体框架总结
一些问题
1、oKHttp请求的请求流程是怎样的?
okhttp的请求流程：Request-->OKHttpclient-->RealCall-->同步或异步。ready:等待执行的异步任务队列； running:正在执行的异步任务队列。当前正在执行的异步任务数不得大于64，从ready队列中迭代出的等待执行的任务 host 不得大于5个
2、oKHttp分发器是怎样工作的?
同步->记录同步任务:RealCall
异步->首先将任务加入ready队列等待执行->是否需要将ready中的任务放入running 执行同时请求的异步任务数不得大于64个;从ready中取出来的异步任务，与其相同的HOST，不得大于5个若已经存在5个相同H0ST的任务在执行，则继续从ready中检查下一个等待任务。
3、oKHttp拦截器是如何工作的?
责任链设计模式将请求者与执行者解耦， 让请求者只需要将请求发给责任链即可，无需关系请求过程与细节。包括重试重定向、桥接、缓存、连接、
4、应用拦截器和网络拦截器有什么区别? 应用拦截器添加在最开头，网络拦截器添加到CallServerInterceptor拦截器之前。自定义的网络拦截器不一定执行。应用拦截器，重试重定向，桥接，缓存。连接，网络拦截器（不一定执行），请求服务。
5、OKHttp如何复用TCP链接？
设置的参数要基本都一样才行。
连接池也会去清理垃圾链接，超过5分钟没有用过链接，超过5个链接后，从最久限制的链接开始清理（LRU算法）。
