之前熟悉了OkHttp大概的調(diào)用 這次學(xué)習(xí)一下各攔截器的作用
貼一貼攔截器調(diào)用入口方法,方便梳理
//RealCall.Java
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()));//橋攔截器？主要是用來(lái)加Header信息
interceptors.add(new CacheInterceptor(client.internalCache()));//緩存攔截器
interceptors.add(new ConnectInterceptor(client));//連接攔截器,用于建立連接
if (!forWebSocket) {
interceptors.addAll(client.networkInterceptors());//僅在建立WebSocket時(shí)才會(huì)加的攔截器
}
interceptors.add(new CallServerInterceptor(forWebSocket));//發(fā)送信息攔截器(之前的連接攔截器負(fù)責(zé)建立連接但是啥也沒(méi)干,這個(gè)攔截器負(fù)責(zé)給服務(wù)器發(fā)消息并接收返回)
Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
originalRequest, this, eventListener, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
return chain.proceed(originalRequest);//開(kāi)啟鏈?zhǔn)秸{(diào)用
}
那么接下來(lái)就是依次探討一下RetryAndFollowUpInterceptor,BridgeInterceptor,CacheInterceptor,ConnectInterceptor,CallServerInterceptor了
重連及重定向攔截器(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();
//申請(qǐng)網(wǎng)絡(luò)連接的關(guān)鍵類,具體在ConnectInterceptor會(huì)進(jìn)一步調(diào)用
StreamAllocation streamAllocation = new StreamAllocation(client.connectionPool(),
createAddress(request.url()), call, eventListener, callStackTrace);
this.streamAllocation = streamAllocation;
int followUpCount = 0;
Response priorResponse = null;
//維護(hù)一個(gè)死循環(huán),該攔截器能實(shí)現(xiàn)重連和重定向的關(guān)鍵
while (true) {
if (canceled) {
//取消則拋異常,諸如異步超時(shí)或者代碼主動(dòng)調(diào)用cancel方法都會(huì)改變布爾值
streamAllocation.release();
throw new IOException("Canceled");
}
Response response;
boolean releaseConnection = true;
try {
//前期工作已做好(主要是建立死循環(huán)和申請(qǐng)連接的StreamAllocation類 ),交給調(diào)用鏈調(diào)度給下一個(gè)攔截器處理
response = realChain.proceed(request, streamAllocation, null, null);
releaseConnection = false; //來(lái)到這里說(shuō)明請(qǐng)求服務(wù)器沒(méi)有問(wèn)題,啥異常都沒(méi)有出現(xiàn),服務(wù)器也有返回
} catch (RouteException e) {
// The attempt to connect via a route failed. The request will not have been sent.
//出現(xiàn)異常,是預(yù)料中的異常,調(diào)用recover方法看看能不能拯救一下(重連)
//注意這里僅僅是判斷能不能救(重連),并沒(méi)有發(fā)起真正的重連
if (!recover(e.getLastConnectException(), streamAllocation, false, request)) {
//判斷救不了了,直接拋異常,跳出死循環(huán)
throw e.getFirstConnectException();
}
releaseConnection = false;
continue;
} catch (IOException e) {
// An attempt to communicate with a server failed. The request may have been sent.
//出現(xiàn)異常,是預(yù)料中的異常,調(diào)用recover方法看看能不能拯救一下(重連)
//注意這里僅僅是判斷能不能救(重連),并沒(méi)有發(fā)起真正的重連
boolean requestSendStarted = !(e instanceof ConnectionShutdownException);
if (!recover(e, streamAllocation, requestSendStarted, request)) throw e; //判斷救不了了,直接拋異常,跳出死循環(huán)
releaseConnection = false;
continue;
} finally {
// We're throwing an unchecked exception. Release any resources.
//假如出現(xiàn)預(yù)料外的錯(cuò)誤(作者都想不出來(lái)會(huì)有這種錯(cuò)誤未加捕獲的),釋放連接
if (releaseConnection) {
streamAllocation.streamFailed(null);
streamAllocation.release();
}
}
// Attach the prior response if it exists. Such responses never have a body.
//如果先前已經(jīng)有一個(gè)response說(shuō)明先前已經(jīng)有過(guò)一次請(qǐng)求了,那么不是重連就是重定向
if (priorResponse != null) {
//標(biāo)記一下,給后面的followUpRequest做進(jìn)一步判斷(大概意思就是告訴后面這個(gè)請(qǐng)求有前科的,之前已經(jīng)請(qǐng)求過(guò)一次了)
response = response.newBuilder()
.priorResponse(priorResponse.newBuilder()
.body(null)
.build())
.build();
}
Request followUp;
try {
//綜合判斷假如是符合能重連或者重定向的條件返回一個(gè)不為空的Request,其他諸如請(qǐng)求成功,不符合重連和重定向條件的一律返回空
//判斷是否要重連或重定向的關(guān)鍵函數(shù)
followUp = followUpRequest(response, streamAllocation.route());
} catch (IOException e) {
streamAllocation.release();
throw e;
}
if (followUp == null) {
//返回request為空,那么嘗試釋放資源(可能是請(qǐng)求成功或者不符合重連重定向條件)
streamAllocation.release();
return response;
}
//來(lái)到這里說(shuō)明不是重連就是重定向了,關(guān)閉當(dāng)前response的輸入流
closeQuietly(response.body());
if (++followUpCount > MAX_FOLLOW_UPS) {
//跟進(jìn)數(shù)量超過(guò)最大值直接拋異常, MAX_FOLLOW_UPS值為20,這里主要是防止不停重定向,重連最多也就一次
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())) {
//假如跟原來(lái)的連接不一樣,那么重新申請(qǐng)一個(gè)連接
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;//賦值用于下個(gè)循環(huán)(重定向或重連)的請(qǐng)求
priorResponse = response;//賦值用于下個(gè)循環(huán)判斷（主要用于重連判斷）
}
}
再看看判斷是否要重連或重定向的關(guān)鍵函數(shù) followUpRequest,這里根據(jù)返回ResposeCode有不同的判斷,這里選取其中的一個(gè)重連的情況分析
private Request followUpRequest(Response userResponse, Route route) throws IOException {
if (userResponse == null) throw new IllegalStateException();
int responseCode = userResponse.code();
final String method = userResponse.request().method();
switch (responseCode) {
.....
//前面省略若干其他狀態(tài)的處理代碼
case HTTP_CLIENT_TIMEOUT:
// 408's are rare in practice, but some servers like HAProxy use this response code. The
// spec says that we may repeat the request without modifications. Modern browsers also
// repeat the request (even non-idempotent ones.)
if (!client.retryOnConnectionFailure()) {
// The application layer has directed us not to retry the request.
// 當(dāng)連接失敗是否進(jìn)行重連 值為false，則返回空Request
// 該布爾值可在OkHttpClient初始化里設(shè)置
return null;
}
if (userResponse.request().body() instanceof UnrepeatableRequestBody) {
return null;
}
if (userResponse.priorResponse() != null
&& userResponse.priorResponse().code() == HTTP_CLIENT_TIMEOUT) {
// 之前已經(jīng)請(qǐng)求過(guò)一次并且之前同樣是 超時(shí),說(shuō)明重連一次已經(jīng)失敗了,不再重連
return null;
}
if (retryAfter(userResponse, 0) > 0) {
//服務(wù)器有重連時(shí)間間隔,那么直接不重連了
return null;
}
//符合重連條件,返回Request對(duì)象以進(jìn)行再一次請(qǐng)求
return userResponse.request();
...........
//前面省略若干其他狀態(tài)的處理代碼
default:
//默認(rèn)返回Null,例如服務(wù)器正常返回200,那么也沒(méi)必要跟進(jìn)了,直接返回Null
return null;
}
}
其他狀態(tài)碼的流程也是相似的,都是根據(jù)返回Response信息進(jìn)行下一步處理,這里就不多敘述了
整體的調(diào)用流程就是攔截器維護(hù)一個(gè)死循環(huán),若是正常返回那么循環(huán)體只走一遍就直接返回；假如是超時(shí)等情況導(dǎo)致重連再請(qǐng)求一次,再走一次循環(huán)體無(wú)論成功與否都會(huì)直接跳出循環(huán),循環(huán)體走了兩遍；假如是重定向,最多允許重定向20次,即循環(huán)體走20次然后結(jié)束死循環(huán)
橋接攔截器(BridgeInterceptor)
@Override public Response intercept(Chain chain) throws IOException {
Request userRequest = chain.request();
Request.Builder requestBuilder = userRequest.newBuilder();
RequestBody body = userRequest.body();
if (body != null) {
//添加各種請(qǐng)求頭
MediaType contentType = body.contentType();
if (contentType != null) {
requestBuilder.header("Content-Type", contentType.toString());
}
long contentLength = body.contentLength();
//如果傳輸長(zhǎng)度不為-1，則表示完整傳輸
if (contentLength != -1) {
//設(shè)置頭信息 Content-Length
requestBuilder.header("Content-Length", Long.toString(contentLength));
requestBuilder.removeHeader("Transfer-Encoding");
} else {
//如果傳輸長(zhǎng)度為-1，則表示分塊傳輸，自動(dòng)設(shè)置頭信息
requestBuilder.header("Transfer-Encoding", "chunked");
requestBuilder.removeHeader("Content-Length");
}
}
if (userRequest.header("Host") == null) {
requestBuilder.header("Host", hostHeader(userRequest.url(), false));
}
//如果沒(méi)有設(shè)置頭信息 Connection，則自動(dòng)設(shè)置為 Keep-Alive
if (userRequest.header("Connection") == null) {
requestBuilder.header("Connection", "Keep-Alive");
}
// If we add an "Accept-Encoding: gzip" header field we're responsible for also decompressing
// the transfer stream.
// 是否運(yùn)用GZIP傳輸
//在 HTTP 傳輸時(shí)是支持 gzip 壓縮的,采用 gzip 壓縮后可以大幅減少傳輸內(nèi)容大小，這樣可以提高傳輸速
// 度，減少流量的使用。
boolean transparentGzip = false;
if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
transparentGzip = true;
requestBuilder.header("Accept-Encoding", "gzip");
}
//查找之前有沒(méi)有訪問(wèn)保存的cookie信息并帶上
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());
}
//轉(zhuǎn)發(fā)下一個(gè)攔截器
Response networkResponse = chain.proceed(requestBuilder.build());
//保存服務(wù)器返回的cookie信息
HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());
Response.Builder responseBuilder = networkResponse.newBuilder()
.request(userRequest);
//如果服務(wù)器傳輸有用到gzip壓縮,那么還要進(jìn)行解壓縮
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)));
}
return responseBuilder.build();
}
該攔截器主要負(fù)責(zé)添加請(qǐng)求頭,同時(shí)在服務(wù)器返回時(shí)保存必要的服務(wù)器信息例如cookie等,如果服務(wù)器有采用gzip傳輸還會(huì)先解壓得到原始數(shù)據(jù)再進(jìn)行返回
緩存攔截器(CacheInterceptor)
OKHttp默認(rèn)是沒(méi)有緩存的,除非我們?cè)贠kHttpClient初始化設(shè)置,基本用法如下
//緩存文件夾
File cacheFile = new File(getExternalCacheDir().toString(),"cache");
//緩存大小為10M
int cacheSize = 10 * 1024 * 1024;
//創(chuàng)建緩存對(duì)象
Cache cache = new Cache(cacheFile,cacheSize);//從這里也能看出緩存是直接存到本地存儲(chǔ)(SD卡)上的
OkHttpClient client = new OkHttpClient.Builder()
.cache(cache)//設(shè)置緩存
.build();
在設(shè)置了Cache后 緩存攔截器才算有用武之地,接下來(lái)慣例分析下緩存攔截器具體流程
@Override public Response intercept(Chain chain) throws IOException {
//是否有設(shè)置緩存,有？那么從本地存儲(chǔ)查找是否有該請(qǐng)求的緩存
Response cacheCandidate = cache != null
? cache.get(chain.request())
: null;
long now = System.currentTimeMillis();
//這個(gè)get方法是判斷最后該請(qǐng)求是用緩存還是網(wǎng)絡(luò)請(qǐng)求的關(guān)鍵,之前的本地存儲(chǔ)并沒(méi)有在時(shí)間上判斷緩存是否可用(例如這個(gè)緩存是一年前得了,那肯定是無(wú)效的),該函數(shù)內(nèi)做進(jìn)一步判斷(當(dāng)然還有其他必要的判斷)
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) {
//本地緩存有,但是不符合某個(gè)條件,舍棄了
closeQuietly(cacheCandidate.body()); // The cache candidate wasn't applicable. Close it.
}
// If we're forbidden from using the network and the cache is insufficient, fail.
//又沒(méi)網(wǎng)絡(luò)請(qǐng)求又沒(méi)緩存,直接返回504
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();
}
// If we don't need the network, we're done.
if (networkRequest == null) {
//通過(guò)new CacheStrategy.Factory類get方法得出網(wǎng)絡(luò)請(qǐng)求為空,同時(shí)越過(guò)上面代碼來(lái)到這里也說(shuō)明了緩存不為空,那么直接返回緩存
return cacheResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.build();
}
Response networkResponse = null;
try {
//拋給下一個(gè)攔截器進(jìn)行網(wǎng)絡(luò)連接
networkResponse = chain.proceed(networkRequest);
} finally {
// If we're crashing on I/O or otherwise, don't leak the cache body.
if (networkResponse == null && cacheCandidate != null) {
closeQuietly(cacheCandidate.body());
}
}
// If we have a cache response too, then we're doing a conditional get.
if (cacheResponse != null) {
if (networkResponse.code() == HTTP_NOT_MODIFIED) {
//假如緩存不為空同時(shí)服務(wù)器返回并無(wú)修改,進(jìn)行合并后關(guān)閉新的網(wǎng)絡(luò)請(qǐng)求Response里的輸入流
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();
// Update the cache after combining headers but before stripping the
// Content-Encoding header (as performed by initContentStream()).
cache.trackConditionalCacheHit();
//同時(shí)更新緩存
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)) {
// Offer this request to the cache.
//用戶有設(shè)置緩存,把本次網(wǎng)絡(luò)請(qǐng)求緩存到本地.同時(shí)返回給上一級(jí)攔截器
CacheRequest cacheRequest = cache.put(response);
return cacheWritingResponse(cacheRequest, response);
}
//檢測(cè)Http的Method參數(shù),諸如POST,DELETE等不需要緩存的請(qǐng)求直接移除
if (HttpMethod.invalidatesCache(networkRequest.method())) {
try {
cache.remove(networkRequest);
} catch (IOException ignored) {
// The cache cannot be written.
}
}
}
return response;
}
連接攔截器(ConnectInterceptor)
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Request request = realChain.request();
//獲取StreamAllocation 對(duì)象,先前分析由RetryAndFollowUpInterceptor生成并傳遞
StreamAllocation streamAllocation = realChain.streamAllocation();
// We need the network to satisfy this request. Possibly for validating a conditional GET.
boolean doExtensiveHealthChecks = !request.method().equals("GET");
//根據(jù)請(qǐng)求類型判斷在申請(qǐng)連接時(shí)是否要對(duì)連接的健全性進(jìn)行檢查
HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);//關(guān)鍵核心
//獲取連接
RealConnection connection = streamAllocation.connection();
//向下傳遞
return realChain.proceed(request, streamAllocation, httpCodec, connection);
}
連接攔截器的代碼看起來(lái)貌似是各攔截器中代碼最小的,其實(shí)代碼量是5個(gè)攔截器中最龐大的,這里的關(guān)鍵在于 streamAllocation.newStream(client, chain, doExtensiveHealthChecks).就是這里面獲取網(wǎng)絡(luò)連接,那么我們就跟進(jìn)去看看
//StreamAllocation.java
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 {
//根據(jù)條件返回一個(gè)符合條件且健全的連接
RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);//關(guān)鍵核心
//同時(shí)返回Http層面處理類(個(gè)人理解這類只對(duì)外暴露Http層面的方法和變量,細(xì)化操作)
HttpCodec resultCodec = resultConnection.newCodec(client, chain, this);
synchronized (connectionPool) {
codec = resultCodec;
return resultCodec;
}
} catch (IOException e) {
throw new RouteException(e);
}
}
//StreamAllocation.java
private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
int writeTimeout, int pingIntervalMillis, boolean connectionRetryEnabled,
boolean doExtensiveHealthChecks) throws IOException {
//維護(hù)一個(gè)死循環(huán)
while (true) {
//找到符合條件的連接
RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
pingIntervalMillis, connectionRetryEnabled);//關(guān)鍵核心
// If this is a brand new connection, we can skip the extensive health checks.
synchronized (connectionPool) {
if (candidate.successCount == 0) {
//沒(méi)請(qǐng)求過(guò)一次說(shuō)明是新連接,直接返回
return candidate;
}
}
// Do a (potentially slow) check to confirm that the pooled connection is still good. If it
// isn't, take it out of the pool and start again.
//否則對(duì)連接進(jìn)行健全性檢查
if (!candidate.isHealthy(doExtensiveHealthChecks)) {
noNewStreams();
//不健康的,那么繼續(xù)死循環(huán),直到找到可用的健全的連接為止
continue;
}
return candidate;
}
}
從findHealthyConnection方法中我們可以很直觀地看到申請(qǐng)連接的表面流程:死循環(huán),返回后對(duì)候選鏈接做健全性檢查,不合格就繼續(xù)申請(qǐng),直到找到可用連接為止
那么具體申請(qǐng)連接的流程我們?cè)俑M(jìn)去看看
//StreamAllocation.java
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");
// Attempt to use an already-allocated connection. We need to be careful here because our
// already-allocated connection may have been restricted from creating new streams.
releasedConnection = this.connection;
toClose = releaseIfNoNewStreams();
if (this.connection != null) {
//如果連接不為空,說(shuō)明之前就有申請(qǐng)一個(gè)了,直接賦值
result = this.connection;
releasedConnection = null;
}
if (!reportedAcquired) {
// If the connection was never reported acquired, don't report it as released!
releasedConnection = null;
}
if (result == null) {
//沒(méi)有現(xiàn)成連接,那么從連接池申請(qǐng)
Internal.instance.get(connectionPool, address, this, null);//從連接池申請(qǐng)并賦值到該類的connection變量
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) {
// If we found an already-allocated or pooled connection, we're done.
return result;
}
// If we need a route selection, make one. This is a blocking operation.
//看看是否需要選定一個(gè)路由,沒(méi)有的話需要選擇一個(gè)路由,并重新從連接池取連接
boolean newRouteSelection = false;
if (selectedRoute == null && (routeSelection == null || !routeSelection.hasNext())) {
newRouteSelection = true;
routeSelection = routeSelector.next();
}
synchronized (connectionPool) {
if (canceled) throw new IOException("Canceled");
if (newRouteSelection) {
// Now that we have a set of IP addresses, make another attempt at getting a connection from
// the pool. This could match due to connection coalescing.
List<Route> routes = routeSelection.getAll();
for (int i = 0, size = routes.size(); i < size; i++) {
Route route = routes.get(i);
//根據(jù)路由,主機(jī)地址等相關(guān)信息在連接池中找出符合條件的連接
Internal.instance.get(connectionPool, address, this, route);
if (connection != null) {
foundPooledConnection = true;
result = connection;
this.route = route;
break;
}
}
}
if (!foundPooledConnection) {
if (selectedRoute == null) {
selectedRoute = routeSelection.next();
}
//沒(méi)有現(xiàn)成的連接,連接池也沒(méi)有復(fù)用連接
route = selectedRoute;
refusedStreamCount = 0;
//那么直接新建一個(gè)連接
result = new RealConnection(connectionPool, selectedRoute);
acquire(result, false);
}
}
// If we found a pooled connection on the 2nd time around, we're done.
if (foundPooledConnection) {
eventListener.connectionAcquired(call, result);
return result;
}
//開(kāi)始握手建立連接,這是一個(gè)阻塞操作
result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,
connectionRetryEnabled, call, eventListener);//關(guān)鍵核心
routeDatabase().connected(result.route());
Socket socket = null;
synchronized (connectionPool) {
reportedAcquired = true;
// 連接已建立,那么存進(jìn)連接池準(zhǔn)備復(fù)用
Internal.instance.put(connectionPool, result);
//假如該RealConnection有多路連接(跟進(jìn)去發(fā)現(xiàn)是Http2相關(guān),下面這段可暫時(shí)無(wú)視),那么關(guān)掉多余的socket連接
if (result.isMultiplexed()) {
socket = Internal.instance.deduplicate(connectionPool, address, this);
result = connection;
}
}
closeQuietly(socket);
eventListener.connectionAcquired(call, result);
return result;
}
具體的找連接流程就是先找找有沒(méi)有之前就用過(guò)的連接(例如重連或重定向的情況),沒(méi)有看看連接池有沒(méi)有復(fù)用的連接,我們?cè)倏纯碦ealConnection類的connect方法
public void connect(int connectTimeout, int readTimeout, int writeTimeout,
int pingIntervalMillis, boolean connectionRetryEnabled, Call call,
EventListener eventListener) {
.....
//前面省略若干代碼
while (true) {
try {
if (route.requiresTunnel()) {
//假如是通過(guò)http端口做https請(qǐng)求
connectTunnel(connectTimeout, readTimeout, writeTimeout, call, eventListener);
if (rawSocket == null) {
// We were unable to connect the tunnel but properly closed down our resources.
break;
}
} else {
//建立socket連接,關(guān)鍵核心
connectSocket(connectTimeout, readTimeout, call, eventListener);
}
establishProtocol(connectionSpecSelector, pingIntervalMillis, call, eventListener);
eventListener.connectEnd(call, route.socketAddress(), route.proxy(), protocol);
break;
} catch (IOException e) {
....
//省略若干代碼
}
}
.....
//省略若干代碼
}
private void connectSocket(int connectTimeout, int readTimeout, Call call,
EventListener eventListener) throws IOException {
Proxy proxy = route.proxy();
Address address = route.address();
rawSocket = proxy.type() == Proxy.Type.DIRECT || proxy.type() == Proxy.Type.HTTP
? address.socketFactory().createSocket()
: new Socket(proxy);
eventListener.connectStart(call, route.socketAddress(), proxy);
rawSocket.setSoTimeout(readTimeout);
try {
//建立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;
}
// The following try/catch block is a pseudo hacky way to get around a crash on Android 7.0
// More details:
// https://github.com/square/okhttp/issues/3245
// https://android-review.googlesource.com/#/c/271775/
try {
//通過(guò)Okio與Socket建立輸入輸出流
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);
}
}
}
至此完成了返回連接的過(guò)程,從代碼也能看出OkHttp3直接用Socket配合Okio輸入輸出流,大家都知道Http本身就是基于Tcp,Tcp用Socket做的,相當(dāng)于OkHttp3重做了一遍Http,至于為什么要重做,我個(gè)人認(rèn)為有兩點(diǎn)原因:
1.不滿意原有JAVAIO 流,用OKIO替代
2.取出底層的Socket連接,用于連接池復(fù)用
呼叫服務(wù)器攔截器(CallServerInterceptor)
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
HttpCodec httpCodec = realChain.httpStream();
StreamAllocation streamAllocation = realChain.streamAllocation();
RealConnection connection = (RealConnection) realChain.connection();
Request request = realChain.request();
long sentRequestMillis = System.currentTimeMillis();
realChain.eventListener().requestHeadersStart(realChain.call());
//寫(xiě)入請(qǐng)求頭信息,內(nèi)部寫(xiě)入到RealBufferSink中的緩沖區(qū)(并未射出到服務(wù)器的輸出流)
httpCodec.writeRequestHeaders(request);
realChain.eventListener().requestHeadersEnd(realChain.call(), request);
Response.Builder responseBuilder = null;
// 100-continue用于客戶端在發(fā)送POST數(shù)據(jù)給服務(wù)器前，征詢服務(wù)器情況，看服務(wù)器是否處理POST的數(shù)據(jù)。
//如果不處理，客戶端則不上傳POST數(shù)據(jù)，如果處理，則POST上傳數(shù)據(jù)。在現(xiàn)實(shí)應(yīng)用中，通過(guò)在POST大數(shù)據(jù)時(shí)，才會(huì)使用100-continue協(xié)議。
if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
//請(qǐng)求頭確實(shí)有100-continue,那么把緩沖區(qū)的輸出內(nèi)容全部射出給服務(wù)器
httpCodec.flushRequest();
realChain.eventListener().responseHeadersStart(realChain.call());
//根據(jù)服務(wù)器返回信息構(gòu)建ResponseBuilder
//這里假如是代碼100返回Null,其他返回均不為Null
responseBuilder = httpCodec.readResponseHeaders(true);
}
if (responseBuilder == null) {
//服務(wù)器返回100,那么開(kāi)始那么就把請(qǐng)求體加上
realChain.eventListener().requestBodyStart(realChain.call());
long contentLength = request.body().contentLength();
CountingSink requestBodyOut =
new CountingSink(httpCodec.createRequestBody(request, contentLength));
BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
//請(qǐng)求體寫(xiě)入緩沖區(qū)
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();
}
}
//再次射出到與服務(wù)器的輸出流里,至此請(qǐng)求已發(fā)送完畢
httpCodec.finishRequest();//跟 httpCodec.flushRequest()一樣的效果
if (responseBuilder == null) {
//之前的分析過(guò)狀態(tài)碼100該類是空的,那么要再讀一次構(gòu)建responseBuilder用來(lái)構(gòu)建真正的Response
realChain.eventListener().responseHeadersStart(realChain.call());
//獲取從服務(wù)器讀取頭信息構(gòu)建responseBuilder
responseBuilder = httpCodec.readResponseHeaders(false);
}
//計(jì)劃構(gòu)建實(shí)際的Response
Response response = responseBuilder
.request(request)
.handshake(streamAllocation.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
int code = response.code();
if (code == 100) {
//服務(wù)器居然還返回100,那么再讀一次構(gòu)建新的ResponseBuilder
responseBuilder = httpCodec.readResponseHeaders(false);
//這下構(gòu)建的ResponseBuilder基本上能構(gòu)建有實(shí)際信息返回的Response了
response = responseBuilder
.request(request)
.handshake(streamAllocation.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
code = response.code();
}
realChain.eventListener()
.responseHeadersEnd(realChain.call(), response);
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 {
//構(gòu)建帶實(shí)際返回信息的Response
// 其實(shí)在之前 httpCodec.finishRequest()中服務(wù)器收到請(qǐng)求立刻就進(jìn)行處理并返回?cái)?shù)據(jù)給客戶端的流里面了,但是客戶端還沒(méi)有通過(guò)輸入流獲取流里面的信息
response = response.newBuilder()
.body(httpCodec.openResponseBody(response))//包裝一個(gè)輸入流,供后面從流讀取數(shù)據(jù)
.build();
}
//假如服務(wù)器或客戶端并不希望建立長(zhǎng)連接,那么這里先把noStream標(biāo)記為true,后面一旦請(qǐng)求到返回?cái)?shù)據(jù)就把連接斷掉
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());
}
//返回Response給上層
return response;
}
看到這里其實(shí)我們發(fā)現(xiàn)即使到了最后一個(gè)攔截器也并沒(méi)有從輸入流讀取數(shù)據(jù)(盡管服務(wù)器早已把數(shù)據(jù)輸入到流里面了),僅僅相當(dāng)于打開(kāi)了輸入流的入口,那么真正從輸入流讀取數(shù)據(jù)是在哪呢
這里以我以前發(fā)起的一個(gè)異步網(wǎng)絡(luò)請(qǐng)求為例,代碼如下
NetWorkUtil.postNetWork(jsonObject, Constants.getBaseUrl(), new Callback() {
@Override
public void onFailure(Call call, IOException e) {
...
}
@Override
public void onResponse(Call call, Response response) throws IOException {
String stringGson = response.body().string();
...
}
});
我們跳進(jìn)去這個(gè)response.body().string()方法看看
//ResponseBody.java
public final String string() throws IOException {
BufferedSource source = source();
try {
//判斷用什么樣的字符集類型解析數(shù)據(jù)
Charset charset = Util.bomAwareCharset(source, charset());
return source.readString(charset);//從輸入流讀取數(shù)據(jù)并根據(jù)字符集類型轉(zhuǎn)換成字符串
} finally {
//關(guān)閉socket輸入流,清空緩沖區(qū)
Util.closeQuietly(source);
}
}
總結(jié)
通過(guò)學(xué)習(xí)大概了解了OkHttp3的基本調(diào)用流程，發(fā)現(xiàn)內(nèi)部原來(lái)很多東西跟自己所想的完全不一樣,也學(xué)到了
很多,其實(shí)OkHttp3還有很多需要探討的東西,本文也只是描繪了OkHttp3個(gè)各攔截器的流程,具體還有很多細(xì)節(jié)需要深挖例如OkIo的內(nèi)部實(shí)現(xiàn),連接池具體是咋操作的等等。
