开源库源码分析:Okhttp源码分析(一)

开源库源码分析:OkHttp源码分析

在这里插入图片描述

导言

接下来就要开始分析一些常用开源库的源码了,作为最常用的网络请求库,OkHttp以其强大的功能深受Android开发者的喜爱(比如说我),还有对该库进行二次封装而成的热门库,比如说Retrofit。本文我们将从源码入手看看OkHttp是如何运作的。注意本文解析的是OkHttp3库,该库是用Kotlin写的,需要大家有一些Kotlin基础。

OkHttp的最佳使用

这个问题是在OkHttp3的OkHttpClient中的注释中发现的:
在这里插入图片描述
这段注释中提到了OkHttp最好是用单例的OkHttpClient来实现请求,我们可以对该单例进行复用。这是因为每一个OkHttpClient都会持有一个连接池和线程池,都称之为池了,那么其作用肯定就是为了复用。通过复用这些连接和线程我们可以显著地降低延迟和节约内存使用。

实际上这段注释中已经提到了OkHttp中的一些复用机制了,连接复用和线程复用。

调度类Dispatcher

调度类的线程池

OkHttp中核心类之一就是这个调度类Dispatcher,所有的网络请求类Call都需要通过这个调度类来执行任务。这个调度类是OkHttpClient类的成员变量,我们是在调度类中执行请求的,那我们就顺便再来介绍一下客户端类中的成员变量:

//调度类
@get:JvmName("dispatcher") val dispatcher: Dispatcher = builder.dispatcher
//连接池
@get:JvmName("connectionPool") val connectionPool: ConnectionPool = builder.connectionPool
//应用拦截器
@get:JvmName("interceptors") val interceptors: List<Interceptor> = builder.interceptors.toImmutableList()
//网络拦截器
@get:JvmName("networkInterceptors") val networkInterceptors: List<Interceptor> =builder.networkInterceptors.toImmutableList()
//事件监听工厂
@get:JvmName("eventListenerFactory") val eventListenerFactory: EventListener.Factory =builder.eventListenerFactory

这几个应当是最核心的几个成员变量,我们将在后面再接触它们。回归正题,我们继续看调度类,既然要执行Call,那么这个调度器类就一定会有线程池,这个线程池有两个手段可以设置:

  private var executorServiceOrNull: ExecutorService? = null//线程池@get:Synchronized //在调用get时设置线程池为一个默认线程池,类似于一个CachedPool@get:JvmName("executorService") val executorService: ExecutorServiceget() {if (executorServiceOrNull == null) {executorServiceOrNull = ThreadPoolExecutor(0, Int.MAX_VALUE, 60, TimeUnit.SECONDS,SynchronousQueue(), threadFactory("$okHttpName Dispatcher", false))}return executorServiceOrNull!!}constructor(executorService: ExecutorService) : this() { //通过构造方法设置线程池this.executorServiceOrNull = executorService}

可以看到,方法一是调用构造函数来设置线程池,第二个方法就是在调用get方法时会自动设置一个效果类似于CachedPool的线程池。

调度类执行请求

我们使用OkHttp时一般是两种方式,同步请求execute异步请求enqueue,我们以异步请求为例来分析调度类是如何执行网络请求的。一般我们发送异步请求是这样的形式:

    public static void main(String[] args) {OkHttpClient client = new OkHttpClient();Request r = new Request.Builder().build();Call call = client.newCall(r);call.enqueue(new Callback() {@Overridepublic void onFailure(@NonNull Call call, @NonNull IOException e) {}@Overridepublic void onResponse(@NonNull Call call, @NonNull Response response) throws IOException {}});}

首先通过RequestBuilder类来构建了一个请求类,然后将该请求传入newCall方法中新创建出了一个Call

override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)

可以看到返回的是RealCall类,该类实现了Call接口。接下来就从Call的enqueue方法入手,这里实际上是RealCall类中:

  override fun enqueue(responseCallback: Callback) {check(executed.compareAndSet(false, true)) { "Already Executed" }callStart() //开启事件监听client.dispatcher.enqueue(AsyncCall(responseCallback)) //通过调度类入队}

这其中callStart方法会开启事件监听,我们先不管这个事件监听,看最后一句通过调度器来入队,这个AsyncCall()类是异步请求类,实现了Runnable接口,这里实际上就是将我们传入的实现了Callback的回调给包装成了AsyncCall。我们继续看Dispatcher中是如何入队的:

  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()}

