在Android世界中前前后后出现过很多网络框架，比如：HttpUrlConnection、HttpClient、Volley、AsyncHttpClient、okhttp、Retrofit等。其中目前比较流行的当属okhttp与Retrofit莫属了，其中Retrofit是基于okhttp的基础上进行的进一步的封装得到的，所以对于知其然还要知其所以然的monkey来说了解okhttp的源码是很必要的，所以下面请跟我一起来看看okhttp的源码吧。
本篇主要三个部分，分别是：OKHTTP流程分析、连接池实现、Dispatcher详解。
一、OKHTTP流程分析
想要分析源码我们需要一个切入口进行跟进，最好的入口不外乎日常使用的流程。我们对okhttp的一般使用方法：
//同步
OkHttpClient client = new OkHttpClient();
Request request = new Request.Builder()
.url(url)
.build();
Response response = client.newCall(request).execute();
if (response.isSuccessful()) {
//...
} else {
//...
}
//异步
OkHttpClient client = new OkHttpClient();
Request request = new Request.Builder()
.url(url)
.build();
client.newCall(request).enqueue(new Callback() {
@Override public void onFailure(Call call, IOException e) {
//...
}
@Override public void onResponse(Call call, Response response) throws IOException {
//...
}
});
由上可知，先初始化一个OkHttpClient，然后调用其newCall()函数返回一个call类。看一下其实现：
@Override
public Call newCall(Request request) {
return RealCall.newRealCall(this, request, false /* for web socket */);
}
直接返回了RealCall的newRealCall()，其中RealCall是call接口的实现类。
static RealCall newRealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
// Safely publish the Call instance to the EventListener.
RealCall call = new RealCall(client, originalRequest, forWebSocket);
call.eventListener = client.eventListenerFactory().create(call);
return call;
}
通过newCall()最终返回了一个RealCall的实例。之后同步调用RealCall的 execute()，异步调用enqueue，我们先来看同步：
@Override
public Response execute() throws IOException {
try {
Response result = getResponseWithInterceptorChain();//执行request请求
return result;
} catch (IOException e) {
//...
} finally {
client.dispatcher().finished(this);//切换下一request执行
}
}
简化了以下只剩下最简单的代码，可以看到直接调用了getResponseWithInterceptorChain()
再看异步：
@Override
public void enqueue(Callback responseCallback) {
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
synchronized void enqueue(AsyncCall call) {
if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
runningAsyncCalls.add(call);//放到正在执行队列
executorService().execute(call);//执行
} else {
readyAsyncCalls.add(call);//放到等待队列
}
}
可以看到生成了一个AsyncCall并在executorService().execute(call);进行了执行，看下AsyncCall：
final class AsyncCall extends NamedRunnable {
AsyncCall(Callback responseCallback) { }
@Override
protected void execute() {
try {
Response response = getResponseWithInterceptorChain();
//...
} catch (IOException e) {
//...
} finally {
client.dispatcher().finished(this);
}
}
}
和同步的execute非常相似，都最终调用了getResponseWithInterceptorChain()，其实同步和异步的区别就是一个直接执行了，一个使用了线程池，具体实现值得学习一下，感兴趣可以看下源码，不在赘述。
下面就主要看getResponseWithInterceptorChain()的实现了：
Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
List<Interceptor> interceptors = new ArrayList<>();
interceptors.addAll(client.interceptors());
interceptors.add(retryAndFollowUpInterceptor);
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, null, null, null, 0,
originalRequest, this, eventListener, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
return chain.proceed(originalRequest);
}
其中就做了两件事：1)创建了一些interceptors；2)新建了RealInterceptorChain并调用了它的proceed的方法。我们直接看该proceed做了什么
@Override
public Response proceed(Request request) throws IOException {
return proceed(request, streamAllocation, httpCodec, connection);
}
public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
RealConnection connection) throws IOException {
// Call the next interceptor in the chain.
RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec,
connection, index + 1, request, call, eventListener, connectTimeout, readTimeout, writeTimeout);
Interceptor interceptor = interceptors.get(index);
Response response = interceptor.intercept(next);
return response;
}
可知，其中又新建了只有index + 1不同的RealInterceptorChain并执行了上面的interceptors链表中某一个interceptor的intercept(next)，我们看下intercept(next)在干嘛(以retryAndFollowUpInterceptor为例)
@Override
public Response intercept(Chain chain) throws IOException {
StreamAllocation streamAllocation = new StreamAllocation(client.connectionPool(),
createAddress(request.url()), call, eventListener, callStackTrace);
while (true) {//如果结果正常则返回，否则使用更新的request进行第二次处理。
Response response = realChain.proceed(request, streamAllocation, null, null);//直接传递给下一个
-----------------------------------------------------------------------------------------
Request followUp = followUpRequest(response, streamAllocation.route());//处理重定向等逻辑.
if (followUp == null) {
return response;
}
//建立新的StreamAllocation，以供followUp 使用。
streamAllocation = new StreamAllocation(client.connectionPool(),
createAddress(followUp.url()), call, eventListener, callStackTrace);
request = followUp;
}
}
可见，主要做了三件事：
1）新建StreamAllocation ；用于建立网络连接
2）执行上面传递过来的next.proceed()执行下一个interceptor的intercept(),直到interceptors全部被执行完。
3）调用followUpRequest，进行检查response中是否含有重定向，没有返回null，有返回新的request。
其中while死循环的目的就是持续检查返回结果中是否有重定向，直到没有在跳出。
注意：其中的分割线，分割线以上都是用来处理request逻辑的，分割线以下都是用来处理response逻辑的，因为realChain.proceed会持续循环调用，直到返回结果，调用链如下图。
image.png
总体流程图：
image.png
其中，总体的调用流程就是上面的部分，下面介绍下几个拦截器的作用，分析方法和上面retryAndFollowUpInterceptor一样，只要看intercept方法就可以了。
retryAndFollowUpInterceptor();//主要负责重定向拦截相关
BridgeInterceptor(client.cookieJar());//主要负责请求相应头添加去除等逻辑
CacheInterceptor(client.internalCache());//主要负责缓存相关，使用了diskLruCache()
ConnectInterceptor(client);//主要负责网络连接相关
CallServerInterceptor(forWebSocket);//最后一个拦截器，负责与服务器建立 Socket 连接
client.interceptors();//用户自定
client.networkInterceptors();//用户自定义，用户可以在拦截器的一头一尾进行自定义功能的数据处理，
//并在client初始化时传入进去即可。
如果只是看流程的到这里就可以结束了。
二、连接池实现
下面我们看一下连接池ConnectionPool实现，我们都直到网络连接是基于TCP/IP基础的，所以必须要经历三次握手与四次挥手，频繁的建立连接与断开连接是很耗时的，所以就建立连接池来实现对连接的最大复用。
ConnectionPool内部维持了一个ArrayDeque来保存连接。
private final Deque<RealConnection> connections = new ArrayDeque<>();
public ConnectionPool() {
this(5, 5, TimeUnit.MINUTES);
}
可维持的最大数量默认5个，维持时间5分钟，会创建一个线程定时进行查询处理超时的链接。虽然ConnectionPool在client初始化时就传入了进来，但是直到ConnectInterceptor时才会调用进行查找，最终会调用其get方法：
@Nullable
RealConnection get(Address address, StreamAllocation streamAllocation, Route route) {
assert (Thread.holdsLock(this));
for (RealConnection connection : connections) {
if (connection.isEligible(address, route)) {
streamAllocation.acquire(connection, true);
return connection;
}
}
return null;
}
其中就是遍历所有链接connections对它进行检查是否可用，如果符合就返回，Connection代表真实的socket物理连接,如下图
image.png
下面看下isEligible做了什么：
/**
* Returns true if this connection can carry a stream allocation to {@code address}. If non-null
* {@code route} is the resolved route for a connection.
*/
public boolean isEligible(Address address, @Nullable Route route) {
// If this connection is not accepting new streams, we're done.
if (allocations.size() >= allocationLimit || noNewStreams) return false;
// If the non-host fields of the address don't overlap, we're done.
if (!Internal.instance.equalsNonHost(this.route.address(), address)) return false;
// If the host exactly matches, we're done: this connection can carry the address.
if (address.url().host().equals(this.route().address().url().host())) {
return true; // This connection is a perfect match.
}
//...
return true; // The caller's address can be carried by this connection.
}
可见，先是检查是否超过每个连接所能链接的数量，默认是:1，然后检查代理、DNS、目标主机等是否相同，如果都相同就说明该链接可用，否则不可用。
接着看一下streamAllocation.acquire(connection, true);
public void acquire(RealConnection connection, boolean reportedAcquired) {
assert (Thread.holdsLock(connectionPool));
if (this.connection != null) throw new IllegalStateException();
this.connection = connection;
this.reportedAcquired = reportedAcquired;
connection.allocations.add(new StreamAllocationReference(this, callStackTrace));
}
可见，将可用的connection保存到streamAllocation中，并将该streamAllocation添加到connection.allocations链表中
public final List<Reference<StreamAllocation>> allocations = new ArrayList<>();
从中我们就可以答题理解connection与StreamAllocation的关系了：StreamAllocation是一个负责查找可用连接，完成链接回调的类，它是物理连接connection的逻辑代表，直到ConnectInterceptor调用时两者合二为一。
三、Dispatcher详解
在第一部分有讲过同步和异步的处理，其实他们都是通过Dispatcher分发实现的。了解一下Dispatcher中几个重要的变量：
/** 异步等待队列 */
private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
/** 异步执行队列 */
private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
/** 同步执行队列 */
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
Dispatcher就是对这三个变量的维护来实现对request的管理。
private int maxRequests = 64;
private int maxRequestsPerHost = 5;
maxRequests 用来限制异步请求runningAsyncCalls 的最大长度，默认为64个；maxRequestsPerHost用来限制runningAsyncCalls 中相同host的最大请求数量，默认为5个；超过以上两个限制的request都将放入readyAsyncCalls 队列中。
同步请求在
Response response = client.newCall(request).execute();
后直接调用
client.dispatcher().executed(this);//入列
synchronized void executed(RealCall call) {
runningSyncCalls.add(call);
}
加入了runningSyncCalls队列，等到执行结束后回调finished方法
client.dispatcher().finished(this);
void finished(RealCall call) {
finished(runningSyncCalls, call, false);
}
异步请求在
client.newCall(request).enqueue(new Callback())
执行结束后回调
client.dispatcher().finished(this);
void finished(AsyncCall call) {
finished(runningAsyncCalls, call, true);
}
是的，它们最后执行的finished是同一个泛型函数，只是最后一个参数不同而已。
private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
synchronized (this) {
//在runningSyncCalls队列中将该请求移除
if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
if (promoteCalls) promoteCalls();
}
}
可见，同步移除之后直接退出了，异步执行了promoteCalls：
private void promoteCalls() {
if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity.
if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote.
for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
AsyncCall call = i.next();
if (runningCallsForHost(call) < maxRequestsPerHost) {
i.remove();
runningAsyncCalls.add(call);
executorService().execute(call);
}
if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
}
}
可见，直接取出等待队列中的一个request执行。
网友评论