Android OkHttp 源码浅析一

演进之路:原生Android框架不好用 ---- HttpUrlConnect   和 Apache HTTPClient  

第一版  底层使用HTTPURLConnect  

第二版 Square构建 从Android4.4开始

基本使用:

 val okhttp = OkHttpClient()val request = Request.Builder().url("http://www.baidu.com").build()okhttp.newCall(request).enqueue(object : okhttp3.Callback{override fun onFailure(call: okhttp3.Call, e: IOException) {Log.e("--->",e.message!!)}override fun onResponse(call: okhttp3.Call, response: okhttp3.Response) {Log.e("--->",response.body!!.string())}})//enqueue Call接口定义的抽象方法
//newCall RealCall创建出 有三个参数 1 okhttpclient  2.quest originnalRequest 3. forWebSocket //Booleaan
//WebSocket 是通过http创建连接
//

 override fun enqueue(responseCallback: Callback) {check(executed.compareAndSet(false, true)) { "Already Executed" }callStart()
//dispatcher 用来做线程调度 管理 Executor 线程池
//MaxRequests  = 64 最大64个线程  
//MaxRequestPerHost = 5 最多同一主机线程  5client.dispatcher.enqueue(AsyncCall(responseCallback))}override fun isExecuted(): Boolean = executed.get()private fun callStart() {
//跟踪错误分析和记录this.callStackTrace = Platform.get().getStackTraceForCloseable("response.body().close()")
//监听一系列事件 连接建立 断开连接 报文 head body 发送等eventListener.callStart(this)}

@get:Synchronized var maxRequests = 64 最多连接数量 64

@get:Synchronized var maxRequestsPerHost = 5 同一主机 5

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) {val existingCall = findExistingCallWithHost(call.host)if (existingCall != null) call.reuseCallsPerHostFrom(existingCall)}}promoteAndExecute()}

同步方法,Deque 双向队列

AsyncCall 共享变量 记录连接数等

if (!call.call.forWebSocket) {//记录数量val existingCall = findExistingCallWithHost(call.host)if (existingCall != null) call.reuseCallsPerHostFrom(existingCall)
}

private fun promoteAndExecute(): Boolean {this.assertThreadDoesntHoldLock()val executableCalls = mutableListOf<AsyncCall>()val isRunning: Booleansynchronized(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 // Host max capacity.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(executorService)}return isRunning}

promoteAndExecute 准备执行函数

readyAsyncCalls 没有执行过的请求 不会超出限制 64 or 5

executableCalls 添加到calls 然后取出遍历 执行 executeOn

val i = readyAsyncCalls.iterator() 遍历被执行的call

if (runningAsyncCalls.size >= this.maxRequests) break // Max capacity.
if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue // Host max capacity. 超出连接限制的数量 终止请求

i.remove() //移除当前请求
asyncCall.callsPerHost.incrementAndGet()
executableCalls.add(asyncCall) 
runningAsyncCalls.add(asyncCall)

runningAsyncCalls 正在执行的Call

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

遍历准备执行的call  然后调用executeOn
      asyncCall.executeOn(executorService)
    }

Okhttp enqueue ----> newCall ---> RealCall enqueue ------>  执行 dispatcher . enqueue ---->  添加到readyAsyncCalls ----> promoteAndExecute ---executeOn 执行

  fun executeOn(executorService: ExecutorService) {client.dispatcher.assertThreadDoesntHoldLock()var success = falsetry {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 dispatcher 内部类管理的对象 线程调度 参数是runnable

        executorService.execute(this) 线程切换到后台线程 Async 实现了Runnable

线程调度

 override fun run() {threadName("OkHttp ${redactedUrl()}") {var signalledCallback = falsetimeout.enter()try {val response = getResponseWithInterceptorChain()signalledCallback = trueresponseCallback.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)}}}

    val response = getResponseWithInterceptorChain() 获取相应 

getResponseWithInterceptorChain 请求数据的方法封装

responseCallback 代码里声明传入的Callback