首先为了确保线程安全,调度器先将其上锁然后将之前包装成的AsyncCall加入到了readyAsyncCall就绪队列中。中间这一段判断是用于处理与主机(host)相关的异步调用复用。如果 call 不是用于 WebSocket 调用,它会尝试查找是否已经存在一个与相同主机(host)相关的异步调用任务。如果找到了已经存在的异步调用,它会尝试共享这两个异步调用任务之间的状态。这是为了复用已经建立的连接,以提高性能和减少资源消耗。

最后会调用promoteAndExecute方法执行,这个方法是执行请求的核心之一:

 private fun promoteAndExecute(): Boolean {this.assertThreadDoesntHoldLock()val executableCalls = mutableListOf<AsyncCall>()//可执行的Callsval 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}

这段代码的核心逻辑就是while循环之中,这很显然是一个通过迭代器开始遍历就绪队列readyAsyncCall的过程,它会将就绪队列中的call移入到可执行队列executableCalls和正在执行异步队列runningAsyncCalls中。其中还有两句判断条件:

        if (runningAsyncCalls.size >= this.maxRequests) break // Max capacity.if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue // Host max capacity.

一句是判断正在运行队列没有超过最大Size,一句是判断每一句的最大执行数不超过最大值,这两个值在Dispatcher中都有定义。最后会跳入到for循环中并调用AsyncCall的executeOn方法,并将线程池传入,也就是接下来就是在线程池中执行该异步请求了。我们最后来看这个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!}}}

