前言
几个月前，跟过 OkHttp 的流程源码，但是时间久了，现在能够回想起来的的，只有几个拦截器了，那我岂不是没什么收获了。所以，好好想想，我从 OkHttp 中能够学到什么
疑问
- OkHttp 是怎么拆分功能的，大概有几个模块
- 它所用到的责任链模式，在实际开发中适合哪些场景
- 它是怎么使用线程池的，这么用有什么好处
OkHttp 是怎么拆分功能的，大概有几个模块
作为一个网络框架，最核心的功能就是发起请求，处理响应了，这俩个是功能部分， OkHttp 使用 Dispatcher 执行任务，内部是一个高并发的线程池，另外整个流程的处理使用到了Interceptor 拦截器
请求执行的调用流程
- Recall.enqueue(Callback)
- client.dispatcher().enqueue(new AsyncCall(responseCallback))
停一下,跟进一下分发器器的 enqueue 方法,内部时怎么处理异步请求的
// 创建 AsyncCall 对象
void enqueue(AsyncCall call) {
synchronized (this) {
// 添加到待执行队列中,双向队列
readyAsyncCalls.add(call);
}
// 执行的重要代码
promoteAndExecute();
}
private boolean promoteAndExecute() {
assert (!Thread.holdsLock(this));
// 创建一个可执行队列,目的是限制 64 个最大连接数,每个 Host 最多 5 个连接的限制
List<AsyncCall> executableCalls = new ArrayList<>();
boolean isRunning;
synchronized (this) {
for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
AsyncCall asyncCall = i.next();
if (runningAsyncCalls.size() >= maxRequests) break; // Max capacity.
if (runningCallsForHost(asyncCall) >= maxRequestsPerHost) continue; // Host max capacity.
i.remove();
// 符合条件的,从准备中的队列挪到其他两个队列中
executableCalls.add(asyncCall);
runningAsyncCalls.add(asyncCall);
}
isRunning = runningCallsCount() > 0;
}
// 放到线程池中执行
for (int i = 0, size = executableCalls.size(); i < size; i++) {
AsyncCall asyncCall = executableCalls.get(i);
asyncCall.executeOn(executorService());
}
return isRunning;
}
首先 AsyncCall 是实现了 Runnable 的一个类,也难怪,毕竟是要放到线程池中执行的
响应执行的调用过程
刚才看了请求入队后,经过筛选,开始遍历放到线程池中去执行,那下来就是等待响应了
具体代码在 Recall 类的 run() 方法中,里面就是我们之前了解到的使用责任链模式-拦截器的代码了
为什么要使用双端队列 ArrayDeque ?
首先 Jdk 提供的双端队列主要有两个:
- LinkedList 链表实现的双端队列
- ArrayDeque 循环数组实现的双端队列
那为什么偏偏就选了 ArrayDeque 了,因为效率高. Jdk 的说明中就说了,ArrayDeque 作为队列使用时,将比 LinkedList 更快
ArrayQueue 是线程不安全的,Okhttp 是怎样保证同步问题的? synchronized 关键字
OkHttp 是怎样使用责任链模式的，Android 源码中还有其他地方用到了嘛？
那就首先要补习下责任链模式是什么，以及怎样用代码实现
在网上查了很多文档，有一种表述我觉得很形象，小张去外地出差回来，其中 2w 要去招公司报销，他去找到组长
- 组长看到发票，面值超过了权限，说让小张去找主管
- 主管一看，自己最大只能签 3k 的，让其去找经理
- 经理最大只能批 1w 的，让小张去找老板
- 最终老板签字处理
整个流程涉及到多个类（组长、主管、经理等），一级一级的处理，最终处理结束
Android 源码中 View 的事件分发也是使用了责任链模式，其中被分发的 MotionEvent 经过 ViewGroup 层层分发，最终被消费或者重新返回到最上层的 View
- 那 OkHttp 是怎样实现责任链的呢？
Let is see fucking code ，Woohooo
// 同步执行的代码中
@Override protected void execute() {
boolean signalledCallback = false;
transmitter.timeoutEnter();
try {
// 使用责任链处理请求响应，重点看这里
Response response = getResponseWithInterceptorChain();
......
}
}
Response getResponseWithInterceptorChain() throws IOException {
// 创建一个拦截器队列
List<Interceptor> interceptors = new ArrayList<>();
interceptors.addAll(client.interceptors());
interceptors.add(new RetryAndFollowUpInterceptor(client));
interceptors.add(new BridgeInterceptor(client.cookieJar()));
interceptors.add(new CacheInterceptor(client.internalCache()));
interceptors.add(new ConnectInterceptor(client));
if (!forWebSocket) {
interceptors.addAll(client.networkInterceptors());
}
interceptors.add(new CallServerInterceptor(forWebSocket));
// 添加完了，开始逐个处理了
Interceptor.Chain chain = new RealInterceptorChain(interceptors, transmitter, null, 0,
originalRequest, this, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
boolean calledNoMoreExchanges = false;
try {
// proceed 执行后得到响应结果
Response response = chain.proceed(originalRequest);
if (transmitter.isCanceled()) {
closeQuietly(response);
throw new IOException("Canceled");
}
return response;
} catch (IOException e) {
calledNoMoreExchanges = true;
throw transmitter.noMoreExchanges(e);
} finally {
if (!calledNoMoreExchanges) {
transmitter.noMoreExchanges(null);
}
}
}
这里使用了队列保存所有的拦截器，然后一股脑传进了 RealInterceptorChain 对象中，最后调用 proceed 就有个返回结果，停，一下子就获取到了结果了嘛？ 进去看看 RealInterceptorChain 里面是怎么处理的
public Response proceed(Request request, Transmitter transmitter, @Nullable Exchange exchange)
throws IOException {
... 省略部分代码
// 看到 next 我就想起来链表里的 next，这个 next 是干嘛的呢？
RealInterceptorChain next = new RealInterceptorChain(interceptors, transmitter, exchange,
index + 1, request, call, connectTimeout, readTimeout, writeTimeout);
// 根据 index 获取对应位置的拦截器
Interceptor interceptor = interceptors.get(index);
// 调用拦截器的拦截方法
Response response = interceptor.intercept(next);
// Confirm that the next interceptor made its required call to chain.proceed()
... 省略部分代码
return response;
}
有一个地方没看明白，创建 Chain 后，调用 proceed ，刚开始 index 为 0，那整个队列是怎么遍历的，没看到有循环遍历语句啊
进入 interceptor.intercept(next) 后，一切截然而止了，怀着好奇心，点开了 interceptor 的实现类 CacheInterceptor ，果不其然，在 intercept 方法中，再次看到了 proceed 的身影，终于破案了
@Override public Response intercept(Chain chain) throws IOException {
... 省略部分代码
// 每调用一次，内部的 index 自增，就意味着不断的传递到下一级拦截器
networkResponse = chain.proceed(networkRequest);
... 省略部分代码
return response;
OkHttp 是怎么使用线程池的
最后一个问题，怎么使用线程池的，首先补习下线程池的各个参数的含义，以及线程池的工作原理
// OkHttp 中的线程池
public synchronized ExecutorService executorService() {
if (executorService == null) {
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
new SynchronousQueue<>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}
// 线程池的构造方法
// corePoolSize – the number of threads to keep in the pool, even if they are idle, unless allowCoreThreadTimeOut is set
// maximumPoolSize – the maximum number of threads to allow in the pool
// keepAliveTime – when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.
// unit – the time unit for the keepAliveTime argument
// workQueue – the queue to use for holding tasks before they are executed. This queue will hold only the Runnable tasks submitted by the execute method.
// threadFactory – the factory to use when the executor creates a new thread
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
创建参数：
- 核心线程数，核心线程将保持存活，及时是空闲的，这里是 0
- 池中允许的最大线程数，这里是 Int 的最大值，实际到不了这么多，OkHttp 有 64 最大连接数的限制
- 等待时间可以理解为非核心线程等待任务时的超时时间 ，这里为 60 秒
- 等待时间的单位
- 工作队列，这里是一个同步的阻塞队列，内部没有容器，传入一个时就会阻塞下一个的传入
- 线程工厂
- 为什么 OkHttp 要这么设置线程池，有什么好处呢？
其实这种参数设置,就是Excutor.newCachedThreadPool() , 首先一个阻塞的同步队列,内部没有容器,意味着什么呢,每当一个网络请求发起,只要核心线程满了,就会在池中创建新的线程
如果线程池中的线程数大于核心线程数且队列满了，且线程数小于最大线程数，则会创建新的线程,刚好 Okhttp 的最大线程数时是一个极大值,那就会不断创建线程,是一个高并发的线程池
最后
您的点赞收藏就是对我最大的鼓励！ 欢迎关注我，分享Android干货，交流Android技术。 对文章有何见解，或者有何技术问题，欢迎在评论区一起留言讨论！最后给大家分享一些Android相关的视频教程，感兴趣的朋友可以去看看。
网友评论