Android 主流开源框架（三）OkHttp 源码解析
2019-10-21 本文已影响0人
wildma
前言
最近有个想法——就是把 Android 主流开源框架进行深入分析，然后写成一系列文章，包括该框架的详细使用与源码解析。目的是通过鉴赏大神的源码来了解框架底层的原理，也就是做到不仅要知其然，还要知其所以然。
这里我说下自己阅读源码的经验，我一般都是按照平时使用某个框架或者某个系统源码的使用流程入手的，首先要知道怎么使用，然后再去深究每一步底层做了什么，用了哪些好的设计模式，为什么要这么设计。
系列文章：
- Android 主流开源框架（一）OkHttp 铺垫-HttpClient 与 HttpURLConnection 使用详解
- Android 主流开源框架（二）OkHttp 使用详解
- Android 主流开源框架（三）OkHttp 源码解析
- Android 主流开源框架（四）Retrofit 使用详解
- Android 主流开源框架（五）Retrofit 源码解析
- Android 主流开源框架（六）Glide 的执行流程源码解析
- 更多框架持续更新中...
更多干货请关注 AndroidNotes
一、OkHttp 的基本使用示例
1.1 同步 GET 请求
// （1）创建 OkHttpClient 对象
OkHttpClient client = new OkHttpClient();
// （2）创建 Request 对象
Request request = new Request.Builder()
.url(url)
.build();
// （3）创建 Call 对象。
Call call = client.newCall(request);
// （4）发送请求并获取服务器返回的数据
Response response = call.execute();
// （5）取出相应的数据
String data = response.body().string();
1.2 异步 GET 请求
// （1）创建 OkHttpClient 对象
OkHttpClient client = new OkHttpClient();
// （2）创建 Request 对象
Request request = new Request.Builder()
.url(url)
.build();
// （3）创建 Call 对象。
Call call = client.newCall(request);
// （4）发送请求并获取服务器返回的数据
call.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
}
@Override
public void onResponse(Call call, Response response) throws IOException {
// （5）取出相应的数据
String data = response.body().string();
}
});
可以看到不管是同步请求还是异步请求，OkHttp 的基本使用就只有 5 步。同步请求与异步请求唯一不同的就是第 （4） 步，前者使用同步方法 execute()，后者使用异步方法 enqueue()。接下来我们就根据这 5 步进行源码阅读。
更多 OkHttp 的使用方法可以看我之前写的文章 Android 主流开源框架（二）OkHttp 使用详解
二、OkHttp 源码分析
源码版本：3.11.0
2.1 （1）创建 OkHttpClient 对象
OkHttpClient client = new OkHttpClient();
首先我们点击创建的 OkHttpClient 对象进去源码是这样的：
/*OkHttpClient*/
public OkHttpClient() {
this(new Builder());
}
然后是走了有参构造：
/*OkHttpClient*/
OkHttpClient(Builder builder) {
this.dispatcher = builder.dispatcher;
this.proxy = builder.proxy;
this.protocols = builder.protocols;
this.connectionSpecs = builder.connectionSpecs;
this.interceptors = Util.immutableList(builder.interceptors);
this.networkInterceptors = Util.immutableList(builder.networkInterceptors);
this.eventListenerFactory = builder.eventListenerFactory;
this.proxySelector = builder.proxySelector;
this.cookieJar = builder.cookieJar;
this.cache = builder.cache;
this.internalCache = builder.internalCache;
this.socketFactory = builder.socketFactory;
boolean isTLS = false;
for (ConnectionSpec spec : connectionSpecs) {
isTLS = isTLS || spec.isTls();
}
if (builder.sslSocketFactory != null || !isTLS) {
this.sslSocketFactory = builder.sslSocketFactory;
this.certificateChainCleaner = builder.certificateChainCleaner;
} else {
X509TrustManager trustManager = Util.platformTrustManager();
this.sslSocketFactory = newSslSocketFactory(trustManager);
this.certificateChainCleaner = CertificateChainCleaner.get(trustManager);
}
if (sslSocketFactory != null) {
Platform.get().configureSslSocketFactory(sslSocketFactory);
}
this.hostnameVerifier = builder.hostnameVerifier;
this.certificatePinner = builder.certificatePinner.withCertificateChainCleaner(
certificateChainCleaner);
this.proxyAuthenticator = builder.proxyAuthenticator;
this.authenticator = builder.authenticator;
this.connectionPool = builder.connectionPool;
this.dns = builder.dns;
this.followSslRedirects = builder.followSslRedirects;
this.followRedirects = builder.followRedirects;
this.retryOnConnectionFailure = builder.retryOnConnectionFailure;
this.connectTimeout = builder.connectTimeout;
this.readTimeout = builder.readTimeout;
this.writeTimeout = builder.writeTimeout;
this.pingInterval = builder.pingInterval;
if (interceptors.contains(null)) {
throw new IllegalStateException("Null interceptor: " + interceptors);
}
if (networkInterceptors.contains(null)) {
throw new IllegalStateException("Null network interceptor: " + networkInterceptors);
}
}
可以看到有很多常量，这里使用了建造者模式，所以这些常量可以通过 build() 进行配置。如果不进行配置则使用无参构造中传进来的默认配置，每个常量的意思具体如下：
/*OkHttpClient*/
public Builder() {
dispatcher = new Dispatcher();// 分发器
protocols = DEFAULT_PROTOCOLS;// HTTP 协议
connectionSpecs = DEFAULT_CONNECTION_SPECS;// 传输层版本和连接协议
eventListenerFactory = EventListener.factory(EventListener.NONE);// 事件监听工厂
proxySelector = ProxySelector.getDefault();// 代理选择器
cookieJar = CookieJar.NO_COOKIES;// cookie
socketFactory = SocketFactory.getDefault();// socket 工厂
hostnameVerifier = OkHostnameVerifier.INSTANCE;// 主机名字确认
certificatePinner = CertificatePinner.DEFAULT;// 证书链
proxyAuthenticator = Authenticator.NONE;// 代理服务器身份验证
authenticator = Authenticator.NONE;// 源服务器身份验证
connectionPool = new ConnectionPool();// 连接池
dns = Dns.SYSTEM;// 域名
followSslRedirects = true;// 是否遵循 ssl 重定向
followRedirects = true;// 是否遵循重定向
retryOnConnectionFailure = true;// 连接失败的时候是否重试
connectTimeout = 10_000;// 连接超时
readTimeout = 10_000;// 读超时
writeTimeout = 10_000;// 写超时
pingInterval = 0;// HTTP / 2 和 Web 套接字 ping 之间的时间间隔
}
2.2 （2）创建 Request 对象
Request request = new Request.Builder()
.url(url)
.build();
可以看到，这里同样使用了建造者模式，我们点击 Request 进去看看：
/*Request*/
//...
final HttpUrl url;
final String method;
final Headers headers;
final @Nullable RequestBody body;
final Map<Class<?>, Object> tags;
//...
发现 Request 还是比较简单的，只是用来设置一些请求链接（url）、请求方法（method）、请求头（headers）、请求体（body）、标签（tag，可作为取消请求的标记）。
2.3 （3）创建 Call 对象
Call call = client.newCall(request);
我们点击 newCall() 方法进去看看：
/*OkHttpClient*/
@Override public Call newCall(Request request) {
return RealCall.newRealCall(this, request, false /* for web socket */);
}
发现是调用了 RealCall 的 newRealCall() 方法，并传入了 OkHttpClient 与 Request 对象。
再跟进到 newRealCall() 方法：
/*RealCall*/
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;
}
发现是创建了一个 RealCall 对象，并返回给上一层。RealCall 是 Call 的实现类，Call 定义了请求相关的操作，例如同步异步、取消请求等方法。所以后续的请求相关操作基本都是在调用 Call 定义的方法，而这些方法真正的执行是它的实现类 RealCall。
最后看看 RealCall 的构造函数，该函数是比较简单的，只是赋值一些常量，然后创建了重试与重定向拦截器（RetryAndFollowUpInterceptor）（这个后面会讲）：
/*RealCall*/
private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
this.client = client;
this.originalRequest = originalRequest;
this.forWebSocket = forWebSocket;
this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
}
2.4 （4）发送请求并获取服务器返回的数据
前面我们已经说了，同步请求与异步请求唯一不同的就是第 （4） 步，前者使用同步方法 execute()，后者使用异步方法 enqueue()。所以我们分 2 种情况来讲。
2.4.1 同步请求
Response response = call.execute();
我们点击 execute() 方法进去看看：
/*RealCall*/
@Override public Response execute() throws IOException {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
captureCallStackTrace();
eventListener.callStart(this);
try {
client.dispatcher().executed(this);// (1)
Response result = getResponseWithInterceptorChain();// (2)
if (result == null) throw new IOException("Canceled");
return result;
} catch (IOException e) {
eventListener.callFailed(this, e);
throw e;
} finally {
client.dispatcher().finished(this);// (3)
}
}
源码中我标注了 3 个关注点，点击关注点（1）的 executed() 方法进去，可以看到是将传进来的 RealCall 加入了一个双端队列：
/*Dispatcher*/
synchronized void executed(RealCall call) {
runningSyncCalls.add(call);
}
其中 runningSyncCalls 是一个双端队列，用来记录正在运行的同步请求队列：
/*Dispatcher*/
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
关注点（2）返回了一个 Response，也就是服务器返回的数据，说明请求就是在这里执行了，这个是我们要研究的重点，放到后面再说。
点击关注点（3）的 finished() 方法进去，是这样的：
/*Dispatcher*/
void finished(RealCall call) {
finished(runningSyncCalls, call, false);
}
private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
int runningCallsCount;
Runnable idleCallback;
synchronized (this) {
if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");// (1)
if (promoteCalls) promoteCalls();
runningCallsCount = runningCallsCount();
idleCallback = this.idleCallback;
}
if (runningCallsCount == 0 && idleCallback != null) {
idleCallback.run();
}
}
可以看到关注点（1）calls.remove(call) 只是把当前 RealCall 又从正在运行的同步请求队列中移除了，说明请求已经完成了。
你应该注意到了，上面还有个 dispatcher 没讲到，其实这是一个分发器，是用来对请求进行分发的。我们刚刚也分析了在同步请求中涉及到的 dispatcher 只是用来记录正在运行的同步请求队列，然后请求完成就移除掉。所以这个分发器主要用在异步请求中，我们放到异步请求中再去讲。
2.4.2 异步请求
call.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
}
@Override
public void onResponse(Call call, Response response) throws IOException {
}
});
我们点击 enqueue() 方法进去看看：
/*RealCall*/
@Override public void enqueue(Callback responseCallback) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
captureCallStackTrace();
eventListener.callStart(this);
client.dispatcher().enqueue(new AsyncCall(responseCallback));// (1)
}
前面几行与同步请求源码一样，我们点击关注点（1）的 enqueue() 方法进去看看：
/*Dispatcher*/
synchronized void enqueue(AsyncCall call) {
if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
runningAsyncCalls.add(call);
executorService().execute(call);
} else {
readyAsyncCalls.add(call);
}
}
可以看到这里面涉及到很多 Dispatcher 对象里面的常量与变量，所以也能看出 Dispatcher 主要用在异步请求中。先看下 Dispatcher 对象里面的常量与变量：
/*Dispatcher*/
// 最大并发请求数
private int maxRequests = 64;
// 每个主机最大请求数
private int maxRequestsPerHost = 5;
// 每次调度程序变为空闲时调用的回调
private @Nullable Runnable idleCallback;
// 用来执行异步任务的线程池
private @Nullable ExecutorService executorService;
// 准备中的异步请求队列
private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
// 正在运行的异步请求队列
private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
// 正在运行的同步请求队列
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
弄懂了这些常量与变量的意思，就很好理解上面关注点（1）的 enqueue() 方法了，即如果 ”正在运行的异步请求队列数“ 小于 ”最大并发请求数“，并且 ”每个主机正在运行的请求数“ 小于 ”每个主机最大请求数“，则将当前请求继续加入 ”正在运行的异步请求队列“ 并在线程池中执行，否则将当前请求加入 ”准备中的异步请求队列“。
我们看到线程池中还传了一个 AsyncCall 进去，点击进去看看：
/*RealCall*/
final class AsyncCall extends NamedRunnable {
private final Callback responseCallback;
AsyncCall(Callback responseCallback) {
super("OkHttp %s", redactedUrl());
this.responseCallback = responseCallback;
}
String host() {
return originalRequest.url().host();
}
Request request() {
return originalRequest;
}
RealCall get() {
return RealCall.this;
}
@Override protected void execute() {
boolean signalledCallback = false;
try {
Response response = getResponseWithInterceptorChain();// (1)
if (retryAndFollowUpInterceptor.isCanceled()) {
signalledCallback = true;
responseCallback.onFailure(RealCall.this, new IOException("Canceled"));// (2)
} else {
signalledCallback = true;
responseCallback.onResponse(RealCall.this, response);// (3)
}
} catch (IOException e) {
if (signalledCallback) {
// Do not signal the callback twice!
Platform.get().log(INFO, "Callback failure for " + toLoggableString(), e);
} else {
eventListener.callFailed(RealCall.this, e);
responseCallback.onFailure(RealCall.this, e);// (4)
}
} finally {
client.dispatcher().finished(this);// (5)
}
}
}
发现他是 RealCall 的内部类，继承 NamedRunnable，实现了 Runnable。里面同样执行了 execute() 方法，仔细看这个方法与之前我们阅读同步请求中的 execute() 类似，关注点（1）（5）都是一样的，不同的是多了 2 个回调，因为是异步请求，所以需要把最终返回的结果通过 responseCallback 回调到最外层我们使用的地方去，其中（2）（4）是失败的回调，（3）是成功的回调。
到这里，OkHttp 基本使用的第（4）步除了 getResponseWithInterceptorChain() 方法，其他都看完了，下面就重点阅读这个方法。
2.4.3 拦截器
点击 getResponseWithInterceptorChain() 方法进去看看：
/*RealCall*/
Response getResponseWithInterceptorChain() throws IOException {
// 创建一个拦截器集合
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));
// (1) 构建责任链
Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
originalRequest, this, eventListener, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
// (2) 处理责任链中的拦截器
return chain.proceed(originalRequest);
}
可以看到，这里用到了很多拦截器，将这些拦截器构建成一条责任链，然后再一个个处理。这里用到了责任链模式，每个拦截器负责相应的功能，上一个拦截器完成会传给下一个拦截器，直到最后一个拦截器执行完再一层层向上返回 Response。
我们先验证下这个责任链的执行过程是否跟我说的一样，然后再看看每个拦截器的具体作用。这里我标记了 2 个关注点：
关注点（1）是构建一条责任链，并把责任链需要用到的参数传过去，其中参数 5 为责任链的索引，这里传 “0” 表示当前正在处理第一个拦截器。
关注点（2）是处理责任链中的拦截器，点击 proceed() 方法进去看看：
/*RealInterceptorChain*/
@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 {
if (index >= interceptors.size()) throw new AssertionError();
calls++;
// If we already have a stream, confirm that the incoming request will use it.
if (this.httpCodec != null && !this.connection.supportsUrl(request.url())) {
throw new IllegalStateException("network interceptor " + interceptors.get(index - 1)
+ " must retain the same host and port");
}
// If we already have a stream, confirm that this is the only call to chain.proceed().
if (this.httpCodec != null && calls > 1) {
throw new IllegalStateException("network interceptor " + interceptors.get(index - 1)
+ " must call proceed() exactly once");
}
//（1）start
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);
// （1）end
// Confirm that the next interceptor made its required call to chain.proceed().
if (httpCodec != null && index + 1 < interceptors.size() && next.calls != 1) {
throw new IllegalStateException("network interceptor " + interceptor
+ " must call proceed() exactly once");
}
// Confirm that the intercepted response isn't null.
if (response == null) {
throw new NullPointerException("interceptor " + interceptor + " returned null");
}
if (response.body() == null) {
throw new IllegalStateException(
"interceptor " + interceptor + " returned a response with no body");
}
return response;
}
可以看到，除了一些判断只需要看关注点（1）即可。这里会构建一个新的责任链，然后把责任链的索引加 1（为了下次从拦截器集合中取出下一个拦截器），接着从拦截器集合中取出当前拦截器并调用 intercept() 方法，这样如果这个拦截器可以完成任务会马上返回 Response，否则会在 intercept() 方法中继续处理责任链，因为该 intercept() 方法中会继续调用责任链的 proceed() 方法。看完源码确实跟我们之前设想的一样的，接下来我们看看每个拦截器的具体作用。
2.4.3.1 重试与重定向拦截器（RetryAndFollowUpInterceptor）
该拦截器主要负责失败后重连以及重定向，从前面的 proceed() 方法可知，每个拦截器被调用的方法都是 intercept() 方法，所以阅读拦截器的入口就是该方法。
重试与重定向拦截器中的 intercept() 方法如下：
/*RetryAndFollowUpInterceptor*/
@Override public Response intercept(Chain chain) throws IOException {
Request request = chain.request();
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Call call = realChain.call();
EventListener eventListener = realChain.eventListener();
// (1) 创建 StreamAllocation 对象，用来协调三个实体（Connections、Streams、Calls）之间的关系
StreamAllocation streamAllocation = new StreamAllocation(client.connectionPool(),
createAddress(request.url()), call, eventListener, callStackTrace);
this.streamAllocation = streamAllocation;
// 重定向次数
int followUpCount = 0;
Response priorResponse = null;
while (true) {
if (canceled) {
streamAllocation.release();
throw new IOException("Canceled");
}
Response response;
boolean releaseConnection = true;
try {
//（2）执行下一个拦截器
response = realChain.proceed(request, streamAllocation, null, null);
releaseConnection = false;
} catch (RouteException e) {
//（3）发生 Route 异常，则尝试进行恢复
if (!recover(e.getLastConnectException(), streamAllocation, false, request)) {
throw e.getFirstConnectException();
}
releaseConnection = false;
continue;
} catch (IOException e) {
//（4）发生 IO 异常，则尝试进行恢复
boolean requestSendStarted = !(e instanceof ConnectionShutdownException);
if (!recover(e, streamAllocation, requestSendStarted, request)) throw e;
releaseConnection = false;
continue;
} finally {
// 如果中途出现异常，则释放所有资源
if (releaseConnection) {
streamAllocation.streamFailed(null);
streamAllocation.release();
}
}
// 构建 body 为空的响应体
if (priorResponse != null) {
response = response.newBuilder()
.priorResponse(priorResponse.newBuilder()
.body(null)
.build())
.build();
}
Request followUp;
try {
// （5）检查是否需要重定向，不需要则 followUp 返回 null
followUp = followUpRequest(response, streamAllocation.route());
} catch (IOException e) {
streamAllocation.release();
throw e;
}
// （6）不需要重定向，则返回之前的 response
if (followUp == null) {
if (!forWebSocket) {
streamAllocation.release();
}
return response;
}
// 关闭资源
closeQuietly(response.body());
// 重定向次数大于最大值，则释放 StreamAllocation 并抛出异常
if (++followUpCount > MAX_FOLLOW_UPS) {
streamAllocation.release();
throw new ProtocolException("Too many follow-up requests: " + followUpCount);
}
if (followUp.body() instanceof UnrepeatableRequestBody) {
streamAllocation.release();
throw new HttpRetryException("Cannot retry streamed HTTP body", response.code());
}
// 如果该请求无法复用之前的连接，则释放后重新创建
if (!sameConnection(response, followUp.url())) {
streamAllocation.release();
streamAllocation = new StreamAllocation(client.connectionPool(),
createAddress(followUp.url()), call, eventListener, callStackTrace);
this.streamAllocation = streamAllocation;
} else if (streamAllocation.codec() != null) {
throw new IllegalStateException("Closing the body of " + response
+ " didn't close its backing stream. Bad interceptor?");
}
request = followUp;
priorResponse = response;
}
}
该方法的注释都写的比较详细了，我们重点看下我标记的关注点。
-
（1）：创建 StreamAllocation 对象，StreamAllocation 相当于一个管理类，用来协调三个实体（Connections、Streams、Calls）之间的关系。这里还传了一个 client.connectionPool()，它是第一步创建 OkHttpClient 对象的时候创建的，是一个连接池。它们会在后面的连接拦截器（ConnectInterceptor）中才被真正的使用到，后面会讲。
其中：
Connections：连接到远程服务器的物理套接字。
Streams：在连接上分层的逻辑 http 请求/响应对。
Calls：流的逻辑序列，通常是初始请求以及它的重定向请求。 -
（2）：是执行下一个拦截器，按顺序调用那就是 BridgeInterceptor。
-
（3）（4）：发生 Route 或 IO 异常，则进行重试，我们看看重试的相关方法：
/*RetryAndFollowUpInterceptor*/
private boolean recover(IOException e, StreamAllocation streamAllocation,
boolean requestSendStarted, Request userRequest) {
streamAllocation.streamFailed(e);
// 客户端配置了出错不再重试
if (!client.retryOnConnectionFailure()) return false;
// 无法再次发送 request body
if (requestSendStarted && userRequest.body() instanceof UnrepeatableRequestBody) return false;
// 发生 isRecoverable() 方法中出现的异常
if (!isRecoverable(e, requestSendStarted)) return false;
// 没有更多的路线可供尝试
if (!streamAllocation.hasMoreRoutes()) return false;
// For failure recovery, use the same route selector with a new connection.
return true;
}
private boolean isRecoverable(IOException e, boolean requestSendStarted) {
// 协议异常
if (e instanceof ProtocolException) {
return false;
}
// 中断异常
if (e instanceof InterruptedIOException) {
return e instanceof SocketTimeoutException && !requestSendStarted;
}
// SSL握手异常
if (e instanceof SSLHandshakeException) {
// If the problem was a CertificateException from the X509TrustManager,
// do not retry.
if (e.getCause() instanceof CertificateException) {
return false;
}
}
// SSL握手未授权异常
if (e instanceof SSLPeerUnverifiedException) {
// e.g. a certificate pinning error.
return false;
}
// An example of one we might want to retry with a different route is a problem connecting to a
// proxy and would manifest as a standard IOException. Unless it is one we know we should not
// retry, we return true and try a new route.
return true;
}
可以看到尝试进行重试的时候，如果出现以下情况则不会重试：
-
客户端配置了出错不再重试
-
无法再次发送 request body
-
发生 ProtocolException（协议异常）、InterruptedIOException（中断异常）、SSLHandshakeException（SSL握手异常）、SSLPeerUnverifiedException（SSL握手未授权异常）中的任意一个异常
-
没有更多的路线可供尝试
-
（5）（6）：检查是否需要重定向，如果不需要则返回之前的 response，需要则进行重定向，也就是继续循环请求重试。是否需要重定向主要根据响应码来决定，具体可以去看看 followUpRequest() 方法，这里就不贴代码了。
ps：如果你想拿重定向的域名来跟一遍源码中重定向的流程，那么你可以试试郭霖的域名（http://guolin.tech）， 该域名会重定向到他的 csdn 博客（https://blog.csdn.net/guolin_blog）， 走一遍流程会让你对源码中重定向的原理有更深的理解。
2.4.3.2 桥拦截器（BridgeInterceptor）
该拦截器相当于一个桥梁，首先将用户的请求转换为发给服务器的请求，然后使用该请求访问网络，最后将服务器返回的响应转换为用户可用的响应。
我们看看该拦截器中的 intercept() 方法：
/*BridgeInterceptor*/
@Override public Response intercept(Chain chain) throws IOException {
Request userRequest = chain.request();
Request.Builder requestBuilder = userRequest.newBuilder();
//（1）将用户的请求转换为发给服务器的请求-start
RequestBody body = userRequest.body();
if (body != null) {
MediaType contentType = body.contentType();
if (contentType != null) {
requestBuilder.header("Content-Type", contentType.toString());
}
long contentLength = body.contentLength();
if (contentLength != -1) {
requestBuilder.header("Content-Length", Long.toString(contentLength));
requestBuilder.removeHeader("Transfer-Encoding");
} else {
requestBuilder.header("Transfer-Encoding", "chunked");
requestBuilder.removeHeader("Content-Length");
}
}
if (userRequest.header("Host") == null) {
requestBuilder.header("Host", hostHeader(userRequest.url(), false));
}
if (userRequest.header("Connection") == null) {
requestBuilder.header("Connection", "Keep-Alive");
}
// 如果我们在创建 Request 的时候添加了 "Accept-Encoding: gzip" 请求头，那么要自己负责解压缩传输流。
boolean transparentGzip = false;
if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
// 默认是 gzip 压缩
transparentGzip = true;
requestBuilder.header("Accept-Encoding", "gzip");
}
List<Cookie> cookies = cookieJar.loadForRequest(userRequest.url());
if (!cookies.isEmpty()) {
requestBuilder.header("Cookie", cookieHeader(cookies));
}
if (userRequest.header("User-Agent") == null) {
requestBuilder.header("User-Agent", Version.userAgent());
}
//（1）将用户的请求转换为发给服务器的请求-end
//（2）执行下一个拦截器进行网络请求
Response networkResponse = chain.proceed(requestBuilder.build());
//（3）将服务器返回的响应转换为用户可用的响应-start
// 解析服务器返回的 header
HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());
Response.Builder responseBuilder = networkResponse.newBuilder()
.request(userRequest);
// gzip 解压
if (transparentGzip
&& "gzip".equalsIgnoreCase(networkResponse.header("Content-Encoding"))
&& HttpHeaders.hasBody(networkResponse)) {
GzipSource responseBody = new GzipSource(networkResponse.body().source());
Headers strippedHeaders = networkResponse.headers().newBuilder()
.removeAll("Content-Encoding")
.removeAll("Content-Length")
.build();
responseBuilder.headers(strippedHeaders);
String contentType = networkResponse.header("Content-Type");
responseBuilder.body(new RealResponseBody(contentType, -1L, Okio.buffer(responseBody)));
}
//（3）将服务器返回的响应转换为用户可用的响应-end
return responseBuilder.build();
}
根据我标记的关注点大概就是：
- （1）：将用户的请求转换为发给服务器的请求。主要是添加一些默认的请求头，例如 Content-Type、Content-Length、Transfer-Encoding、Host、Connection。因为我们在创建 Request 的时候可以不添加任何请求头，如果这里不加上一些默认的请求头是无法完成请求的。
- （2）：执行下一个拦截器进行网络请求。
- （3）：将服务器返回的响应转换为用户可用的响应。主要是解析服务器返回的 header，进行 gzip 解压。
2.4.3.3 缓存拦截器（CacheInterceptor）
该拦截器主要用来实现缓存的读取和存储，即进行网络请求的时候执行到缓存拦截器会先判断是否有缓存，如果有会直接返回缓存，没有则会执行后面的拦截器继续请求网络，请求成功会将请求到的数据缓存起来。
我们看看该拦截器中的 intercept() 方法：
/*CacheInterceptor*/
@Override public Response intercept(Chain chain) throws IOException {
//（1）通过 Request 得到缓存
Response cacheCandidate = cache != null
? cache.get(chain.request())
: null;
long now = System.currentTimeMillis();
//（2）通过缓存策略获取是使用缓存还是使用网络请求，或者 2 者同时使用或都不使用
CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();
Request networkRequest = strategy.networkRequest;
Response cacheResponse = strategy.cacheResponse;
if (cache != null) {
cache.trackResponse(strategy);
}
// 有缓存，但是策略中不使用缓存，需要释放资源
if (cacheCandidate != null && cacheResponse == null) {
closeQuietly(cacheCandidate.body());
}
// （3）如果策略中不使用网络请求，也不使用缓存，那么直接返回失败
if (networkRequest == null && cacheResponse == null) {
return new Response.Builder()
.request(chain.request())
.protocol(Protocol.HTTP_1_1)
.code(504)
.message("Unsatisfiable Request (only-if-cached)")
.body(Util.EMPTY_RESPONSE)
.sentRequestAtMillis(-1L)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
}
//（4）如果策略中不使用网络请求，执行到这里说明是使用缓存的，则直接返回缓存
if (networkRequest == null) {
return cacheResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.build();
}
Response networkResponse = null;
try {
//（5）执行下一个拦截器进行网络请求
networkResponse = chain.proceed(networkRequest);
} finally {
// 如果发生 IO 或者其他崩溃，为了不泄漏缓存体，需要释放资源
if (networkResponse == null && cacheCandidate != null) {
closeQuietly(cacheCandidate.body());
}
}
//（6）如果策略中使用缓存，并且响应码为 304，则返回缓存
if (cacheResponse != null) {
if (networkResponse.code() == HTTP_NOT_MODIFIED) {
Response response = cacheResponse.newBuilder()
.headers(combine(cacheResponse.headers(), networkResponse.headers()))
.sentRequestAtMillis(networkResponse.sentRequestAtMillis())
.receivedResponseAtMillis(networkResponse.receivedResponseAtMillis())
.cacheResponse(stripBody(cacheResponse))
.networkResponse(stripBody(networkResponse))
.build();
networkResponse.body().close();
cache.trackConditionalCacheHit();
// 更新缓存
cache.update(cacheResponse, response);
return response;
} else {
closeQuietly(cacheResponse.body());
}
}
Response response = networkResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.networkResponse(stripBody(networkResponse))
.build();
if (cache != null) {
if (HttpHeaders.hasBody(response) && CacheStrategy.isCacheable(response, networkRequest)) {
//（7）将请求返回的结果存进缓存
CacheRequest cacheRequest = cache.put(response);
return cacheWritingResponse(cacheRequest, response);
}
if (HttpMethod.invalidatesCache(networkRequest.method())) {
try {
cache.remove(networkRequest);
} catch (IOException ignored) {
// The cache cannot be written.
}
}
}
return response;
}
根据我标记的关注点大概流程就是：
- （1）：通过 Request 得到缓存。这里的 cache 是 InternalCache，但是因为 InternalCache 是一个接口，而且只有一个实现类 Cache，所以 cache 其实就是 Cache。进入 Cache 可以发现它底层使用的是 DiskLruCache 缓存机制，也就是使用 “最近最少使用” 算法将数据缓存到磁盘内。
- （2）：通过缓存策略获取是使用缓存还是使用网络请求，或者 2 者同时使用或都不使用。networkRequest 为 null 表示不使用网络请求，cacheResponse 为 null 表示不使用缓存。
- （3）：如果策略中不使用网络请求，也不使用缓存，那么直接返回失败。这样就直接停止了后面拦截器的执行，结束了整个请求。
- （4）：如果策略中不使用网络请求，执行到这里说明是使用缓存的，则直接返回缓存。这样就直接停止了后面拦截器的执行，结束了整个请求。
- （5）：执行到这里，说明需要从网络获取数据，则会继续执行下一个拦截器进行网络请求。
- （6）：如果策略中使用缓存，并且响应码为 304，则返回缓存，并且更新缓存。
- （7）：最后将请求返回的结果进行缓存。
2.4.3.4 连接拦截器（ConnectInterceptor）
该拦截器主要用来打开与目标服务器的连接，然后继续执行下一个拦截器。
我们看看该拦截器中的 intercept() 方法：
/*ConnectInterceptor*/
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Request request = realChain.request();
//（1）获取 StreamAllocation
StreamAllocation streamAllocation = realChain.streamAllocation();
// We need the network to satisfy this request. Possibly for validating a conditional GET.
boolean doExtensiveHealthChecks = !request.method().equals("GET");
//（2）创建 HttpCodec
HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
//（3）获取 RealConnection
RealConnection connection = streamAllocation.connection();
//（4）执行下一个拦截器
return realChain.proceed(request, streamAllocation, httpCodec, connection);
}
根据我标记的关注点大概流程就是：
- （1）：获取 StreamAllocation，这里获取的其实就是第一个拦截器 RetryAndFollowUpInterceptor 中创建的。
- （2）：创建 HttpCodec，是通过 StreamAllocation 的 newStream() 方法获取的，我们看下 newStream() 方法：
/*StreamAllocation*/
public HttpCodec newStream(
OkHttpClient client, Interceptor.Chain chain, boolean doExtensiveHealthChecks) {
int connectTimeout = chain.connectTimeoutMillis();
int readTimeout = chain.readTimeoutMillis();
int writeTimeout = chain.writeTimeoutMillis();
int pingIntervalMillis = client.pingIntervalMillis();
boolean connectionRetryEnabled = client.retryOnConnectionFailure();
try {
//（5）寻找可用的连接
RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
//（6）通过这个可用的连接创建 HttpCodec
HttpCodec resultCodec = resultConnection.newCodec(client, chain, this);
synchronized (connectionPool) {
codec = resultCodec;
return resultCodec;
}
} catch (IOException e) {
throw new RouteException(e);
}
}
我们看下关注点（5）中的 findHealthyConnection() 方法：
/*StreamAllocation*/
private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
int writeTimeout, int pingIntervalMillis, boolean connectionRetryEnabled,
boolean doExtensiveHealthChecks) throws IOException {
while (true) {
//（7）寻找一个连接
RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
pingIntervalMillis, connectionRetryEnabled);
// 如果这是一个全新的连接，则不需要后面的健康检查，而是在这里直接返回连接
synchronized (connectionPool) {
if (candidate.successCount == 0) {
return candidate;
}
}
// 如果不健康，则禁止创建新流，并且继续循环查找可用的链接
if (!candidate.isHealthy(doExtensiveHealthChecks)) {
noNewStreams();
continue;
}
return candidate;
}
}
可以看到，findHealthyConnection() 方法中又通过 findConnection() 方法去寻找，看下这个方法：
/*StreamAllocation*/
private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException {
boolean foundPooledConnection = false;
RealConnection result = null;
Route selectedRoute = null;
Connection releasedConnection;
Socket toClose;
synchronized (connectionPool) {
if (released) throw new IllegalStateException("released");
if (codec != null) throw new IllegalStateException("codec != null");
if (canceled) throw new IOException("Canceled");
//（8）start
// 尝试使用已分配的连接
releasedConnection = this.connection;
toClose = releaseIfNoNewStreams();
if (this.connection != null) {
// 已经分配的连接，并且是可用的，则将该已分配的连接赋值为可用的连接
result = this.connection;
releasedConnection = null;
}
//（8）end
if (!reportedAcquired) {
// 如果这个连接从未标记过已获取，那么请不要标记为为已发布
releasedConnection = null;
}
//（9）start 尝试从连接池中获取连接
if (result == null) {
Internal.instance.get(connectionPool, address, this, null);
if (connection != null) {
foundPooledConnection = true;
result = connection;
} else {
selectedRoute = route;
}
}
}
closeQuietly(toClose);
if (releasedConnection != null) {
eventListener.connectionReleased(call, releasedConnection);
}
if (foundPooledConnection) {
eventListener.connectionAcquired(call, result);
}
if (result != null) {
// 如果找到一个可用的连接，那么直接返回
return result;
}
boolean newRouteSelection = false;
if (selectedRoute == null && (routeSelection == null || !routeSelection.hasNext())) {
newRouteSelection = true;
routeSelection = routeSelector.next();
}
synchronized (connectionPool) {
if (canceled) throw new IOException("Canceled");
//（10）根据不同的路由再次从连接池中获取可用的连接
if (newRouteSelection) {
List<Route> routes = routeSelection.getAll();
for (int i = 0, size = routes.size(); i < size; i++) {
Route route = routes.get(i);
Internal.instance.get(connectionPool, address, this, route);
if (connection != null) {
foundPooledConnection = true;
result = connection;
this.route = route;
break;
}
}
}
//（11）还是没有找到可用的连接，那么重新创建一个新的连接
if (!foundPooledConnection) {
if (selectedRoute == null) {
selectedRoute = routeSelection.next();
}
route = selectedRoute;
refusedStreamCount = 0;
result = new RealConnection(connectionPool, selectedRoute);
acquire(result, false);
}
}
// 如果在第二次找到了可用的连接，则直接返回
if (foundPooledConnection) {
eventListener.connectionAcquired(call, result);
return result;
}
//（12）进行 TCP 和 TLS 握手
result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,
connectionRetryEnabled, call, eventListener);
routeDatabase().connected(result.route());
Socket socket = null;
synchronized (connectionPool) {
reportedAcquired = true;
//（13）将新创建的连接放进连接池中
Internal.instance.put(connectionPool, result);
// 如果同时创建了到同一地址的另一个多路复用连接，则释放这个连接并获取那个多路复用连接。
if (result.isMultiplexed()) {
socket = Internal.instance.deduplicate(connectionPool, address, this);
result = connection;
}
}
closeQuietly(socket);
eventListener.connectionAcquired(call, result);
return result;
}
通过上面的代码分析，findConnection() 方法大概流程就是：
- （8）：判断当前连接是否可用，可用则进行赋值，在后面直接返回
- （9）：如果当前连接不可用，那么尝试从连接池中获取可用连接
- （10）：如果连接池中找不到可用的连接，那么切换不同的路由再次从连接池中获取可用的连接
- （11）：还是没有找到可用的连接，那么只能重新创建一个新的连接
- （12）：进行 TCP 和 TLS 握手
- （13）：最后将新创建的连接放进连接池中
可以看到，关注点（9）（13）分别是从连接池中取出连接和存入连接到连接池，分别调用的是 Internal.instance.get() 与 Internal.instance.put（）。
我们看下 get() 方法是怎样的，点击 get() 方法进去，发现 Internal 是一个抽象类，它有一个静态的实例，在 OkHttpClient 的静态代码快中被初始化：
/*OkHttpClient*/
static {
Internal.instance = new Internal() {
// 省略部分代码...
@Override public RealConnection get(ConnectionPool pool, Address address,
StreamAllocation streamAllocation, Route route) {
return pool.get(address, streamAllocation, route);
}
// 省略部分代码...
}
可以看到 Internal 的 get() 方法中调用的是 ConnectionPool（连接池）的 get() 方法，所以可以肯定这个连接池就是用来操作这些连接的，内部具体怎么操作我们放到后面去讲，这里只需要知道它可以用来存取连接就可以了。
关注点（12）其实就是与服务器建立连接的核心代码，我们看下这个方法：
/*RealConnection*/
public void connect(int connectTimeout, int readTimeout, int writeTimeout,
int pingIntervalMillis, boolean connectionRetryEnabled, Call call,
EventListener eventListener) {
if (protocol != null) throw new IllegalStateException("already connected");
/*线路选择*/
RouteException routeException = null;
List<ConnectionSpec> connectionSpecs = route.address().connectionSpecs();
ConnectionSpecSelector connectionSpecSelector = new ConnectionSpecSelector(connectionSpecs);
if (route.address().sslSocketFactory() == null) {
if (!connectionSpecs.contains(ConnectionSpec.CLEARTEXT)) {
throw new RouteException(new UnknownServiceException(
"CLEARTEXT communication not enabled for client"));
}
String host = route.address().url().host();
if (!Platform.get().isCleartextTrafficPermitted(host)) {
throw new RouteException(new UnknownServiceException(
"CLEARTEXT communication to " + host + " not permitted by network security policy"));
}
} else {
if (route.address().protocols().contains(Protocol.H2_PRIOR_KNOWLEDGE)) {
throw new RouteException(new UnknownServiceException(
"H2_PRIOR_KNOWLEDGE cannot be used with HTTPS"));
}
}
while (true) {
try {
//（14）如果需要隧道连接，则进行隧道连接
if (route.requiresTunnel()) {
connectTunnel(connectTimeout, readTimeout, writeTimeout, call, eventListener);
if (rawSocket == null) {
// We were unable to connect the tunnel but properly closed down our resources.
break;
}
} else {
//（15）不需要隧道连接，则直接进行 socket 连接
connectSocket(connectTimeout, readTimeout, call, eventListener);
}
// 建立协议
establishProtocol(connectionSpecSelector, pingIntervalMillis, call, eventListener);
// 连接结束
eventListener.connectEnd(call, route.socketAddress(), route.proxy(), protocol);
break;
} catch (IOException e) {
closeQuietly(socket);
closeQuietly(rawSocket);
socket = null;
rawSocket = null;
source = null;
sink = null;
handshake = null;
protocol = null;
http2Connection = null;
// 连接失败
eventListener.connectFailed(call, route.socketAddress(), route.proxy(), null, e);
if (routeException == null) {
routeException = new RouteException(e);
} else {
routeException.addConnectException(e);
}
if (!connectionRetryEnabled || !connectionSpecSelector.connectionFailed(e)) {
throw routeException;
}
}
}
if (route.requiresTunnel() && rawSocket == null) {
ProtocolException exception = new ProtocolException("Too many tunnel connections attempted: "
+ MAX_TUNNEL_ATTEMPTS);
throw new RouteException(exception);
}
if (http2Connection != null) {
synchronized (connectionPool) {
allocationLimit = http2Connection.maxConcurrentStreams();
}
}
}
关注点（14）（15）最终都会调用 connectSocket() 方法：
/*RealConnection*/
private void connectSocket(int connectTimeout, int readTimeout, Call call,
EventListener eventListener) throws IOException {
Proxy proxy = route.proxy();
Address address = route.address();
// 创建 socket
rawSocket = proxy.type() == Proxy.Type.DIRECT || proxy.type() == Proxy.Type.HTTP
? address.socketFactory().createSocket()
: new Socket(proxy);
eventListener.connectStart(call, route.socketAddress(), proxy);
// 设置 socket 超时时间
rawSocket.setSoTimeout(readTimeout);
try {
//（16）进行 socket 连接
Platform.get().connectSocket(rawSocket, route.socketAddress(), connectTimeout);
} catch (ConnectException e) {
ConnectException ce = new ConnectException("Failed to connect to " + route.socketAddress());
ce.initCause(e);
throw ce;
}
try {
source = Okio.buffer(Okio.source(rawSocket));
sink = Okio.buffer(Okio.sink(rawSocket));
} catch (NullPointerException npe) {
if (NPE_THROW_WITH_NULL.equals(npe.getMessage())) {
throw new IOException(npe);
}
}
}
可以看到 okhttp 底层是通过 socket 进行连接的。
看完关注点（5）中的 findHealthyConnection() 方法，我们继续回去看关注点（6）的方法：
/*StreamAllocation*/
public HttpCodec newCodec(OkHttpClient client, Interceptor.Chain chain,
StreamAllocation streamAllocation) throws SocketException {
if (http2Connection != null) {
return new Http2Codec(client, chain, streamAllocation, http2Connection);
} else {
socket.setSoTimeout(chain.readTimeoutMillis());
source.timeout().timeout(chain.readTimeoutMillis(), MILLISECONDS);
sink.timeout().timeout(chain.writeTimeoutMillis(), MILLISECONDS);
return new Http1Codec(client, streamAllocation, source, sink);
}
}
该方法是创建 HttpCodec，HttpCodec 的作用主要是进行 HTTP 请求和响应的编码与解码操作。它有两个实现类，分别是 Http1Codec 与 Http2Codec，这里主要判断如果是 HTTP/2，则创建 Http2Codec，否则创建 Http1Codec。
-
（3）：继续回去看关注点（3），点击 connection() 方法进去发现，这里获取的 RealConnection 其实就是关注点（7） findConnection()
方法中从连接池中取出连接或重新创建的连接。 -
（4）：关注点（4）则拿到连接后继续执行下一个拦截器。
2.4.3.5 服务器请求拦截器（CallServerInterceptor）
该拦截器主要用来向服务器发起请求并获取数据，它是责任链中的最后一个拦截器，获取到服务器的数据后会直接返回给上一个拦截器。
我们看看该拦截器中的 intercept() 方法：
/*CallServerInterceptor*/
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
// 获取 ConnectInterceptor 中创建的 HttpCodec
HttpCodec httpCodec = realChain.httpStream();
// 获取 RetryAndFollowUpInterceptor 中创建的 StreamAllocation
StreamAllocation streamAllocation = realChain.streamAllocation();
// 获取 ConnectInterceptor 中新创建或者从连接池中拿到的 RealConnection
RealConnection connection = (RealConnection) realChain.connection();
Request request = realChain.request();
long sentRequestMillis = System.currentTimeMillis();
realChain.eventListener().requestHeadersStart(realChain.call());
//（1）写入请求头
httpCodec.writeRequestHeaders(request);
realChain.eventListener().requestHeadersEnd(realChain.call(), request);
Response.Builder responseBuilder = null;
if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
// If there's a "Expect: 100-continue" header on the request, wait for a "HTTP/1.1 100
// Continue" response before transmitting the request body. If we don't get that, return
// what we did get (such as a 4xx response) without ever transmitting the request body.
if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
httpCodec.flushRequest();
realChain.eventListener().responseHeadersStart(realChain.call());
responseBuilder = httpCodec.readResponseHeaders(true);
}
//（2）写入请求体
if (responseBuilder == null) {
// Write the request body if the "Expect: 100-continue" expectation was met.
realChain.eventListener().requestBodyStart(realChain.call());
long contentLength = request.body().contentLength();
CountingSink requestBodyOut =
new CountingSink(httpCodec.createRequestBody(request, contentLength));
BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
request.body().writeTo(bufferedRequestBody);
bufferedRequestBody.close();
realChain.eventListener()
.requestBodyEnd(realChain.call(), requestBodyOut.successfulCount);
} else if (!connection.isMultiplexed()) {
// If the "Expect: 100-continue" expectation wasn't met, prevent the HTTP/1 connection
// from being reused. Otherwise we're still obligated to transmit the request body to
// leave the connection in a consistent state.
streamAllocation.noNewStreams();
}
}
httpCodec.finishRequest();
if (responseBuilder == null) {
realChain.eventListener().responseHeadersStart(realChain.call());
//（3）读取响应头
responseBuilder = httpCodec.readResponseHeaders(false);
}
Response response = responseBuilder
.request(request)
.handshake(streamAllocation.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
int code = response.code();
if (code == 100) {
// server sent a 100-continue even though we did not request one.
// try again to read the actual response
responseBuilder = httpCodec.readResponseHeaders(false);
response = responseBuilder
.request(request)
.handshake(streamAllocation.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
code = response.code();
}
realChain.eventListener()
.responseHeadersEnd(realChain.call(), response);
//（4）读取响应体
if (forWebSocket && code == 101) {
// Connection is upgrading, but we need to ensure interceptors see a non-null response body.
response = response.newBuilder()
.body(Util.EMPTY_RESPONSE)
.build();
} else {
response = response.newBuilder()
.body(httpCodec.openResponseBody(response))
.build();
}
if ("close".equalsIgnoreCase(response.request().header("Connection"))
|| "close".equalsIgnoreCase(response.header("Connection"))) {
streamAllocation.noNewStreams();
}
if ((code == 204 || code == 205) && response.body().contentLength() > 0) {
throw new ProtocolException(
"HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
}
return response;
}
可以看到，这个拦截器还是比较简单的，上一个拦截器 ConnectInterceptor 已经连接到服务器了并创建了 HttpCodec 对象，HttpCodec 对象封装了 okio 提供的输出流（BufferedSink）与输入流（BufferedSource），所以这里就主要通过 HttpCodec 对象与服务器进行读写操作。例如写入请求头与请求体，读取响应头与响应体。
2.4.4 ConnectionPool（连接池）
-
简介
连接池是用来管理 HTTP 和 HTTP / 2 连接的复用，以减少网络延迟。从上面我们阅读 findConnection() 方法源码也可以得出，即如果从连接池中找到了可用的连接，那么就不用重新创建新的连接，也省去了 TCP 和 TLS 握手。 -
ConnectionPool 类中的主要常量
/*ConnectionPool*/
// 线程池，用于清除过期的连接
private static final Executor executor = new ThreadPoolExecutor(0 /* corePoolSize */,
Integer.MAX_VALUE /* maximumPoolSize */, 60L /* keepAliveTime */, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp ConnectionPool", true));
// 最大允许空闲的连接数量
private final int maxIdleConnections;
// 连接的存活时间
private final long keepAliveDurationNs;
// 清理任务，用来清理无效的连接
private final Runnable cleanupRunnable = new Runnable() {
//...
};
// 用来记录连接的双端队列
private final Deque<RealConnection> connections = new ArrayDeque<>();
- 构造函数
/*ConnectionPool*/
public ConnectionPool() {
this(5, 5, TimeUnit.MINUTES);
}
public ConnectionPool(int maxIdleConnections, long keepAliveDuration, TimeUnit timeUnit) {
this.maxIdleConnections = maxIdleConnections;
this.keepAliveDurationNs = timeUnit.toNanos(keepAliveDuration);
// Put a floor on the keep alive duration, otherwise cleanup will spin loop.
if (keepAliveDuration <= 0) {
throw new IllegalArgumentException("keepAliveDuration <= 0: " + keepAliveDuration);
}
}
可以看到构造函数设置了默认的最大允许空闲的连接数量为 5 个，连接的存活时间为 5 分钟。
- 主要函数
这里主要讲下前面连接拦截器中用到的 get()、put() 方法。
get() 方法：
/*ConnectionPool*/
@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;
}
该方法是从连接池中获取可复用的连接，这里的逻辑是遍历记录连接的双端队列，取出可复用的连接。
put() 方法：
/*ConnectionPool*/
void put(RealConnection connection) {
assert (Thread.holdsLock(this));
if (!cleanupRunning) {
cleanupRunning = true;
// 执行清理任务
executor.execute(cleanupRunnable);
}
// 将新创建的连接添加进记录连接的双端队列中
connections.add(connection);
}
该方法是将新创建的连接放进连接池中，这里的逻辑是先清理无效的连接，然后再将新创建的连接添加进记录连接的双端队列中。
我们先看下清理任务：
/*ConnectionPool*/
private final Runnable cleanupRunnable = new Runnable() {
@Override public void run() {
while (true) {
// 清理无效连接
long waitNanos = cleanup(System.nanoTime());
if (waitNanos == -1) return;
if (waitNanos > 0) {
long waitMillis = waitNanos / 1000000L;
waitNanos -= (waitMillis * 1000000L);
synchronized (ConnectionPool.this) {
try {
ConnectionPool.this.wait(waitMillis, (int) waitNanos);
} catch (InterruptedException ignored) {
}
}
}
}
}
};
这是一个阻塞的清理任务，并且通过无限循环来清理。这里首先调用 cleanup() 方法清理无效连接，并返回下次需要清理的间隔时间，然后调用 wait() 方法进行等待以释放锁与时间片，当等待时间到了后，再次循环清理。
我们看下 cleanup() 方法：
/*ConnectionPool*/
long cleanup(long now) {
int inUseConnectionCount = 0;
int idleConnectionCount = 0;
RealConnection longestIdleConnection = null;
long longestIdleDurationNs = Long.MIN_VALUE;
// 遍历连接，找出无效连接进行清理
synchronized (this) {
for (Iterator<RealConnection> i = connections.iterator(); i.hasNext(); ) {
RealConnection connection = i.next();
//（1）查询此连接的 StreamAllocation 的引用数量，大于 0 则 inUseConnectionCount 加 1，否则 idleConnectionCount 加 1。
if (pruneAndGetAllocationCount(connection, now) > 0) {
inUseConnectionCount++;
continue;
}
idleConnectionCount++;
// 标记空闲连接
long idleDurationNs = now - connection.idleAtNanos;
if (idleDurationNs > longestIdleDurationNs) {
longestIdleDurationNs = idleDurationNs;
longestIdleConnection = connection;
}
}
if (longestIdleDurationNs >= this.keepAliveDurationNs
|| idleConnectionCount > this.maxIdleConnections) {
// 如果连接存活时间大于等于 5 分钟，或者空闲的连接数量大于 5 个，则将该链接从队列中移除
connections.remove(longestIdleConnection);
} else if (idleConnectionCount > 0) {
// 如果空闲的连接数量大于 0，返回此连接即将到期的时间
return keepAliveDurationNs - longestIdleDurationNs;
} else if (inUseConnectionCount > 0) {
// 如果没有空闲连接，则返回 5 分钟，也就是下次需要清理的间隔时间为 5 分钟
return keepAliveDurationNs;
} else {
// 没有任何连接，则跳出循环
cleanupRunning = false;
return -1;
}
}
closeQuietly(longestIdleConnection.socket());
// 马上进行下一次清理
return 0;
}
可以看到，这里主要通过判断连接存活时间是否大于等于 5 分钟，或者空闲的连接数量是否大于 5 个来进行连接的清理。连接是否空闲是通过关注点（1）中的 pruneAndGetAllocationCount() 方法来判断的，我们看下这个方法：
/*ConnectionPool*/
private int pruneAndGetAllocationCount(RealConnection connection, long now) {
// 获得 allocations 的弱引用列表
List<Reference<StreamAllocation>> references = connection.allocations;
// 遍历 allocations 的弱引用列表
for (int i = 0; i < references.size(); ) {
Reference<StreamAllocation> reference = references.get(i);
// 说明 StreamAllocation 被使用，则继续下一次循环
if (reference.get() != null) {
i++;
continue;
}
// We've discovered a leaked allocation. This is an application bug.
StreamAllocation.StreamAllocationReference streamAllocRef =
(StreamAllocation.StreamAllocationReference) reference;
String message = "A connection to " + connection.route().address().url()
+ " was leaked. Did you forget to close a response body?";
Platform.get().logCloseableLeak(message, streamAllocRef.callStackTrace);
// 说明 StreamAllocation 没有被使用，则从列表中移除
references.remove(i);
connection.noNewStreams = true;
// 列表为空，说明都被移除了，这个时候返回 allocationCount 为 0，表示该连接是空闲的。
if (references.isEmpty()) {
connection.idleAtNanos = now - keepAliveDurationNs;
return 0;
}
}
// 列表不为空，返回列表的大小，大于 0 表示该连接是在使用的。
return references.size();
}
该方法比较简单，主要是遍历 allocations 的弱引用列表，如果 StreamAllocation 没有被使用，则从列表中移除，最后返回该列表的大小，通过该大小即可判断是否是空闲连接，小于等于 0 才是空闲连接。
2.5 （5）取出相应的数据
String data = response.body().string();
在第（4）步同步请求或者异步请求执行完都会返回 Response，这个就是最终返回的数据，可以通过它获取到 code、message、header、body 等。
这里讲下 body，点击 body() 进去是这样的：
/*Response*/
public @Nullable ResponseBody body() {
return body;
}
可以看到这里的 body 就是 ResponseBody，它是一个抽象类，不能被实例化，一般用它的子类 RealResponseBody 进行实例化。它是在前面讲的 “2.4.3.5 服务器请求拦截器（CallServerInterceptor）” 小节中赋值的：
/*CallServerInterceptor*/
if (forWebSocket && code == 101) {
response = response.newBuilder()
.body(Util.EMPTY_RESPONSE)
.build();
} else {
// openResponseBody() 方法中创建了 RealResponseBody 对象返回
response = response.newBuilder()
.body(httpCodec.openResponseBody(response))
.build();
}
如果有缓存则会在缓存拦截器（CacheInterceptor）中赋值。
ResponseBody 中常用的方法有如下几种：
/*ResponseBody*/
public final String string() throws IOException {
BufferedSource source = source();
try {
Charset charset = Util.bomAwareCharset(source, charset());
return source.readString(charset);
} finally {
Util.closeQuietly(source);
}
}
public final InputStream byteStream() {
return source().inputStream();
}
public final byte[] bytes() throws IOException {
long contentLength = contentLength();
if (contentLength > Integer.MAX_VALUE) {
throw new IOException("Cannot buffer entire body for content length: " + contentLength);
}
BufferedSource source = source();
byte[] bytes;
try {
bytes = source.readByteArray();
} finally {
Util.closeQuietly(source);
}
if (contentLength != -1 && contentLength != bytes.length) {
throw new IOException("Content-Length ("
+ contentLength
+ ") and stream length ("
+ bytes.length
+ ") disagree");
}
return bytes;
}
可以看到，这三个方法内部都调用了 source() 来获取 BufferedSource，BufferedSource 就是 okio 提供的输入流，拿到输入流就可以将 body 数据转换为你需要的类型。例如：
希望返回 String，则调用 response.body().string()，适用于不超过 1 MB 的数据。
希望返回输入流，则调用 response.body().byteStream()，适用于超过 1 MB 的数据，例如下载文件。
希望返回二进制字节数组，则调用 response.body().bytes()。
需要注意的是，response.body().string() 只能调用一次，否则会抛出如下异常：
W/System.err: java.lang.IllegalStateException: closed
W/System.err: at okio.RealBufferedSource.rangeEquals(RealBufferedSource.java:408)
W/System.err: at okio.RealBufferedSource.rangeEquals(RealBufferedSource.java:402)
W/System.err: at okhttp3.internal.Util.bomAwareCharset(Util.java:469)
W/System.err: at okhttp3.ResponseBody.string(ResponseBody.java:175)
根据报错日志可以看到，是在 RealBufferedSource 类的 408 行报的错，我们跳转过去看看：
/*RealBufferedSource*/
@Override
public boolean rangeEquals(long offset, ByteString bytes, int bytesOffset, int byteCount)
throws IOException {
if (closed) throw new IllegalStateException("closed");
//...
}
可以看到，这里做了个判断，closed 为 true 就抛出该异常，继续跟踪 closed 赋值的地方：
/*RealBufferedSource*/
@Override public void close() throws IOException {
if (closed) return;
closed = true;
source.close();
buffer.clear();
}
可以看到，closed 唯一赋值的地方在 close() 方法中，而该方法正是 string() 方法中的 Util.closeQuietly(source); 调用的：
/*ResponseBody*/
public static void closeQuietly(Closeable closeable) {
if (closeable != null) {
try {
closeable.close();
} catch (RuntimeException rethrown) {
throw rethrown;
} catch (Exception ignored) {
}
}
}
到这里我们就明白了为什么 response.body().string() 只能调用一次的原因，即 string() 方法中获取到 String后又调用了 Util.closeQuietly(source) 方法关闭了输入流，并且标记 closed 为 true，然后第二次调用 string() 方法的时候会在 RealBufferedSource.rangeEquals() 方法进行判断，为 true 就抛出异常。
这样设计的原因是服务器返回的 body 可能会很大，所以 OkHttp 不会将其存储在内存中，只有当你需要的时候才去获取它，如果没有新的请求则无法获取 2 次。
三、总结
看完源码，发现 OkHttp 是一个设计得非常优秀的框架。该框架运用了很多设计模式，例如建造者模式、责任链模式等等。知道了 OkHttp 的核心是拦截器，这里采用的就是责任链模式，每个拦截器负责相应的功能，发起请求的时候由上往下依次执行每个拦截器，响应的数据则层层往上传递。
参考资料：