这个方法真是开门见山。在try语句的第一句处就通过线程池执行了该任务,还记得这个AsyncCall是由我们传入的Call回调包装而成吗?接下来就会回调到我们的函数之中去了。众所周知,由于executorService.execute(this)这一句,线程池接下来会执行AsyncCall的run方法了:

    override fun run() {threadName("OkHttp ${redactedUrl()}") {var signalledCallback = falsetimeout.enter()try {val response = getResponseWithInterceptorChain() //1--通过拦截器链来获得响应signalledCallback = trueresponseCallback.onResponse(this@RealCall, response)//2--请求成功,回调到我们传入的onResponse方法中} 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) //3--请求失败,回调到我们传入的onFailure方法之中去}} 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)//结束本次请求}}}

我已经在代码之中加入一些注释了,在注释一处会通过拦截器链来获得响应,这个拦截器机制我们之后再将,这里只要知道它在这里获得了响应即可。在注释二处我们就可以看到这里显然是调用了我们传入的Call回调了,请求成功调用onResponse回调,要是请求失败的话则回调到onFailure方法之中去。最后在finally块中将本次请求结束,实际上这个finished方法还会开启下一次循环:

  internal fun finished(call: AsyncCall) {call.callsPerHost.decrementAndGet()finished(runningAsyncCalls, call)}private fun <T> finished(calls: Deque<T>, call: T) {val idleCallback: Runnable?synchronized(this) {if (!calls.remove(call)) throw AssertionError("Call wasn't in-flight!")idleCallback = this.idleCallback}val isRunning = promoteAndExecute()if (!isRunning && idleCallback != null) {idleCallback.run()}}

由第一个函数会调转到第二个函数,从第二个函数来看,它首先会将本次的call从runningAsyncCalls正在运行队列中移除,然后会继续调用我们之前提到过的promoteAndExecute方法开启下一次循环,如果没有任务了的话并且空闲处理程序不为空的话,还会执行空闲处理程序。

拦截器链

之前提到网络请求的相应是通过拦截器链获得的:

val response = getResponseWithInterceptorChain()

首先,什么是拦截器呢?

在计算机编程领域,拦截器(Interceptor)是一种常见的设计模式,用于在软件系统的不同组件之间添加或修改功能。拦截器的主要目的是拦截请求或操作,允许在请求进入目标组件之前或之后执行自定义逻辑。

简单来说就是有许多个拦截器,一个请求的执行需要经过这一整条拦截器链,在一个请求到达一个拦截器的时候我们就可以判断是否要拦截这个请求并进行处理,简单来说就是类似于旅行途中的设置关口。Android中这种模式也运用,比如说广播View的事件分发和处理,这都是拦截器模式的运用。

好了,言归正传,我们继续回到上面的方法上:

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 = falsetry { //启动责任链val response = chain.proceed(originalRequest)if (isCanceled()) {response.closeQuietly()throw IOException("Canceled")}return response} catch (e: IOException) {calledNoMoreExchanges = truethrow noMoreExchanges(e) as Throwable} finally {if (!calledNoMoreExchanges) {noMoreExchanges(null)}}}

具体代码的含义已经写在注释中了,首先要提醒大家上面的+=号是Kotlin的重载,就是相当于添加集合。其实把上面的方法切分一下的话主要就是做了两件事:创建和整理拦截器集合创建并启动责任链。下面是一张从官方文档中截取的图片

在这里插入图片描述
我们看右下部分就是说明网络响应是通过网络拦截器来层层向上传递的,并且拦截器是HTTP工作真正发生的地方。
下面再来介绍各个拦截器的作用:

  • interceptor:应用拦截器,通过Client设置
  • RetryAndFollowUpInterceptor:重试拦截器,负责网络请求中的重试重定向。比如网络请求过程中出现异常的时候就需要进行重试。
  • BridgeInterceptor:桥接拦截器,用于桥接应用层和网络层的数据。请求时将应用层的数据类型转化为网络层的数据类型,响应时则将网络层的数据类型转化为应用层的数据类型。
  • CacheInterceptor:缓存拦截器,负责读取和更新缓存。可以配置自定义的缓存拦截器。
  • ConnectInterceptor:网络连接拦截器,其内部会获取一个连接。
  • networkInterceptor:网络拦截器,通过Client设置。
  • CallServerInterceptor:请求服务拦截器。它是拦截器中处于末尾的拦截器,用于向服务端发送数据并获取响应

我们回到上面的方法中,在try块中的第一句就是启动责任链chain.proceed(),这个方法将会启动责任链并获取响应:

override fun proceed(request: Request): Response {check(index < interceptors.size)......// 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}

这里面的核心就是上面的这一小段,首先通过copy方法获得下一个拦截器:

internal fun copy(index: Int = this.index,exchange: Exchange? = this.exchange,request: Request = this.request,connectTimeoutMillis: Int = this.connectTimeoutMillis,readTimeoutMillis: Int = this.readTimeoutMillis,writeTimeoutMillis: Int = this.writeTimeoutMillis) = RealInterceptorChain(call, interceptors, index, exchange, request, connectTimeoutMillis,readTimeoutMillis, writeTimeoutMillis)

这个copy本质还是调用了构造函数,这这里它调用更改的唯一参数就是index参数,它将原来的index+1,显然是想要达到遍历的效果,那么这个index参数到底会影响什么呢?我们再将其与一开始的构造相比较:

     //创建责任链val chain = RealInterceptorChain(call = this,interceptors = interceptors,index = 0,exchange = null,request = originalRequest,connectTimeoutMillis = client.connectTimeoutMillis,readTimeoutMillis = client.readTimeoutMillis,writeTimeoutMillis = client.writeTimeoutMillis)

一开始这个index的值是0,通过查找整个类可以发现唯一有意义的就是我们上面提到的proceed方法:

val interceptor = interceptors[index]
val response = interceptor.intercept(next) ?: throw NullPointerException("interceptor $interceptor returned null")

就是用来从拦截器集合中获取拦截器的。之后又会用我们获取到的拦截器调用其intercept方法,并将之前copy出来的对象传进去(也就是新的责任链,只不过index向后指了一格)。这个intercept方法是一个接口方法,实现该接口的类有很多,我们以比较简单的ConnectInterceptor类为例:

  override fun intercept(chain: Interceptor.Chain): Response {val realChain = chain as RealInterceptorChainval exchange = realChain.call.initExchange(chain)val connectedChain = realChain.copy(exchange = exchange)return connectedChain.proceed(realChain.request)}

这里它做的唯一就是初始化了一个exchange参数然后将其设置进了责任链中,最后又会调用新设置好的责任链的proceed方法中。这就又回到了之前的方法中。这样看来,这个责任链的大体行为模式还是很好懂的,首先启动责任链,责任链调用proceed方法启动,proceed方法中会获得下一个拦截器并且调用下一个拦截器中的intercept拦截方法,在这个拦截方法中首先会进行该拦截器的一些拦截逻辑,拦截逻辑完成之后会再次调用proceed方法继续获得下一个拦截器,然后再调用它的拦截器方法,以此类推直到整个拦截器链上的拦截器方法都执行一遍,最后返回出一个Response响应。整个过程差不多如下所示:
在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/109667.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

无涯教程-JavaScript - COLUMNS函数

描述 COLUMNS函数返回数组或引用中的列数。 语法 COLUMNS (array)争论 Argument描述Required/OptionalarrayAn array or array formula, or a reference to a range of cells for which you want the number of Columns.Required Notes COLUMNS(1:1)返回Excel中的列数,即…

20230912在ubuntu18.04下使用pigz来提高tar命令压缩解压缩的速度

20230912在ubuntu18.04下使用pigz来提高tar命令压缩解压缩的速度 2023/9/15 22:19 https://blog.csdn.net/wb4916/article/details/128447298 20221226编译Toybrick的TB-RK3588X开发板的Android12系统2-SDK预处理 4、单线程压缩。 建议使用&#xff1a;pigz多线程压缩&#xf…

极光笔记 | 推送服务数据中心选择:合规性与传输效率的双重考量

随着全球化进程的深入&#xff0c;跨境数据传输与存储问题已经变得愈发重要。推送服务的数据中心节点选择不仅关乎数据访问速度和用户体验&#xff0c;同时也直接牵扯到数据合规性和安全保障。EngageLab Push深知这一点&#xff0c;为了满足更多国际客户和全球用户触达需求&…

sqlserver存储过程报错:当前事务无法提交,而且无法支持写入日志文件的操作。请回滚该事务。

现象&#xff1a; 系统出现异常&#xff0c;手动执行过程提示如上。 问题排查&#xff1a; 1.直接执行的过程事务挂起&#xff08;排除&#xff09; 2.重启数据库实例&#xff08;重启后无效&#xff09; 3.过程中套用过程&#xff0c;套用的过程中使用事务&#xff0c;因为…

sql存储引擎

-- 查询建表语句 --可以查看引擎 show create table account; -- 可以看到默认引擎 InnoDB ENGINEInnoDB -- 查看当前数据库支持得存储引擎 show engines ; # InnoDB 默认 存储引擎 # MyISAM sql早期默认 存储引擎 # MEMORY 存储在内存中 用来做临时表和缓存 存储引擎 …

伺服取料机构控制

伺服运动控制系统相关介绍大家可以查看运动控制专栏&#xff0c;这篇文章主要讨论常用的伺服取料机构的控制问题。常用运动控制文章链接如下&#xff1a; EtherCAT总线运动控制学习笔记&#xff08;RXXW_Dor&#xff09;_ethercat控制字_RXXW_Dor的博客-CSDN博客说到总线控制&…

管理方法论:4. 一对一沟通——了解真实情况

团队人越多&#xff0c;管理者了解真实情况就越困难&#xff0c;为什么&#xff1f; 缺乏安静的、可以专注于沟通的环境。有些话不能跟太多人说&#xff0c;尤其是负面的想法和意见。有些人本来就内向、被动&#xff0c;其他人说话时&#xff0c;自己往往习惯于听&#xff0c;…

ESP32-BOX的组件配置添加核心部分详细介绍

前言 &#xff08;1&#xff09;为了方便开发&#xff0c;ESP32提供了组件库方便用户进行二次开发。 github仓库&#xff1b;gitee仓库 &#xff08;2&#xff09;在学习本章之前最好有CMake或者Makefile的基础&#xff0c;如果没有也不要慌&#xff0c;有的话最好。 &#xff…

闲鱼自动化软件——筛选/发送系统 V22已经测试完毕

更新 因为闲鱼版本更新&#xff0c;以及闲鱼整个程序维护记录&#xff0c;又增加了一些优化和提升的代码&#xff0c;所以又一次在整体上更新了一版闲鱼的此款软件。 主要更新点&#xff1a; 1、添加显示自定义按钮&#xff0c;可以自动显示最新数据&#xff0c;也可以手动翻…

Linux 内存泄漏检测的基本原理

一、mtrace分析内存泄露 mtrace&#xff08;memory trace&#xff09;&#xff0c;是 GNU Glibc 自带的内存问题检测工具&#xff0c;它可以用来协助定位内存泄露问题。 它的实现源码在glibc源码的malloc目录下&#xff0c;其基本设计原理为设计一个函数 void mtrace ()&…

CSS - 鼠标移入整行高亮显示,适用于会员套餐各参数对比页面(display: table,div 转表格形式)

效果图 可根据基础示例和进阶示例&#xff0c;复制进行改造样式。 如下图所示&#xff0c;本文提供 2 个示例。 基础示例 找个 HTML 页面&#xff0c;一键复制运行。 <body><h1 style"text-align: center;">基础示例</h1><section class"…

【Cocos Creator 3.5实现赛车游戏】10.实现汽车节点的运动逻辑

转载知识星球 | 深度连接铁杆粉丝&#xff0c;运营高品质社群&#xff0c;知识变现的工具 项目地址&#xff1a;赛车小游戏-基于Cocos Creator 3.5版本实现: 课程的源码&#xff0c;基于Cocos Creator 3.5版本实现 上一节的学习后&#xff0c;您已经完成了对汽车节点的控制逻…