OkHttp3實現原理分析(二)
概述
前言:前一節https://mp.csdn.net/postedit/84941253,總結了一下OkHttp3的簡單使用教程。在專案中使用了這個網路框架,在看完基本的原始碼之後,還是想總結一下OkHttp的實現流程。在學習框架的過程中,從使用方法出發,首先是怎麼使用,其次是我們使用的功能在內部是如何實現的,實現方案上有什麼技巧,有什麼正規化。
OkHttp的整體流程
整個流程是:通過OkHttpClient將構建的Request轉換為Call,然後在RealCall中進行非同步或同步任務,最後通過一些的攔截器interceptor發出網路請求和得到返回的response
Okhttp3整體流程.png
拆元件
在整體流程中,主要的元件是OkHttpClient,其次有Call,RealCall,Disptcher,各種Interceptors,Request和Response元件。Request和Response已經在上一篇的對其結構原始碼進行了分析。
1. OkHttpClient物件:網路請求的主要操控者
建立OkHttpClient物件
//通過Builder構造OkHttpClient OkHttpClient.Builder builder = new OkHttpClient.Builder() .connectTimeout(20, TimeUnit.SECONDS) .writeTimeout(20, TimeUnit.SECONDS) .readTimeout(20, TimeUnit.SECONDS); return builder.build();
OkHttpClient.Builder
類有很多變數,OkHttpClient有很多的成員變數:
final Dispatcher dispatcher; //重要:分發器,分發執行和關閉由request構成的Call final Proxy proxy; //代理 final List<Protocol> protocols; //協議 final List<ConnectionSpec> connectionSpecs; //傳輸層版本和連線協議 final List<Interceptor> interceptors; //重要:攔截器 final List<Interceptor> networkInterceptors; //網路攔截器 final ProxySelector proxySelector; //代理選擇 final CookieJar cookieJar; //cookie final Cache cache; //快取 final InternalCache internalCache; //內部快取 final SocketFactory socketFactory; //socket 工廠 final SSLSocketFactory sslSocketFactory; //安全套接層socket 工廠,用於HTTPS final CertificateChainCleaner certificateChainCleaner; // 驗證確認響應證書 適用 HTTPS 請求連線的主機名。 final HostnameVerifier hostnameVerifier; // 主機名字確認 final CertificatePinner certificatePinner; // 證書鏈 final Authenticator proxyAuthenticator; //代理身份驗證 final Authenticator authenticator; // 本地身份驗證 final ConnectionPool connectionPool; //連線池,複用連線 final Dns dns; //域名 final boolean followSslRedirects; //安全套接層重定向 final boolean followRedirects; //本地重定向 final boolean retryOnConnectionFailure; //重試連線失敗 final int connectTimeout; //連線超時 final int readTimeout; //read 超時 final int writeTimeout; //write 超時
OkHttpClient完成整個請求設計到很多引數,都可以通過OkHttpClient.builder使用建立者模式構建。事實上,你能夠通過它來設定改變一些引數,因為他是通過建造者模式實現的,因此你可以通過builder()來設定。如果不進行設定,在Builder中就會使用預設的設定:
public Builder() {
dispatcher = new Dispatcher();
protocols = DEFAULT_PROTOCOLS;
connectionSpecs = DEFAULT_CONNECTION_SPECS;
eventListenerFactory = EventListener.factory(EventListener.NONE);
proxySelector = ProxySelector.getDefault();
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;
connectTimeout = 10_000;
readTimeout = 10_000;
writeTimeout = 10_000;
pingInterval = 0;
}
2,RealCall:真正的請求執行者
2.1之前文章中的Http發起同步請求的程式碼:
Request request = new Request.Builder()
.url(url)
.build();
Response response = client.newCall(request).execute();
client.newCall(request).execute()
建立了Call執行了網路請求獲得response響應。重點看一看這個執行的請求者的內部是什麼鬼。
/**
* Prepares the {@code request} to be executed at some point in the future.
*/
//OkHttpClient中的方法,可以看出RealCall的真面目
@Override public Call newCall(Request request) {
return RealCall.newRealCall(this, request, false /* for web socket */);
}
RealCall的建構函式:
private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
//client請求
this.client = client;
//我們構造的請求
this.originalRequest = originalRequest;
this.forWebSocket = forWebSocket;
//負責重試和重定向攔截器
this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
}
Call
物件其實是一個介面,Call的原始碼:
public interface Call extends Cloneable {
/** Returns the original request that initiated this call. */
//用於返回Call物件中的request物件
Request request();
//用於執行同步請求的方法
Response execute() throws IOException;
//用於執行非同步請求的方法,通過responseCallback回撥結果
void enqueue(Callback responseCallback);
/** Cancels the request, if possible. Requests that are already complete cannot be canceled. */
//取消這個call,當call被取消時請求不在執行,丟擲異常。可以用於終止請求
void cancel();
/**
* Returns true if this call has been either {@linkplain #execute() executed} or {@linkplain
* #enqueue(Callback) enqueued}. It is an error to execute a call more than once.
*/
//是否被執行
boolean isExecuted();
//是否被取消
boolean isCanceled();
/**
* Create a new, identical call to this one which can be enqueued or executed even if this call
* has already been.
*/
Call clone();
interface Factory {
Call newCall(Request request);
}
}
Realcall
是Call的實現類。顯然重要的執行任務就交個RealCall物件execute()和enqueue(Callback responseCallback)方法了。
我們首先看 RealCall#execute:
@Override public Response execute() throws IOException {
//(1)
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
captureCallStackTrace();
//事件監聽器調
eventListener.callStart(this);
try {
//(2)
client.dispatcher().executed(this);
//(3)
Response result = getResponseWithInterceptorChain();
if (result == null) throw new IOException("Canceled");
return result;
} catch (IOException e) {
eventListener.callFailed(this, e);
throw e;
} finally {
//(4)
client.dispatcher().finished(this);
}
}
(1)檢查這個 call 是否已經被執行了,每個 call 只能被執行一次,如果想要一個完全一樣的 call,可以利用 call#clone 方法進行克隆。
(2)利用 client.dispatcher().executed(this) 來進行實際執行,分發器負責分發。dispatcher 是剛才看到的 OkHttpClient.Builder 的成員之一,它的文件說自己是非同步 HTTP 請求的執行策略,現在看來,同步請求它也有摻和。
(3)呼叫 getResponseWithInterceptorChain() 函式獲取 HTTP 返回結果,從函式名可以看出,這一步還會進行一系列“攔截”操作。
(4)最後還要通知 dispatcher 自己已經執行完畢
dispatcher 這裡我們不過度關注,在同步執行的流程中,涉及到 dispatcher 的內容只不過是告知它我們的執行狀態,比如開始執行了(呼叫 executed),比如執行完畢了(呼叫 finished),在非同步執行流程中它會有更多的參與。
Dispatcher的原始碼:主要在非同步請求時參與多,這裡有執行非同步請求的執行緒池
/**
* Policy on when async requests are executed.
*
* <p>Each dispatcher uses an {@link ExecutorService} to run calls internally. If you supply your
* own executor, it should be able to run {@linkplain #getMaxRequests the configured maximum} number
* of calls concurrently.
*/
//請求分發器,**主要在非同步請求時參與多,這裡有執行非同步請求的執行緒池**
public final class Dispatcher {
//最大的請求數量
private int maxRequests = 64;
//每個主機的請求數量,預設在摸個主機上同時請求5個
private int maxRequestsPerHost = 5;
private @Nullable Runnable idleCallback;
/** Executes calls. Created lazily. */
//執行非同步call時的執行緒池,就在這兒
private @Nullable ExecutorService executorService;
/** Ready async calls in the order they'll be run. */
//即將被執行的非同步call佇列
private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
/** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
//正在執行的非同步call,包括被取消的還沒有完成的
private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
/** Running synchronous calls. Includes canceled calls that haven't finished yet. */
//正在執行的同步call。包括被取消的還沒有完成的
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
//可以執行自定義執行緒池,傳進來
public Dispatcher(ExecutorService executorService) {
this.executorService = executorService;
}
public Dispatcher() {
}
//構造執行緒池
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;
}
/**
* Set the maximum number of requests to execute concurrently. Above this requests queue in
* memory, waiting for the running calls to complete.
*
* <p>If more than {@code maxRequests} requests are in flight when this is invoked, those requests
* will remain in flight.
*/
public synchronized void setMaxRequests(int maxRequests) {
if (maxRequests < 1) {
throw new IllegalArgumentException("max < 1: " + maxRequests);
}
this.maxRequests = maxRequests;
promoteCalls();
}
public synchronized int getMaxRequests() {
return maxRequests;
}
/**
* Set the maximum number of requests for each host to execute concurrently. This limits requests
* by the URL's host name. Note that concurrent requests to a single IP address may still exceed
* this limit: multiple hostnames may share an IP address or be routed through the same HTTP
* proxy.
*
* <p>If more than {@code maxRequestsPerHost} requests are in flight when this is invoked, those
* requests will remain in flight.
*/
public synchronized void setMaxRequestsPerHost(int maxRequestsPerHost) {
if (maxRequestsPerHost < 1) {
throw new IllegalArgumentException("max < 1: " + maxRequestsPerHost);
}
this.maxRequestsPerHost = maxRequestsPerHost;
promoteCalls();
}
public synchronized int getMaxRequestsPerHost() {
return maxRequestsPerHost;
}
/**
* Set a callback to be invoked each time the dispatcher becomes idle (when the number of running
* calls returns to zero).
*
* <p>Note: The time at which a {@linkplain Call call} is considered idle is different depending
* on whether it was run {@linkplain Call#enqueue(Callback) asynchronously} or
* {@linkplain Call#execute() synchronously}. Asynchronous calls become idle after the
* {@link Callback#onResponse onResponse} or {@link Callback#onFailure onFailure} callback has
* returned. Synchronous calls become idle once {@link Call#execute() execute()} returns. This
* means that if you are doing synchronous calls the network layer will not truly be idle until
* every returned {@link Response} has been closed.
*/
public synchronized void setIdleCallback(@Nullable Runnable idleCallback) {
this.idleCallback = idleCallback;
}
//分發非同步執行的call,是提交到執行緒池
synchronized void enqueue(AsyncCall call) {
if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
runningAsyncCalls.add(call);
//提交到執行緒此執行
executorService().execute(call);
} else {
readyAsyncCalls.add(call);
}
}
/**
* Cancel all calls currently enqueued or executing. Includes calls executed both {@linkplain
* Call#execute() synchronously} and {@linkplain Call#enqueue asynchronously}.
*/
public synchronized void cancelAll() {
for (AsyncCall call : readyAsyncCalls) {
call.get().cancel();
}
for (AsyncCall call : runningAsyncCalls) {
call.get().cancel();
}
for (RealCall call : runningSyncCalls) {
call.cancel();
}
}
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.
}
}
/** Returns the number of running calls that share a host with {@code call}. */
private int runningCallsForHost(AsyncCall call) {
int result = 0;
for (AsyncCall c : runningAsyncCalls) {
if (c.host().equals(call.host())) result++;
}
return result;
}
/** Used by {@code Call#execute} to signal it is in-flight. */
//分發同步call,只加入到正在運行同步call的佇列
synchronized void executed(RealCall call) {
runningSyncCalls.add(call);
}
/** Used by {@code AsyncCall#run} to signal completion. */
void finished(AsyncCall call) {
finished(runningAsyncCalls, call, true);
}
/** Used by {@code Call#execute} to signal completion. */
//同步call已經完成,移除佇列
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!");
if (promoteCalls) promoteCalls();
runningCallsCount = runningCallsCount();
idleCallback = this.idleCallback;
}
if (runningCallsCount == 0 && idleCallback != null) {
idleCallback.run();
}
}
/** Returns a snapshot of the calls currently awaiting execution. */
//返回等待執行call的集合
public synchronized List<Call> queuedCalls() {
List<Call> result = new ArrayList<>();
for (AsyncCall asyncCall : readyAsyncCalls) {
result.add(asyncCall.get());
}
return Collections.unmodifiableList(result);
}
/** Returns a snapshot of the calls currently being executed. */
public synchronized List<Call> runningCalls() {
List<Call> result = new ArrayList<>();
result.addAll(runningSyncCalls);
for (AsyncCall asyncCall : runningAsyncCalls) {
result.add(asyncCall.get());
}
return Collections.unmodifiableList(result);
}
public synchronized int queuedCallsCount() {
return readyAsyncCalls.size();
}
public synchronized int runningCallsCount() {
return runningAsyncCalls.size() + runningSyncCalls.size();
}
}
在上面的同步call中,真正發出網路請求,解析返回結果的,還是getResponseWithInterceptorChain:
//重要的攔截器的責任鏈
Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
List<Interceptor> interceptors = new ArrayList<>();
interceptors.addAll(client.interceptors()); //(1)
interceptors.add(retryAndFollowUpInterceptor); //(2)
interceptors.add(new BridgeInterceptor(client.cookieJar())); //(3)
interceptors.add(new CacheInterceptor(client.internalCache())); //(4)
interceptors.add(new ConnectInterceptor(client)); //(5)
if (!forWebSocket) {
interceptors.addAll(client.networkInterceptors());
}
interceptors.add(new CallServerInterceptor(forWebSocket));
Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
originalRequest, this, eventListener, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
return chain.proceed(originalRequest);
}
3,在獲得相應之前經過的最後一關就是攔截器Interceptor
the whole thing is just a stack of built-in interceptors.
可見 Interceptor 是 OkHttp 最核心的一個東西,不要誤以為它只負責攔截請求進行一些額外的處理(例如 cookie),實際上它把實際的網路請求、快取、透明壓縮等功能都統一了起來,每一個功能都只是一個 Interceptor,它們再連線成一個 Interceptor.Chain,環環相扣,最終圓滿完成一次網路請求。
從 getResponseWithInterceptorChain 函式我們可以看到,Interceptor.Chain 的分佈依次是:
image.png
(1)在配置 OkHttpClient時設定的interceptors;
(2)負責失敗重試以及重定向的 RetryAndFollowUpInterceptor;
(3)負責把使用者構造的請求轉換為傳送到伺服器的請求、把伺服器返回的響應轉換為使用者友好的響應的BridgeInterceptor;
(4)負責讀取快取直接返回、更新快取的 CacheInterceptor
(5)負責和伺服器建立連線的ConnectInterceptor;
(6)配置 OkHttpClient 時設定的 networkInterceptors;
(7)負責向伺服器傳送請求資料、從伺服器讀取響應資料的 CallServerInterceptor
在這裡,位置決定了功能,最後一個 Interceptor 一定是負責和伺服器實際通訊的,重定向、快取等一定是在實際通訊之前的
2.2 在2.1中我們深入討論了同步請求的過程,下面講講非同步請求原理
程式碼:
Request request = new Request.Builder()
.url("http://publicobject.com/helloworld.txt")
.build();
//用request新建的call使用enqueue非同步請求
client.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 {
//相應成功回撥,response,非主執行緒
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());
}
});
由程式碼中client.newCall(request).enqueue(Callback),開始我們知道client.newCall(request)方法返回的是RealCall物件,接下來繼續向下看enqueue()方法:
//非同步任務使用
@Override
public void enqueue(Callback responseCallback) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
//送給分發器Dispatcher分發,其實Dispatcher中有執行緒池,把AsyncCall這個任務提交到執行緒池執行,通過responseCallback回撥
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
我們先看一下上面的Dispatcher類中的enqueue(Call )方法,在看看AsyncCall類:
synchronized void enqueue(AsyncCall call) {
if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
runningAsyncCalls.add(call);
executorService().execute(call);
} else {
readyAsyncCalls.add(call);
}
}
如果中的runningAsynCalls不滿,且call佔用的host小於最大數量,則將call加入到runningAsyncCalls中執行,同時利用執行緒池執行call;否者將call加入到readyAsyncCalls中。runningAsyncCalls和readyAsyncCalls是什麼呢?在把上面將同步Http請求時講過了,可以瞄一眼。
call加入到執行緒池中執行了。現在再看AsynCall的程式碼,它是RealCall中的內部類
//非同步請求,顯然是繼承了NamedRunnable ,在NamedRunnable 的run方法中執行繼承的execute() 方法
final class AsyncCall extends NamedRunnable {
private final Callback responseCallback;
private 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();
if (retryAndFollowUpInterceptor.isCanceled()) {
signalledCallback = true;
//回撥異常
responseCallback.onFailure(RealCall.this, new IOException("Canceled"));
} else {
signalledCallback = true;
//回撥成功
responseCallback.onResponse(RealCall.this, response);
}
} catch (IOException e) {
if (signalledCallback) {
// Do not signal the callback twice!
Platform.get().log(INFO, "Callback failure for " + toLoggableString(), e);
} else {
//回撥失敗
responseCallback.onFailure(RealCall.this, e);
}
} finally {
//告訴分發器Dispatcher請求執行完成
client.dispatcher().finished(this);
}
}
}
AysncCall中的execute()中的方法,同樣是通過Response response = getResponseWithInterceptorChain();來獲得response,這樣非同步任務也同樣通過了interceptor,剩下的就想看看上面的幾個攔截器是什麼鬼。
責任鏈攔截器Interceptor
RetryAndFollowUpInterceptor:負責失敗重試以及重定向
BridgeInterceptor:負責把使用者構造的請求轉換為傳送到伺服器的請求、把伺服器返回的響應轉換為使用者友好的響應的 。
ConnectInterceptor:建立連線
NetworkInterceptors:配置OkHttpClient時設定的 NetworkInterceptors
CallServerInterceptor:傳送和接收資料
作者:木有粗麵_9602
連結:https://www.jianshu.com/p/9f2c982cd500
來源:簡書
簡書著作權歸作者所有,任何形式的轉載都請聯絡作者獲得授權並註明出處。