Request/Response
Request是發(fā)送請求封裝類，內(nèi)部有url缚甩，header骆莹， method颗搂，body等常見的參數(shù)，Response是請求的結(jié)果幕垦，包含code，message傅联，header先改，body；這兩個(gè)類的定義是完全符合Http協(xié)議所定義的請求內(nèi)容和響應(yīng)內(nèi)容蒸走。
OkHttpClient
Call和WebSocket實(shí)例對象的一個(gè)工廠類仇奶，用于發(fā)送HTTP請求和讀取響應(yīng)。
OkHttpClient.Builder
內(nèi)部類Builder用于構(gòu)建OkHttpClient實(shí)例比驻。其無參構(gòu)造方法設(shè)置OkHttpClient的默認(rèn)參數(shù)值,其方法設(shè)置OkHttpClient的特定參數(shù)值该溯。
// Builer默認(rèn)構(gòu)造函數(shù),OkHttpClient的默認(rèn)參數(shù)都在這里設(shè)置
public Builder() {
dispatcher = new Dispatcher();
protocols = DEFAULT_PROTOCOLS;
connectionSpecs = DEFAULT_CONNECTION_SPECS;
eventListenerFactory = EventListener.factory(EventListener.NONE);
proxySelector = ProxySelector.getDefault();
if (proxySelector == null) {
proxySelector = new NullProxySelector();
}
cookieJar = CookieJar.NO_COOKIES;
socketFactory = SocketFactory.getDefault();
hostnameVerifier = OkHostnameVerifier.INSTANCE;
certificatePinner = CertificatePinner.DEFAULT;
proxyAuthenticator = Authenticator.NONE;
authenticator = Authenticator.NONE;
connectionPool = new ConnectionPool();
dns = Dns.SYSTEM;
followSslRedirects = true;
followRedirects = true;
retryOnConnectionFailure = true;
callTimeout = 0;
connectTimeout = 10_000;
readTimeout = 10_000;
writeTimeout = 10_000;
pingInterval = 0;
}
重要的方法
// 構(gòu)造Call對象
@Override
public Call newCall(Request request) {
return RealCall.newRealCall(this, request, false /* for web socket */);
}
// 構(gòu)造WebSocket對象
@Override
public WebSocket newWebSocket(Request request, WebSocketListener listener) {
RealWebSocket webSocket = new RealWebSocket(request, listener, new Random(), pingInterval);
webSocket.connect(this);
return webSocket;
}
RealCall
真正的Call的實(shí)現(xiàn)類。負(fù)責(zé)請求的調(diào)度（同步和異步别惦，同步即是走當(dāng)前線程發(fā)送請求狈茉，異步則使用OkHttp內(nèi)部的線程池進(jìn)行）；負(fù)責(zé)構(gòu)造內(nèi)部邏輯責(zé)任鏈掸掸，并執(zhí)行責(zé)任鏈相關(guān)邏輯氯庆，知道獲取結(jié)果。
重要的方法
// 真正構(gòu)造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);
// 構(gòu)造了Transimitter對象
call.transmitter = new Transmitter(client, call);
return call;
}
// 執(zhí)行同步請求的方法
@Override
public Response execute() throws IOException {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
transmitter.timeoutEnter();
transmitter.callStart();
try {
// 將同步的Call對象加入Dispatcher的runningSyncCalls隊(duì)列中
client.dispatcher().executed(this);
// 重點(diǎn)在此處,OkHttpClient的責(zé)任鏈模式
return getResponseWithInterceptorChain();
} finally {
// 將同步的Call對象從Dispatcher的runningSyncCalls隊(duì)列中刪除// ,并尋找可以加入到Dispatcher的runningAsyncCalls隊(duì)列中的
// AsyncCall對象,將其加入到線程池中等待執(zhí)行扰付。最后走OkHttpCl// ient的責(zé)任鏈模式.
client.dispatcher().finished(this);
}
}
// 執(zhí)行異步請求的方法
@Override
public void enqueue(Callback responseCallback) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
transmitter.callStart();
// 構(gòu)造異步的AsyncCall對象,將AsyncCall對象加入Dispatcher的
// readyAsyncCalls隊(duì)列中
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
// 重點(diǎn)堤撵，走責(zé)任鏈模式
Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
// 構(gòu)造Interceptor的集合
List<Interceptor> interceptors = new ArrayList<>();
// 自定義的interceptors,被稱為application interceptors
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,被稱為network interceptors
interceptors.addAll(client.networkInterceptors());
}
interceptors.add(new CallServerInterceptor(forWebSocket));
// 用上面的集合構(gòu)造責(zé)任鏈對象,并傳遞index進(jìn)去,index決定了責(zé)任鏈的哪一個(gè)interceptors
Interceptor.Chain chain = new RealInterceptorChain(interceptors, transmitter, null, 0,
originalRequest, this, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
boolean calledNoMoreExchanges = false;
try {
// 責(zé)任鏈的proceed方法,開始責(zé)任鏈
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);
}
}
}
Interceptor（OkHttp的核心）
RetryAndFollowUpInterceptor（失敗和重定向攔截器）
重要的方法
@Override
public Response intercept(Chain chain) throws IOException {
Request request = chain.request();
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Transmitter transmitter = realChain.transmitter();
int followUpCount = 0;
Response priorResponse = null;
while (true) {
// 為發(fā)送數(shù)據(jù)做準(zhǔn)備,會創(chuàng)建ExchangeFinder對象,為后面獲取exchange對象做準(zhǔn)備
transmitter.prepareToConnect(request);
if (transmitter.isCanceled()) {
throw new IOException("Canceled");
}
Response response;
boolean success = false;
try {
// 下一責(zé)任鏈
response = realChain.proceed(request, transmitter, null);
success = true;
} catch (RouteException e) {
// The attempt to connect via a route failed. The request will not have been sent.
// 是否發(fā)生Route類型的異常,如果有,判斷是否滿足重試條件,滿足則continue重試,重試的邏輯在recover方法中
if (!recover(e.getLastConnectException(), transmitter, false, request)) {
throw e.getFirstConnectException();
}
continue;
} catch (IOException e) {
// An attempt to communicate with a server failed. The request may have been sent.
// 是否發(fā)生IO類型的異常,如果有,判斷是否滿足重試條件,滿足則continue重試
boolean requestSendStarted = !(e instanceof ConnectionShutdownException);
if (!recover(e, transmitter, requestSendStarted, request)) throw e;
continue;
} finally {
// The network call threw an exception. Release any resources.
if (!success) {
transmitter.exchangeDoneDueToException();
}
}
// Attach the prior response if it exists. Such responses never have a body.
if (priorResponse != null) {
response = response.newBuilder()
.priorResponse(priorResponse.newBuilder()
.body(null)
.build())
.build();
}
Exchange exchange = Internal.instance.exchange(response);
Route route = exchange != null ? exchange.connection().route() : null;
// 這里處理是否重定向的邏輯并開始構(gòu)造重定向請求,具體情況看源碼分析
Request followUp = followUpRequest(response, route);
if (followUp == null) {
if (exchange != null && exchange.isDuplex()) {
transmitter.timeoutEarlyExit();
}
return response;
}
RequestBody followUpBody = followUp.body();
if (followUpBody != null && followUpBody.isOneShot()) {
return response;
}
closeQuietly(response.body());
if (transmitter.hasExchange()) {
exchange.detachWithViolence();
}
if (++followUpCount > MAX_FOLLOW_UPS) {
throw new ProtocolException("Too many follow-up requests: " + followUpCount);
}
request = followUp;
priorResponse = response;
}
}
Interceptors和NetworkInterceptors的區(qū)別
在 OkHttpClient.Builder 中，使用者可以通過 addInterceptor 和 addNetworkdInterceptor 添加自定義的攔截器羽莺，分析完 RetryAndFollowUpInterceptor 就可以知道這兩種自動(dòng)攔截器的區(qū)別了实昨。
從添加攔截器的順序可以知道 Interceptors 和 networkInterceptors 剛好一個(gè)在 RetryAndFollowUpInterceptor 的前面，一個(gè)在后面盐固。
可以分析出來荒给，假如一個(gè)請求在 RetryAndFollowUpInterceptor 這個(gè)攔截器內(nèi)部重試或者重定向了 N 次丈挟，那么其內(nèi)部嵌套的所有攔截器也會被調(diào)用N次，同樣 networkInterceptors 自定義的攔截器也會被調(diào)用 N 次锐墙。而相對的 Interceptors 則一個(gè)請求只會調(diào)用一次礁哄，所以在OkHttp的內(nèi)部也將其稱之為 Application Interceptor。
BridgeInterceptor（封裝request和response攔截器）
重要的方法
@Override
public Response intercept(Chain chain) throws IOException {
Request userRequest = chain.request();
Request.Builder requestBuilder = userRequest.newBuilder();
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");
}
// If we add an "Accept-Encoding: gzip" header field we're responsible for also decompressing
// the transfer stream.
boolean transparentGzip = false;
if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
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());
}
Response networkResponse = chain.proceed(requestBuilder.build());
HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());
Response.Builder responseBuilder = networkResponse.newBuilder()
.request(userRequest);
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();
}
這個(gè)攔截器比較簡單,功能如下:
- 負(fù)責(zé)把用戶構(gòu)造的請求轉(zhuǎn)換為發(fā)送到服務(wù)器的請求 溪北、把服務(wù)器返回的響應(yīng)轉(zhuǎn)換為用戶友好的響應(yīng)桐绒，是從應(yīng)用程序代碼到網(wǎng)絡(luò)代碼的橋梁
- 設(shè)置內(nèi)容長度，內(nèi)容編碼
- 設(shè)置gzip壓縮之拨，并在接收到內(nèi)容后進(jìn)行解壓茉继。省去了應(yīng)用層處理數(shù)據(jù)解壓的麻煩
- 添加cookie
- 設(shè)置其他報(bào)頭，如User-Agent,Host,Keep-alive等蚀乔。其中Keep-Alive是實(shí)現(xiàn)連接復(fù)用的必要步驟
CacheInterceptor（緩存攔截器）
重要的方法
@Override
public Response intercept(Chain chain) throws IOException {
// 通過Request在Cache中拿緩存,前提是OkHttpClient中配置了緩存,默認(rèn)不支持
Response cacheCandidate = cache != null
? cache.get(chain.request())
: null;
long now = System.currentTimeMillis();
// 根據(jù)response,time,request構(gòu)造一個(gè)緩存策略烁竭，用于判斷怎樣使用緩存。
CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();
// 如果該請求沒有使用網(wǎng)絡(luò)就為null
Request networkRequest = strategy.networkRequest;
// 如果該請求沒有使用緩存則為空
Response cacheResponse = strategy.cacheResponse;
if (cache != null) {
cache.trackResponse(strategy);
}
if (cacheCandidate != null && cacheResponse == null) {
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.
// 如果緩存策略中設(shè)置禁止使用網(wǎng)絡(luò)吉挣，并且緩存也為空派撕，則構(gòu)建一個(gè)Response直接返回，注意返回碼=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.
// 如果不使用網(wǎng)絡(luò)但有緩存,則返回緩存
if (networkRequest == null) {
return cacheResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.build();
}
Response networkResponse = null;
try {
// 走后續(xù)攔截器流程
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.
// 緩存存在且網(wǎng)絡(luò)返回的Response為304,則使用緩存的Response
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();
// Update the cache after combining headers but before stripping the
// Content-Encoding header (as performed by initContentStream()).
cache.trackConditionalCacheHit();
// 更新緩存
cache.update(cacheResponse, response);
return response;
} else {
closeQuietly(cacheResponse.body());
}
}
// 構(gòu)建網(wǎng)絡(luò)請求的Resposne
Response response = networkResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.networkResponse(stripBody(networkResponse))
.build();
// OkHttpClient中配置了Cache的話,緩存Response
if (cache != null) {
if (HttpHeaders.hasBody(response) && CacheStrategy.isCacheable(response, networkRequest)) {
// Offer this request to the cache.
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;
}
ConnectInterceptor（網(wǎng)絡(luò)連接攔截器,負(fù)責(zé)和服務(wù)器建立連接,重點(diǎn)睬魂，未完成）
重要的方法
@Override
public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Request request = realChain.request();
Transmitter transmitter = realChain.transmitter();
// We need the network to satisfy this request. Possibly for validating a conditional GET.
boolean doExtensiveHealthChecks = !request.method().equals("GET");
// 注釋1 從Transmitter中獲取新的Exchange對象
Exchange exchange = transmitter.newExchange(chain, doExtensiveHealthChecks);
return realChain.proceed(request, transmitter, exchange);
}
源碼邏輯跳轉(zhuǎn):
// 源碼位置: Transmitter.java
/** Returns a new exchange to carry a new request and response. */
Exchange newExchange(Interceptor.Chain chain, boolean doExtensiveHealthChecks) {
synchronized (connectionPool) {
if (noMoreExchanges) {
throw new IllegalStateException("released");
}
if (exchange != null) {
throw new IllegalStateException("cannot make a new request because the previous response "
+ "is still open: please call response.close()");
}
}
// 調(diào)用ExchangeFinder的find()獲取ExchangeCodec
ExchangeCodec codec = exchangeFinder.find(client, chain, doExtensiveHealthChecks);
// 用上面獲取的codec對象構(gòu)建新的Exchange對象
Exchange result = new Exchange(this, call, eventListener, exchangeFinder, codec);
synchronized (connectionPool) {
this.exchange = result;
this.exchangeRequestDone = false;
this.exchangeResponseDone = false;
return result;
}
}
// 源碼位置: ExchangFinder.java
public ExchangeCodec find(
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 {
// 調(diào)用自身findHealthyConnection方法獲取RealConnection
RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
return resultConnection.newCodec(client, chain);
} catch (RouteException e) {
trackFailure();
throw e;
} catch (IOException e) {
trackFailure();
throw new RouteException(e);
}
}
// 源碼位置: ExchangeFinder.java
/**
* Finds a connection and returns it if it is healthy. If it is unhealthy the process is repeated
* until a healthy connection is found.
*/
private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
int writeTimeout, int pingIntervalMillis, boolean connectionRetryEnabled,
boolean doExtensiveHealthChecks) throws IOException {
while (true) {
// 在ExchangeFinder的findConnection方法循環(huán)獲取可用的RealConnection
RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
pingIntervalMillis, connectionRetryEnabled);
// If this is a brand new connection, we can skip the extensive health checks.
synchronized (connectionPool) {
// 判斷獲取的RealConnection是否可用,若可用返回,不可用繼續(xù)尋找
if (candidate.successCount == 0 && !candidate.isMultiplexed()) {
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.
if (!candidate.isHealthy(doExtensiveHealthChecks)) {
candidate.noNewExchanges();
continue;
}
return candidate;
}
}
// 源碼位置: ExchangeFinder.java
/**
* Returns a connection to host a new stream. This prefers the existing connection if it exists,
* then the pool, finally building a new connection.
*/
private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException {
boolean foundPooledConnection = false;
RealConnection result = null;
Route selectedRoute = null;
RealConnection releasedConnection;
Socket toClose;
synchronized (connectionPool) {
if (transmitter.isCanceled()) throw new IOException("Canceled");
hasStreamFailure = false; // This is a fresh attempt.
// 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 exchanges.
// 嘗試使用已分配的連接,已經(jīng)分配的連接可能已經(jīng)被限制創(chuàng)建新的流
releasedConnection = transmitter.connection;
toClose = transmitter.connection != null && transmitter.connection.noNewExchanges
? transmitter.releaseConnectionNoEvents()
: null;
if (transmitter.connection != null) {
// We had an already-allocated connection and it's good.
// 已分配連接,并且該連接可用
result = transmitter.connection;
releasedConnection = null;
}
if (result == null) {
// Attempt to get a connection from the pool.
// 嘗試從連接池中獲取一個(gè)連接
if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, null, false)) {
foundPooledConnection = true;
result = transmitter.connection;
} else if (nextRouteToTry != null) {
selectedRoute = nextRouteToTry;
nextRouteToTry = null;
} else if (retryCurrentRoute()) {
selectedRoute = transmitter.connection.route();
}
}
}
// 關(guān)閉連接
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.
// 如果已經(jīng)從連接池中獲取到了一個(gè)連接终吼，就將其返回
return result;
}
// If we need a route selection, make one. This is a blocking operation.
boolean newRouteSelection = false;
if (selectedRoute == null && (routeSelection == null || !routeSelection.hasNext())) {
newRouteSelection = true;
routeSelection = routeSelector.next();
}
List<Route> routes = null;
synchronized (connectionPool) {
if (transmitter.isCanceled()) 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.
// 根據(jù)一系列的 IP 地址從連接池中獲取一個(gè)鏈接
routes = routeSelection.getAll();
if (connectionPool.transmitterAcquirePooledConnection(
address, transmitter, routes, false)) {
foundPooledConnection = true;
result = transmitter.connection;
}
}
if (!foundPooledConnection) {
if (selectedRoute == null) {
selectedRoute = routeSelection.next();
}
// Create a connection and assign it to this allocation immediately. This makes it possible
// for an asynchronous cancel() to interrupt the handshake we're about to do.
// 創(chuàng)建一個(gè)新的連接，并將其分配氯哮，這樣我們就可以在握手之前進(jìn)行終端
result = new RealConnection(connectionPool, selectedRoute);
connectingConnection = result;
}
}
// If we found a pooled connection on the 2nd time around, we're done.
// 如果我們在第二次的時(shí)候發(fā)現(xiàn)了一個(gè)池連接际跪，那么我們就將其返回
if (foundPooledConnection) {
eventListener.connectionAcquired(call, result);
return result;
}
// Do TCP + TLS handshakes. This is a blocking operation.
// 進(jìn)行 TCP 和 TLS 握手
result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,
connectionRetryEnabled, call, eventListener);
connectionPool.routeDatabase.connected(result.route());
Socket socket = null;
synchronized (connectionPool) {
connectingConnection = null;
// Last attempt at connection coalescing, which only occurs if we attempted multiple
// concurrent connections to the same host.
if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, routes, true)) {
// We lost the race! Close the connection we created and return the pooled connection.
result.noNewExchanges = true;
socket = result.socket();
result = transmitter.connection;
// It's possible for us to obtain a coalesced connection that is immediately unhealthy. In
// that case we will retry the route we just successfully connected with.
nextRouteToTry = selectedRoute;
} else {
connectionPool.put(result);
transmitter.acquireConnectionNoEvents(result);
}
}
closeQuietly(socket);
eventListener.connectionAcquired(call, result);
return result;
}
注釋1處(跟源碼)內(nèi)部邏輯如下:
- ConnectInterceptor調(diào)用transmitter.newExchange
- Transmitter先調(diào)用ExchangeFinder的find()獲得ExchangeCodec
- ExchangeFinder調(diào)用自身的findHealthyConnection獲得RealConnection
- ExchangeFinder的findHealthyConnection方法調(diào)用自身的findConnection獲得RealConnection
- ExchangeFinder通過剛才獲取的RealConnection的codec()方法獲得ExchangeCodec
- Transmitter獲取到了ExchangeCodec，然后new了一個(gè)ExChange喉钢，將剛才的ExchangeCodec包含在內(nèi)姆打。
通過上面的邏輯,ConnectInterceptor可以獲得一個(gè)Exchange類,這個(gè)類有兩個(gè)實(shí)現(xiàn),一個(gè)是Http1ExchangeCodec,一個(gè)是Http2ExchangeCodec,分別對應(yīng)Http1和Http2協(xié)議。
Exchange類里面包含了ExchangeCodec對象肠虽，而這個(gè)對象里面又包含了一個(gè)RealConnection對象幔戏，RealConnection的屬性成員有socket、handlShake舔痕、protocol等评抚，可見它應(yīng)該是一個(gè)Socket連接的包裝類，而ExchangeCode對象是對RealConnection操作（writeRequestHeader伯复、readResposneHeader）的封裝慨代。
CallServerInterceptor（執(zhí)行流操作攔截器,負(fù)責(zé)向服務(wù)器發(fā)送請求數(shù)據(jù)、從服務(wù)器讀取響應(yīng)數(shù)據(jù) 進(jìn)行http請求報(bào)文的封裝與請求報(bào)文的解析）
重要的方法
@Override
public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Exchange exchange = realChain.exchange();
Request request = realChain.request();
long sentRequestMillis = System.currentTimeMillis();
// 寫入請求頭
exchange.writeRequestHeaders(request);
boolean responseHeadersStarted = false;
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"))) {
exchange.flushRequest();
responseHeadersStarted = true;
exchange.responseHeadersStart();
responseBuilder = exchange.readResponseHeaders(true);
}
// 寫入請求體
if (responseBuilder == null) {
if (request.body().isDuplex()) {
// Prepare a duplex body so that the application can send a request body later.
exchange.flushRequest();
BufferedSink bufferedRequestBody = Okio.buffer(
exchange.createRequestBody(request, true));
request.body().writeTo(bufferedRequestBody);
} else {
// Write the request body if the "Expect: 100-continue" expectation was met.
BufferedSink bufferedRequestBody = Okio.buffer(
exchange.createRequestBody(request, false));
request.body().writeTo(bufferedRequestBody);
bufferedRequestBody.close();
}
} else {
exchange.noRequestBody();
if (!exchange.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.
exchange.noNewExchangesOnConnection();
}
}
} else {
exchange.noRequestBody();
}
if (request.body() == null || !request.body().isDuplex()) {
exchange.finishRequest();
}
if (!responseHeadersStarted) {
exchange.responseHeadersStart();
}
if (responseBuilder == null) {
// 讀取響應(yīng)頭
responseBuilder = exchange.readResponseHeaders(false);
}
Response response = responseBuilder
.request(request)
.handshake(exchange.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
response = exchange.readResponseHeaders(false)
.request(request)
.handshake(exchange.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
code = response.code();
}
exchange.responseHeadersEnd(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 {
// 讀取響應(yīng)體
response = response.newBuilder()
.body(exchange.openResponseBody(response))
.build();
}
if ("close".equalsIgnoreCase(response.request().header("Connection"))
|| "close".equalsIgnoreCase(response.header("Connection"))) {
exchange.noNewExchangesOnConnection();
}
if ((code == 204 || code == 205) && response.body().contentLength() > 0) {
throw new ProtocolException(
"HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
}
return response;
}
CallServerInterceptor由以下步驟組成:
- 向服務(wù)器發(fā)送 request header
- 如果有 request body啸如，就向服務(wù)器發(fā)送
- 讀取 response header侍匙，先構(gòu)造一個(gè) Response 對象
- 如果有 response body，就在 3 的基礎(chǔ)上加上 body 構(gòu)造一個(gè)新的 Response 對象
