开源库源码分析: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.mzph.cn/news/79506.shtml

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

相关文章

前端设计模式基础笔记

前端设计模式是指在前端开发中经常使用的一些解决问题的模式或思想。它们是经过实践证明的最佳实践&#xff0c;可以帮助我们更好地组织和管理我们的代码。 一、单例模式&#xff08;Singleton Pattern&#xff09; 单例模式是一种创建型模式&#xff0c;它保证一个类只有一个…

⑩ vue新特性

ref 或者reactive ref相当于data methods props和context props &#xff01;&#xff01;&#xff01;setup中没有this关键字&#xff0c;使用context&#xff08;简写&#xff1a;ctx&#xff09;就是this 在steup中使用生命周期函数 Provide / Inject 1、原来是 a传…

基于YOLOv2和传感器的多功能门禁系统

文章和项目源码已经归档至【Github仓库&#xff1a;https://github.com/timerring/face-recognition-door 】或者公众号【AIShareLab】回复 人脸识别门禁 也可获取。 文章目录 1.系统制作方案概述1.1系统设计的立意1.2系统的主要组成1.3系统的制作方案1.3.1制作方案框图1.3.2制…

python-xpath语法-爬取彼岸图4k高清动漫壁纸

安装 pip install lxml导入 from lxml import etreexpath使用路径表达式提取html文档中的元素或元素集&#xff0c;然后元素通过沿路径path或步steps来选取数据 XPath常用语法格式 表达式描述div选取div元素的所有子元素/div选取根元素divul//li选取ul元素下的所有li子元素…

大数据-玩转数据-oracel字符串分割转化为多列

一、建表 create table split_string_test(id integer primary key,test_string varchar2(500) );二、插入测试数据 insert into split_string_test values(1, 10,11,12,13,14,22); insert into split_string_test values(2, 22,23,24); insert into split_string_test valu…

TMS320F280049最小系统原理图

TMS320F280049最小系统原理图 1.概述2. 典型的 F2800x 系统方框图3. 最小系统原理图设计3.1 封装和器件决策3.2 电源及去耦电容3.3 晶振3.4 GPIO3.5 ADC模块3.6 JTAG 最近做了个新车规项目&#xff0c;第一次接触TMS320F280049&#xff0c;记录一下&#xff0c;最小系统原理图设…

PostgreSQL 事务并发锁

文章目录 PostgreSQL 事务大家都知道的 ACID事务的基本使用保存点 PostgreSQL 并发并发问题MVCC PostgreSQL 锁机制表锁行锁 总结 PostgreSQL 事务 大家都知道的 ACID 在日常操作中&#xff0c;对于一组相关操作&#xff0c;通常要求要么都成功&#xff0c;要么都失败。在关系…

Linux下运行Jmeter压测

一、在Linux服务器先安装SDK 1、先从官网下载jdk1.8.0_131.tar.gz&#xff0c;再从本地上传到Linux服务器 2、解压&#xff1a;tar -xzf jdk1.8.0_131.tar.gz&#xff0c;生成文件夹 jdk1.8.0_131 3、在/usr/目录下创建java文件夹&#xff0c;再将 jdk1.8.0_131目录移动到/u…

长胜证券:十大流通股东占比例高好还是低好?

近年来&#xff0c;跟着我国本钱商场的不断发展&#xff0c;越来越多的投资者开始了解和关注股东占比这个目标。而在股东占比中&#xff0c;十大流转股东的持股份额是一个重要的目标。可是&#xff0c;关于投资者来说&#xff0c;十大流转股东占比是高好还是低好&#xff1f;本…

微信小程序+echart实现点亮旅游地图

背景 最近看抖音有个很火的特效就是点亮地图&#xff0c;去过哪些地方&#xff0c;于是乎自己也想做一个&#xff0c;结合自己之前做的以家庭为单位的小程序&#xff0c;可以考虑做一个家庭一起点亮地图的功能。 效果图 过程 1&#xff0c;首先就是得去下微信小程序适配的ec…

