源碼解析之前先看之前請(qǐng)求過程
private OkHttpClient mClient=new OkHttpClient();
public static final String BASE_URL="http://greatfeng.top/app";
public void run() throws Exception {
Request request = new Request.Builder()
.url(BASE_URL+"/weather?cityname="+"上海")
.build();
mClient.newCall(request).enqueue(new Callback() {
@Override public void onFailure(Call call, IOException e) {
e.printStackTrace();
}
@Override public void onResponse(Call call, Response response) throws IOException {
if (!response.isSuccessful()) throw new IOException("Unexpected code " + response);
Headers responseHeaders = response.headers();
for (int i = 0, size = responseHeaders.size(); i < size; i++) {
System.out.println(responseHeaders.name(i) + ": " + responseHeaders.value(i));
}
System.out.println(response.body().string());
}
});
}
1.0 初始化一個(gè)OkHttpClient對(duì)象
new OkHttpClient();
1.1. 新建一個(gè)對(duì)象，會(huì)先加載這個(gè)類，會(huì)初始化這個(gè)類的變量和靜態(tài)代碼塊
// 默認(rèn)支持的http協(xié)議版本 http 2.0,http 1.1
static final List<Protocol> DEFAULT_PROTOCOLS = Util.immutableList(
Protocol.HTTP_2, Protocol.HTTP_1_1);
// 默認(rèn)支持的 https,http
static final List<ConnectionSpec> DEFAULT_CONNECTION_SPECS = Util.immutableList(
ConnectionSpec.MODERN_TLS, ConnectionSpec.CLEARTEXT);
// //新建一個(gè)內(nèi)部使用的輔助類,基本上就是調(diào)用第一個(gè)參數(shù)的,同名方法,后面的參數(shù)作為方法的參數(shù)
static {
Internal.instance = new Internal() {
@Override public void addLenient(Headers.Builder builder, String line) {
builder.addLenient(line);
}
...
...
@Override public Call newWebSocketCall(OkHttpClient client, Request originalRequest) {
return new RealCall(client, originalRequest, true);
}
};
}
1.2 執(zhí)行OkHttpClient構(gòu)造函數(shù)
public OkHttpClient() {
this(new Builder());
}
1.3 初始化builder的成員變量，執(zhí)行builder的構(gòu)造函數(shù)
/* 這兩個(gè)攔截器主要是用來我們對(duì)請(qǐng)求過程進(jìn)行攔截處理,添加一些我們直接的處理,
如打印請(qǐng)求日志,為每個(gè)請(qǐng)求添加相同的請(qǐng)求頭字段,如 apikey 等
兩個(gè)區(qū)別在于interceptors是在和服務(wù)器未連接的時(shí)候攔截處理,
networkInterceptors是和服務(wù)器建立連接后添加的攔截處理
*/
final List<Interceptor> interceptors = new ArrayList<>(); // 攔截器
final List<Interceptor> networkInterceptors = new ArrayList<>(); // 網(wǎng)絡(luò)攔截器
@Nullable Cache cache; // 配置緩存的
@Nullable Proxy proxy; //配置代理服務(wù)器
SocketFactory socketFactory; // http 用于創(chuàng)建 Socket 對(duì)象
@Nullable SSLSocketFactory sslSocketFactory; // https 用于配置創(chuàng)建 SSLSocket 對(duì)象
public Builder() {
// 新建一個(gè) Dispatcher 對(duì)象,主要是用來構(gòu)建線程池分發(fā)請(qǐng)求的
dispatcher = new Dispatcher();
protocols = DEFAULT_PROTOCOLS; // 協(xié)議版本 http 2.0, http 1.1
connectionSpecs = DEFAULT_CONNECTION_SPECS; // 協(xié)議類型 https,http
/*
事件監(jiān)聽器 默認(rèn)為NONE,如果要對(duì)網(wǎng)絡(luò)請(qǐng)求過程(請(qǐng)求開始,連接 請(qǐng)求結(jié)束,
發(fā)送請(qǐng)求頭,發(fā)送請(qǐng)求體,dns解析)進(jìn)行監(jiān)聽,可以進(jìn)行自定義 EventListener
*/
eventListenerFactory = EventListener.factory(EventListener.NONE);
// 設(shè)置默認(rèn)代理選擇器,是 sun.net.spi.DefaultProxySelector
proxySelector = ProxySelector.getDefault();
// 處理 cookie
cookieJar = CookieJar.NO_COOKIES;
socketFactory = SocketFactory.getDefault(); // 設(shè)置默認(rèn)的 SocketFactory
hostnameVerifier = OkHostnameVerifier.INSTANCE; // 配置 https 檢測(cè)服務(wù)器返回的證書信息與域名匹配認(rèn)證
certificatePinner = CertificatePinner.DEFAULT; // 配置 https證書校驗(yàn)信息sha256
// 當(dāng)有設(shè)置代理服務(wù)器訪問時(shí),代理服務(wù)器需要賬號(hào)密碼認(rèn)證,默認(rèn)不啟用代理
proxyAuthenticator = Authenticator.NONE;
// http 認(rèn)證,有些服務(wù)器進(jìn)行訪問時(shí)需要賬號(hào)密碼的,如 basic認(rèn)證 默認(rèn)為空
authenticator = Authenticator.NONE;
// 緩存連接池
connectionPool = new ConnectionPool();
// DNS解析器,默認(rèn)使用系統(tǒng)的,可以自己自定義的,也可以使用阿里的HttpDns
dns = Dns.SYSTEM;
// https連接是否重定向
followSslRedirects = true;
// http 連接是否重定向
followRedirects = true;
// 連接失敗是否重試
retryOnConnectionFailure = true;
// 連接超時(shí)時(shí)間
connectTimeout = 10_000;
// 讀取服務(wù)器數(shù)據(jù)超時(shí)時(shí)間
readTimeout = 10_000;
// 發(fā)送給服務(wù)器數(shù)據(jù)超時(shí)時(shí)間
writeTimeout = 10_000;
// http 2.0 版本新增的功能(用于計(jì)算往返時(shí)間，執(zhí)行“ 活性” 檢活)。設(shè)置間隔多少毫秒執(zhí)行一次,為 0 默認(rèn)不啟用
pingInterval = 0;
}
1.4. Dispatcher對(duì)象的初始化，里面有個(gè)線程池，使用懶加載的形式，只有發(fā)生第一次請(qǐng)求的時(shí)候才會(huì)初始化線程池，使用線程池進(jìn)行管理發(fā)送的網(wǎng)絡(luò)請(qǐng)求的Call
// 線程池中最大同時(shí)請(qǐng)求個(gè)數(shù)，超過個(gè)數(shù)得進(jìn)行排隊(duì) 。可進(jìn)行設(shè)置
private int maxRequests = 64;
// 為避免線程池中請(qǐng)求中被同一個(gè)host的請(qǐng)求占滿了，同時(shí)也是減輕服務(wù)器壓力，可設(shè)置每個(gè)host的最大請(qǐng)求數(shù) ,若超過就會(huì)進(jìn)入排隊(duì)序列
private int maxRequestsPerHost = 5;
private @Nullable Runnable idleCallback;
private @Nullable ExecutorService executorService;
/* 準(zhǔn)備執(zhí)行的異常請(qǐng)求集合,由于設(shè)置了線程池中最大同時(shí)請(qǐng)求個(gè)數(shù),每個(gè)host的最大請(qǐng)求數(shù),
如果超過就會(huì)存起來,當(dāng)正在請(qǐng)求的 Call請(qǐng)求完成了,就會(huì)調(diào)度該集合里面的 Call 去請(qǐng)求
*/
private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
// 正在執(zhí)行的異步請(qǐng)求 Call集合
private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
// 正在執(zhí)行的同步請(qǐng)求 Call 集合,只做統(tǒng)計(jì)
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
// 單例模式,懶加載,線程池
public synchronized ExecutorService executorService() {
if (executorService == null) {
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}
1.5. ConnectionPool對(duì)象的初始化
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));
/** The maximum number of idle connections for each address. */
private final int maxIdleConnections;
private final long keepAliveDurationNs;
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) {
}
}
}
}
}
};
private final Deque<RealConnection> connections = new ArrayDeque<>();
final RouteDatabase routeDatabase = new RouteDatabase();
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);
}
}
1.6. 對(duì)應(yīng)1.2中OKHttpClient構(gòu)造函數(shù)調(diào)用，OkHttpClient對(duì)象已生成
OkHttpClient(Builder builder) {
this.dispatcher = builder.dispatcher;
this.proxy = builder.proxy;
.......
this.pingInterval = builder.pingInterval;
}
2.0 創(chuàng)建Request對(duì)象，包含一個(gè)網(wǎng)絡(luò)請(qǐng)求的信息，有請(qǐng)求頭，請(qǐng)求體，請(qǐng)求網(wǎng)址等
Request request = new Request.Builder()
.url(BASE_URL+"/weather?cityname="+"上海")
.build();
3.0 進(jìn)行網(wǎng)絡(luò)請(qǐng)求
mClient.newCall(request).enqueue(new Callback() {}）
3.1 創(chuàng)建RealCall對(duì)象
@Override public Call newCall(Request request) {
return new RealCall(this, request, false /* for web socket */);
}
RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
final EventListener.Factory eventListenerFactory = client.eventListenerFactory();
this.client = client;
this.originalRequest = originalRequest;
this.forWebSocket = forWebSocket;
this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
// TODO(jwilson): this is unsafe publication and not threadsafe.
this.eventListener = eventListenerFactory.create(this);
}
3.2 把請(qǐng)求發(fā)送到Dispatcher的線程池隊(duì)列
@Override public void enqueue(Callback responseCallback) {
synchronized (this) {
//新建的RealCall的executed 變量為false,執(zhí)行完為ture,這就是RealCall對(duì)象不能執(zhí)行兩次的原因
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
captureCallStackTrace();
// 創(chuàng)建AsyncCall對(duì)象加入到線程池
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
3.3 AsyncCall繼承NamedRunnable 實(shí)現(xiàn)了Runnable接口
final class AsyncCall extends NamedRunnable {
}
public abstract class NamedRunnable implements Runnable {
@Override public final void run() {
execute();
}
protected abstract void execute();
}
3.4 加入線程池后就會(huì)執(zhí)行在子線程中執(zhí)行AsyncCall 的execute的方法
synchronized void enqueue(AsyncCall call) {
// 1.4已說明
if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
runningAsyncCalls.add(call);
executorService().execute(call);
} else {
readyAsyncCalls.add(call);
}
}
3.5 execute 調(diào)用getResponseWithInterceptorChain方法獲取到Response對(duì)象,即服務(wù)器返回?cái)?shù)據(jù)
@Override protected void execute() {
Response response = getResponseWithInterceptorChain();
if (retryAndFollowUpInterceptor.isCanceled()) {
signalledCallback = true;
// 若請(qǐng)求失敗會(huì)回調(diào) 3.0 中設(shè)置Callback onFailure方法
responseCallback.onFailure(RealCall.this, new IOException("Canceled"));
} else {
signalledCallback = true;
// 若請(qǐng)求成功會(huì)回調(diào) 3.0 中設(shè)置 Callback onResponse方法
responseCallback.onResponse(RealCall.this, response);
}
finally {
client.dispatcher().finished(this);
}
}
3.6 getResponseWithInterceptorChain
// 把之前設(shè)置的自定義interceptors，RetryAndFollowUpInterceptor，BridgeInterceptor
//，CacheInterceptor，ConnectInterceptor，自定義的networkInterceptors，CallServerInterceptor對(duì)象
// 保存在interceptors里面作為參數(shù)進(jìn)行構(gòu)造RealInterceptorChain
Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
List<Interceptor> interceptors = new ArrayList<>();
// addInterceptor
interceptors.addAll(client.interceptors());
interceptors.add(retryAndFollowUpInterceptor);
interceptors.add(new BridgeInterceptor(client.cookieJar()));
interceptors.add(new CacheInterceptor(client.internalCache()));
interceptors.add(new ConnectInterceptor(client));
// addNetworkInterceptor
if (!forWebSocket) {
interceptors.addAll(client.networkInterceptors());
}
interceptors.add(new CallServerInterceptor(forWebSocket));
Interceptor.Chain chain = new RealInterceptorChain(
interceptors, null, null, null, 0, originalRequest);
return chain.proceed(originalRequest);
}
3.7. interceptors 保存很多攔截器
OkHttpClient Builder有兩個(gè)添加Interceptor 兩者處于不通的調(diào)用階段，我們可以在網(wǎng)絡(luò)請(qǐng)求的這兩個(gè)階段自定義攔截器，做不同的處理如上篇博文提到的HttpLoggingInterceptor HttpCacheInterceptor等
public Builder addInterceptor(Interceptor interceptor) {
interceptors.add(interceptor);
return this;
}
public Builder addNetworkInterceptor(Interceptor interceptor) {
networkInterceptors.add(interceptor);
return this;
}
Interceptor 和 Chain 接口都比較簡單
public interface Interceptor {
Response intercept(Chain chain) throws IOException;
interface Chain {
Request request();
Response proceed(Request request) throws IOException;
/**
* Returns the connection the request will be executed on. This is only available in the chains
* of network interceptors; for application interceptors this is always null.
*/
@Nullable Connection connection();
}
}
3.8 創(chuàng)建RealInterceptorChain
public RealInterceptorChain(List<Interceptor> interceptors, StreamAllocation streamAllocation,
HttpCodec httpCodec, RealConnection connection, int index, Request request) {
this.interceptors = interceptors;
this.connection = connection;
this.streamAllocation = streamAllocation;
this.httpCodec = httpCodec;
this.index = index;
this.request = request;
}
3.9 調(diào)用proceed方法
- proceed方法先進(jìn)行了數(shù)據(jù)的校驗(yàn)，數(shù)據(jù)如果不正確，拋出異常
- proceed每次都會(huì)用之前構(gòu)造的這個(gè)RealInterceptorChain 的interceptors, streamAllocation, httpCodec, connection, index + 1, request 或者傳入單個(gè)或者多個(gè)新的值去替換部分原來的值 從而去構(gòu)建新的RealInterceptorChain對(duì)象，然后獲取到第index個(gè)Interceptor,調(diào)用Interceptor的intercept方法，并把新建的RealInterceptorChain對(duì)象作為參數(shù)傳入
- 然后Interceptor的intercept方法又會(huì)調(diào)用上部分傳入的RealInterceptorChain對(duì)象的proceed方法 這就成了一個(gè)循環(huán)直到最后一個(gè)CallServerInterceptor，沒有調(diào)用 RealInterceptorChain的proceed方法跳出循環(huán)返回服務(wù)器返回的Response，請(qǐng)求完成
這個(gè)就使得，interceptors中的上一個(gè)Interceptor的intercept方法依賴下一個(gè)Interceptor的intercept的返回結(jié)果，依次調(diào)用完interceptors中的所有Interceptor的intercept方法。
@Override public Response proceed(Request request) throws IOException {
return proceed(request, streamAllocation, httpCodec, connection);
}
public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
RealConnection connection) throws IOException {
// Call the next interceptor in the chain.
RealInterceptorChain next = new RealInterceptorChain(
interceptors, streamAllocation, httpCodec, connection, index + 1, request);
Interceptor interceptor = interceptors.get(index);
Response response = interceptor.intercept(next);
return response;
}
所以一個(gè)正常的請(qǐng)求會(huì)依次調(diào)用下列Interceptor的intercept方法得到返回Response
4.0 RetryAndFollowUpInterceptor
4.1 在RealCall構(gòu)造函數(shù)中新建RetryAndFollowUpInterceptor 對(duì)象
// 見 3.1
this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
// 見 3.6
interceptors.add(retryAndFollowUpInterceptor);
4.2 若未添加自定義Intercept,這是第一個(gè)調(diào)用的Intercept的intercept
==followUpRequest方法和HTTP協(xié)議相關(guān)是對(duì)HTTP協(xié)議的實(shí)現(xiàn)==
@Override public Response intercept(Chain chain) throws IOException {
Request request = chain.request();
// 新建 StreamAllocation對(duì)象
streamAllocation = new StreamAllocation(
client.connectionPool(), createAddress(request.url()), callStackTrace);
int followUpCount = 0;
Response priorResponse = null;
// 無限循環(huán)，可以多次重試請(qǐng)求，followUpCount請(qǐng)求次數(shù)大于MAX_FOLLOW_UPS（20）會(huì)拋出ProtocolException，跳出循環(huán)
while (true) {
//進(jìn)行取消請(qǐng)求，跳出循環(huán)就是由這個(gè)標(biāo)志位控制的
if (canceled) {
streamAllocation.release();
throw new IOException("Canceled");
}
Response response = null;
boolean releaseConnection = true;
// 把streamAllocation對(duì)象傳入下一個(gè) BridgeInterceptor
response = ((RealInterceptorChain) chain).proceed(request, streamAllocation, null, null);
releaseConnection = false;
...
// Attach the prior response if it exists. Such responses never have a body.
// 有些請(qǐng)求可能要多次和服務(wù)器進(jìn)行通信才能完成,如代理，認(rèn)證等，若之前有先前有返回響應(yīng)體，存儲(chǔ)到Response的priorResponse中
if (priorResponse != null) {
response = response.newBuilder()
.priorResponse(priorResponse.newBuilder()
.body(null)
.build())
.build();
}
// 根據(jù)服務(wù)器返回代碼和HTTP協(xié)議格式，判斷是否需要多次和服務(wù)器進(jìn)行通信
Request followUp = followUpRequest(response);
// 若不需要，直接返回請(qǐng)求結(jié)果，循環(huán)結(jié)束
if (followUp == null) {
if (!forWebSocket) {
streamAllocation.release();
}
return response;
}
// 關(guān)閉這個(gè)返回體的流，進(jìn)行判斷重試次數(shù)，及異常檢查 繼續(xù)進(jìn)行下一次循環(huán)
...
closeQuietly(response.body());
if (++followUpCount > MAX_FOLLOW_UPS) {
streamAllocation.release();
throw new ProtocolException("Too many follow-up requests: " + followUpCount);
}
request = followUp;
priorResponse = response;
}
}
4.3 新建Address對(duì)象
private Address createAddress(HttpUrl url) {
SSLSocketFactory sslSocketFactory = null;
HostnameVerifier hostnameVerifier = null;
CertificatePinner certificatePinner = null;
if (url.isHttps()) {
sslSocketFactory = client.sslSocketFactory();
hostnameVerifier = client.hostnameVerifier();
certificatePinner = client.certificatePinner();
}
return new Address(url.host(), url.port(), client.dns(), client.socketFactory(),
sslSocketFactory, hostnameVerifier, certificatePinner, client.proxyAuthenticator(),
client.proxy(), client.protocols(), client.connectionSpecs(), client.proxySelector());
}
4.4 新建StreamAllocation對(duì)象
public StreamAllocation(ConnectionPool connectionPool, Address address, Object callStackTrace) {
this.connectionPool = connectionPool;
this.address = address;
this.routeSelector = new RouteSelector(address, routeDatabase());
this.callStackTrace = callStackTrace;
}
4.5 構(gòu)建RouteSelector
public RouteSelector(Address address, RouteDatabase routeDatabase) {
this.address = address;
this.routeDatabase = routeDatabase;
resetNextProxy(address.url(), address.proxy());
}
/** Prepares the proxy servers to try. */
private void resetNextProxy(HttpUrl url, Proxy proxy) {
if (proxy != null) {
// If the user specifies a proxy, try that and only that.
proxies = Collections.singletonList(proxy);
} else {
// Try each of the ProxySelector choices until one connection succeeds.
List<Proxy> proxiesOrNull = address.proxySelector().select(url.uri());
proxies = proxiesOrNull != null && !proxiesOrNull.isEmpty()
? Util.immutableList(proxiesOrNull)
: Util.immutableList(Proxy.NO_PROXY);
}
nextProxyIndex = 0;
}
5.0 BridgeInterceptor
5.1
// 見3.6
interceptors.add(new BridgeInterceptor(client.cookieJar()));
public BridgeInterceptor(CookieJar cookieJar) {
this.cookieJar = cookieJar;
}
5.2 構(gòu)建完整的請(qǐng)求，把一些默認(rèn)的請(qǐng)求頭添加上去，默認(rèn) Keep-Alive ，對(duì)請(qǐng)求進(jìn)行g(shù)zip壓縮傳輸，用okio庫中GzipSource進(jìn)行解壓
@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 (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());
}
// 把構(gòu)建好的新的請(qǐng)求傳入CacheInterceptor替換原來的Request對(duì)象
Response networkResponse = chain.proceed(requestBuilder.build());
HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());
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);
responseBuilder.body(new RealResponseBody(strippedHeaders, Okio.buffer(responseBody)));
}
return responseBuilder.build();
}
6.0 CacheInterceptor
6.1 Cache的使用見 OkHttp使用介紹
中Cache-Control 字段
// 見3.6
interceptors.add(new CacheInterceptor(client.internalCache()));
// 默認(rèn)情況下 Okhttp是沒有設(shè)置cache和internalCache的
InternalCache internalCache() {
return cache != null ? cache.internalCache : internalCache;
}
public CacheInterceptor(InternalCache cache) {
this.cache = cache;
}
6.2 我們考慮設(shè)置了Cache的情況
==Cache的情況也是HTTP協(xié)議相關(guān)是對(duì)HTTP協(xié)議的實(shí)現(xiàn)==
使用的是DiskLruCache作為緩存，磁盤緩存庫
@Override public Response intercept(Chain chain) throws IOException {
Response cacheCandidate = cache != null
// 6.3
? cache.get(chain.request())
: null;
long now = System.currentTimeMillis();
// 6.4
CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();
Request networkRequest = strategy.networkRequest;
Response cacheResponse = strategy.cacheResponse;
// If we don't need the network, we're done.
if (networkRequest == null) {
return cacheResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.build();
}
// ...
Response networkResponse = null;
try {
// 若沒有緩存，進(jìn)入ConnectInterceptor的intercept方法
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());
}
}
// ...
return response;
}
6.3.1
@Override public Response get(Request request) throws IOException {
return Cache.this.get(request);
}
6.3.2
@Nullable Response get(Request request) {
String key = key(request.url());
DiskLruCache.Snapshot snapshot;
Entry entry;
try {
snapshot = cache.get(key);
if (snapshot == null) {
return null;
}
} catch (IOException e) {
// Give up because the cache cannot be read.
return null;
}
try {
entry = new Entry(snapshot.getSource(ENTRY_METADATA));
} catch (IOException e) {
Util.closeQuietly(snapshot);
return null;
}
Response response = entry.response(snapshot);
if (!entry.matches(request, response)) {
Util.closeQuietly(response.body());
return null;
}
return response;
}
6.3.3
public Response response(DiskLruCache.Snapshot snapshot) {
String contentType = responseHeaders.get("Content-Type");
String contentLength = responseHeaders.get("Content-Length");
Request cacheRequest = new Request.Builder()
.url(url)
.method(requestMethod, null)
.headers(varyHeaders)
.build();
return new Response.Builder()
.request(cacheRequest)
.protocol(protocol)
.code(code)
.message(message)
.headers(responseHeaders)
.body(new CacheResponseBody(snapshot, contentType, contentLength))
.handshake(handshake)
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(receivedResponseMillis)
.build();
}
}
6.4.1
public Factory(long nowMillis, Request request, Response cacheResponse) {
this.nowMillis = nowMillis;
this.request = request;
this.cacheResponse = cacheResponse;
if (cacheResponse != null) {
this.sentRequestMillis = cacheResponse.sentRequestAtMillis();
this.receivedResponseMillis = cacheResponse.receivedResponseAtMillis();
Headers headers = cacheResponse.headers();
for (int i = 0, size = headers.size(); i < size; i++) {
String fieldName = headers.name(i);
String value = headers.value(i);
if ("Date".equalsIgnoreCase(fieldName)) {
servedDate = HttpDate.parse(value);
servedDateString = value;
} else if ("Expires".equalsIgnoreCase(fieldName)) {
expires = HttpDate.parse(value);
} else if ("Last-Modified".equalsIgnoreCase(fieldName)) {
lastModified = HttpDate.parse(value);
lastModifiedString = value;
} else if ("ETag".equalsIgnoreCase(fieldName)) {
etag = value;
} else if ("Age".equalsIgnoreCase(fieldName)) {
ageSeconds = HttpHeaders.parseSeconds(value, -1);
}
}
}
}
6.4.2
public CacheStrategy get() {
CacheStrategy candidate = getCandidate();
if (candidate.networkRequest != null && request.cacheControl().onlyIfCached()) {
// We're forbidden from using the network and the cache is insufficient.
return new CacheStrategy(null, null);
}
return candidate;
}
private CacheStrategy getCandidate() {
// No cached response.
if (cacheResponse == null) {
return new CacheStrategy(request, null);
}
// Drop the cached response if it's missing a required handshake.
if (request.isHttps() && cacheResponse.handshake() == null) {
return new CacheStrategy(request, null);
}
// If this response shouldn't have been stored, it should never be used
// as a response source. This check should be redundant as long as the
// persistence store is well-behaved and the rules are constant.
if (!isCacheable(cacheResponse, request)) {
return new CacheStrategy(request, null);
}
CacheControl requestCaching = request.cacheControl();
if (requestCaching.noCache() || hasConditions(request)) {
return new CacheStrategy(request, null);
}
long ageMillis = cacheResponseAge();
long freshMillis = computeFreshnessLifetime();
if (requestCaching.maxAgeSeconds() != -1) {
freshMillis = Math.min(freshMillis, SECONDS.toMillis(requestCaching.maxAgeSeconds()));
}
long minFreshMillis = 0;
if (requestCaching.minFreshSeconds() != -1) {
minFreshMillis = SECONDS.toMillis(requestCaching.minFreshSeconds());
}
long maxStaleMillis = 0;
CacheControl responseCaching = cacheResponse.cacheControl();
if (!responseCaching.mustRevalidate() && requestCaching.maxStaleSeconds() != -1) {
maxStaleMillis = SECONDS.toMillis(requestCaching.maxStaleSeconds());
}
if (!responseCaching.noCache() && ageMillis + minFreshMillis < freshMillis + maxStaleMillis) {
Response.Builder builder = cacheResponse.newBuilder();
if (ageMillis + minFreshMillis >= freshMillis) {
builder.addHeader("Warning", "110 HttpURLConnection \"Response is stale\"");
}
long oneDayMillis = 24 * 60 * 60 * 1000L;
if (ageMillis > oneDayMillis && isFreshnessLifetimeHeuristic()) {
builder.addHeader("Warning", "113 HttpURLConnection \"Heuristic expiration\"");
}
return new CacheStrategy(null, builder.build());
}
// Find a condition to add to the request. If the condition is satisfied, the response body
// will not be transmitted.
String conditionName;
String conditionValue;
if (etag != null) {
conditionName = "If-None-Match";
conditionValue = etag;
} else if (lastModified != null) {
conditionName = "If-Modified-Since";
conditionValue = lastModifiedString;
} else if (servedDate != null) {
conditionName = "If-Modified-Since";
conditionValue = servedDateString;
} else {
return new CacheStrategy(request, null); // No condition! Make a regular request.
}
Headers.Builder conditionalRequestHeaders = request.headers().newBuilder();
Internal.instance.addLenient(conditionalRequestHeaders, conditionName, conditionValue);
Request conditionalRequest = request.newBuilder()
.headers(conditionalRequestHeaders.build())
.build();
return new CacheStrategy(conditionalRequest, cacheResponse);
}
7.0 ConnectInterceptor
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Request request = realChain.request();
// 見4.2 獲取新建的StreamAllocation對(duì)象
StreamAllocation streamAllocation = realChain.streamAllocation();
// We need the network to satisfy this request. Possibly for validating a conditional GET.
boolean doExtensiveHealthChecks = !request.method().equals("GET");
HttpCodec httpCodec = streamAllocation.newStream(client, doExtensiveHealthChecks);
RealConnection connection = streamAllocation.connection();
// 把request, streamAllocation, httpCodec, connection傳入
//進(jìn)入CallServerInterceptor的intercept
return realChain.proceed(request, streamAllocation, httpCodec, connection);
}
public HttpCodec newStream(OkHttpClient client, boolean doExtensiveHealthChecks) {
// ...
// 7.1.0
RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
writeTimeout, connectionRetryEnabled, doExtensiveHealthChecks);
// 7.2.0
HttpCodec resultCodec = resultConnection.newCodec(client, this);
synchronized (connectionPool) {
codec = resultCodec;
return resultCodec;
}
7.1.1
private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
int writeTimeout, boolean connectionRetryEnabled, boolean doExtensiveHealthChecks)
throws IOException {
while (true) {
RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
connectionRetryEnabled);
.....
return candidate;
}
}
7.1.2
private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
boolean connectionRetryEnabled) throws IOException {
Route selectedRoute;
synchronized (connectionPool) {
...
// Attempt to use an already-allocated connection.
// 嘗試使用自身存儲(chǔ)的連接
RealConnection allocatedConnection = this.connection;
if (allocatedConnection != null && !allocatedConnection.noNewStreams) {
return allocatedConnection;
}
// Attempt to get a connection from the pool.
//嘗試從連接池中獲取
Internal.instance.get(connectionPool, address, this, null);
if (connection != null) {
return connection;
}
selectedRoute = route;
}
// If we need a route, make one. This is a blocking operation.
if (selectedRoute == null) {
selectedRoute = routeSelector.next();
}
RealConnection result;
synchronized (connectionPool) {
if (canceled) throw new IOException("Canceled");
// Now that we have an IP address, make another attempt at getting a connection from the pool.
// This could match due to connection coalescing.
//有了selectedRoute后再次從連接池中獲取連接
Internal.instance.get(connectionPool, address, this, selectedRoute);
if (connection != null) {
route = selectedRoute;
return connection;
}
// 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.
route = selectedRoute;
refusedStreamCount = 0;
// 新建連接，并把這個(gè)RealConnection對(duì)象和StreamAllocation對(duì)象綁定
result = new RealConnection(connectionPool, selectedRoute);
acquire(result);
}
// Do TCP + TLS handshakes. This is a blocking operation.
result.connect(connectTimeout, readTimeout, writeTimeout, connectionRetryEnabled);
Socket socket = null;
synchronized (connectionPool) {
// Pool the connection.
// 把新建的RealConnection對(duì)象，放入到連接池
Internal.instance.put(connectionPool, result);
// If another multiplexed connection to the same address was created concurrently, then
// release this connection and acquire that one.
if (result.isMultiplexed()) {
socket = Internal.instance.deduplicate(connectionPool, address, this);
result = connection;
}
}
closeQuietly(socket);
return result;
}
7.1.3
public void connect(
int connectTimeout, int readTimeout, int writeTimeout, boolean connectionRetryEnabled) {
// ....
while (true) {
try {
// 若為https
if (route.requiresTunnel()) {
connectTunnel(connectTimeout, readTimeout, writeTimeout);
} else {
//如果為http
// 7.1.3.1.0
connectSocket(connectTimeout, readTimeout);
}
// 7.1.3.2.0
establishProtocol(connectionSpecSelector);
break;
}
7.1.3.1.1
==和服務(wù)建立socket連接，source，sink是Okio庫的IO讀寫類==,它補(bǔ)充了java.io和java.nio的不足，以便能夠更加方便，快速的訪問、存儲(chǔ)和處理你的數(shù)據(jù).
private void connectSocket(int connectTimeout, int readTimeout) throws IOException {
Proxy proxy = route.proxy();
Address address = route.address();
rawSocket = proxy.type() == Proxy.Type.DIRECT || proxy.type() == Proxy.Type.HTTP
? address.socketFactory().createSocket() //創(chuàng)建 Socket
: new Socket(proxy);
rawSocket.setSoTimeout(readTimeout);
try {
Platform.get().connectSocket(rawSocket, route.socketAddress(), connectTimeout); //和服務(wù)器建立連接
} catch (ConnectException e) {
// ...
}
// ...
source = Okio.buffer(Okio.source(rawSocket));
sink = Okio.buffer(Okio.sink(rawSocket));
// ...
}
7.1.3.1.2
public Socket createSocket() {
return new Socket();
}
public void connectSocket(Socket socket, InetSocketAddress address,
int connectTimeout) throws IOException {
socket.connect(address, connectTimeout);
}
7.1.3.2.2
private void establishProtocol(ConnectionSpecSelector connectionSpecSelector) throws IOException {
if (route.address().sslSocketFactory() == null) {
protocol = Protocol.HTTP_1_1;
// 普通http,socket連接
socket = rawSocket;
return;
}
// 構(gòu)建https連接，暫不分析
connectTls(connectionSpecSelector);
if (protocol == Protocol.HTTP_2) {
socket.setSoTimeout(0); // HTTP/2 connection timeouts are set per-stream.
http2Connection = new Http2Connection.Builder(true)
.socket(socket, route.address().url().host(), source, sink)
.listener(this)
.build();
http2Connection.start();
}
}
7.2 構(gòu)建Http1Codec對(duì)象
public HttpCodec newCodec(
OkHttpClient client, StreamAllocation streamAllocation) throws SocketException {
// 若為http，http2Connection為null
if (http2Connection != null) {
return new Http2Codec(client, streamAllocation, http2Connection);
} else {
socket.setSoTimeout(client.readTimeoutMillis());
source.timeout().timeout(client.readTimeoutMillis(), MILLISECONDS);
sink.timeout().timeout(client.writeTimeoutMillis(), MILLISECONDS);
// 把source, sink 負(fù)責(zé)和服務(wù)器IO讀寫流傳入
return new Http1Codec(client, streamAllocation, source, sink);
}
}
8.0 CallServerInterceptor
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
// ...
// 8.1
httpCodec.writeRequestHeaders(request);
Response.Builder responseBuilder = null;
// 8.2 向服務(wù)器傳輸請(qǐng)求體
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();
responseBuilder = httpCodec.readResponseHeaders(true);
}
if (responseBuilder == null) {
// Write the request body if the "Expect: 100-continue" expectation was met.
Sink requestBodyOut = httpCodec.createRequestBody(request, request.body().contentLength());
BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
request.body().writeTo(bufferedRequestBody);
bufferedRequestBody.close();
} 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();
}
}
// ... flush()請(qǐng)求
httpCodec.finishRequest();
// 8.3
if (responseBuilder == null) {
responseBuilder = httpCodec.readResponseHeaders(false);
}
// ...
int code = response.code();
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 {
// 8.4
response = response.newBuilder()
.body(httpCodec.openResponseBody(response))
.build();
}
//...
return response;
}
==HTTP請(qǐng)求格式==
image
8.1 向服務(wù)器傳輸請(qǐng)求頭
@Override public void writeRequestHeaders(Request request) throws IOException {
String requestLine = RequestLine.get(
request, streamAllocation.connection().route().proxy().type());
writeRequest(request.headers(), requestLine);
}
public static String get(Request request, Proxy.Type proxyType) {
StringBuilder result = new StringBuilder();
result.append(request.method());
result.append(' ');
if (includeAuthorityInRequestLine(request, proxyType)) {
result.append(request.url());
} else {
result.append(requestPath(request.url()));
}
result.append(" HTTP/1.1");
return result.toString();
}
public void writeRequest(Headers headers, String requestLine) throws IOException {
if (state != STATE_IDLE) throw new IllegalStateException("state: " + state);
sink.writeUtf8(requestLine).writeUtf8("\r\n");
for (int i = 0, size = headers.size(); i < size; i++) {
sink.writeUtf8(headers.name(i))
.writeUtf8(": ")
.writeUtf8(headers.value(i))
.writeUtf8("\r\n");
}
sink.writeUtf8("\r\n");
state = STATE_OPEN_REQUEST_BODY;
}
8.3 獲取響應(yīng)頭
@Override public Response.Builder readResponseHeaders(boolean expectContinue) throws IOException {
if (state != STATE_OPEN_REQUEST_BODY && state != STATE_READ_RESPONSE_HEADERS) {
throw new IllegalStateException("state: " + state);
}
try {
StatusLine statusLine = StatusLine.parse(source.readUtf8LineStrict());
Response.Builder responseBuilder = new Response.Builder()
.protocol(statusLine.protocol)
.code(statusLine.code)
.message(statusLine.message)
.headers(readHeaders());
if (expectContinue && statusLine.code == HTTP_CONTINUE) {
return null;
}
state = STATE_OPEN_RESPONSE_BODY;
return responseBuilder;
} catch (EOFException e) {
// Provide more context if the server ends the stream before sending a response.
IOException exception = new IOException("unexpected end of stream on " + streamAllocation);
exception.initCause(e);
throw exception;
}
}
8.4 獲取響應(yīng)體
@Override public ResponseBody openResponseBody(Response response) throws IOException {
Source source = getTransferStream(response);
return new RealResponseBody(response.headers(), Okio.buffer(source));
}
private Source getTransferStream(Response response) throws IOException {
if (!HttpHeaders.hasBody(response)) {
return newFixedLengthSource(0);
}
if ("chunked".equalsIgnoreCase(response.header("Transfer-Encoding"))) {
return newChunkedSource(response.request().url());
}
long contentLength = HttpHeaders.contentLength(response);
if (contentLength != -1) {
return newFixedLengthSource(contentLength);
}
// Wrap the input stream from the connection (rather than just returning
// "socketIn" directly here), so that we can control its use after the
// reference escapes.
return newUnknownLengthSource();
}
9. 整個(gè)網(wǎng)絡(luò)請(qǐng)求過程已經(jīng)完成
見3.5 Responsed對(duì)象返回回調(diào)Callback
// 不管請(qǐng)求成功還是失敗都會(huì)調(diào)用finished方法
finally {
client.dispatcher().finished(this);
}
void finished(AsyncCall call) {
finished(runningAsyncCalls, call, true);
}
private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
...
...
if (promoteCalls) promoteCalls();
}
// 每次運(yùn)行完一個(gè)AsyncCall如果運(yùn)行的個(gè)數(shù)不超過限制，就添加一個(gè)請(qǐng)求進(jìn)入線程池
private void promoteCalls() {
if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity.
if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote.
for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
AsyncCall call = i.next();
if (runningCallsForHost(call) < maxRequestsPerHost) {
i.remove();
runningAsyncCalls.add(call);
executorService().execute(call);
}
if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
}
}