牛客: BM4 合并两个排序的链表

牛客: BM4 合并两个排序的链表 文章目录 牛客: BM4 合并两个排序的链表题目描述题解思路题解代码 题目描述 题解思路 以链表一为主链表,遍历两条链表 若当前链表二的节点val小于当前链表一的下一个节点val,则将链表链表二的该节点连到链表一的节点的下一个,链表一的当前节点往…

智慧工地:实现作业区域安全管控

智慧工地是围绕工程现场人、机、料、法、环及施工过程中质量、安全、进度、成本等各项数据满足工地多角色、多视角的有效监管,实现工程建设管理的降本增效。 建设工程安全文明施工与质量提升,全方位的监测施工人员、各类器械设备、消防安全隐患&#xff0c;并提前对风险进行预警…

CANoe-Model Editor无法修改ARXML文件的问题、E2E在SOME/IP通信中的使用问题

1、Model Editor无法修改ARXML文件的问题 在CANoe 15软件版本中,Communication Setup导入arxml文件后,可以在model editor中打开arxml并修改配置。关闭model editor后再打开,可以看到修改的配置被保存了。 但是,当我把arxml文件从Communication Setup中移除后,再导入。此…

MacBook苹果电脑重装、降级系统

1、下载balenaEtcher镜像启动盘制作工具 https://tails.net/etcher/balenaEtcher-portable.exe 2、选择从文件烧录选择下载好的Mac 镜像文件 百度网盘 请输入提取码&#xff08;Mac OS 10.10-12版本镜像文件&#xff09; 第二步选择目标磁盘&#xff0c;这里需要准备一块1…

Dockerfile解析

Dockerfile是什么 Dockerfile是用来构建Docker镜像的文本文件&#xff0c;是由一条条构建镜像所需的指令和参数构成的脚本。 概述 官网 https://docs.docker.com/engine/reference/builder/ 构建三步骤 编写Dockerfile文件docker build命令构建镜像&#xff1a;docker bui…

【教程】IDEA操作GIT

不小心推送代码之后 进行回退 1 找到需要回退的记录 比如要回退13分钟之前提交的代码 选中 右键还原提交 最后再重新推送被还原的提交 就可以了

2327. 知道秘密的人数;1722. 执行交换操作后的最小汉明距离;2537. 统计好子数组的数目

2327. 知道秘密的人数 核心思想&#xff1a;动态规划&#xff0c;每天的人可以分为三种&#xff0c;可分享秘密的人&#xff0c;不可分享秘密的人&#xff0c;忘记秘密的人。定义f[i]为第i天可分享秘密的人&#xff0c;那么第(idelay ,iforget)天&#xff0c;会增加f[i]个可分…

性能测试之压力测试

文章目录 一.基本介绍二.性能指标三.下载安装JMeter1.下载安装包2.启动JMeter 四.使用JMeter1.模拟用户请求2.填写测试地址3.接收测试结果4.结果解释 一.基本介绍 压力测试考察当前软硬件条件下系统所能承受的最大负荷并找到系统瓶颈所在。压测是为了系统在线上的处理能力和稳定…

Kubernetes学习篇之组件

Kubernetes学习篇之组件 文章目录 Kubernetes学习篇之组件前言概述控制平面组件(Control Plane Components)kube-apiserveretcdkube-schedulerkube-controller-managercloud-controller-manager Node 组件kubeletkube-proxy容器运行时(Container Runtime) 插件(Addons)DNSWeb界…

SpringSecurity学习

1.认证 密码校验用户 密码加密存储 Configuration public class SecurityConfig extends WebSecurityConfigurerAdapter {Beanpublic PasswordEncoder passwordEncoder(){return new BCryptPasswordEncoder();}} 我们没有以上代码配置&#xff0c;默认明文存储, {id}password…