以如下請(qǐng)求為例進(jìn)行說(shuō)明
OkHttpClient httpClient = new OkHttpClient();
Request request = new Request.Builder().url("http://gank.io/api/data/Android/10/1").build();
Call call = httpClient.newCall(request);
call.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
Log.e(TAG, e.getMessage());
}
@Override
public void onResponse(Call call, Response response) throws IOException {
String result = response.body().string();
Log.d(TAG, result);
}
});
OkHttp和Request的構(gòu)建就不說(shuō)了，我們來(lái)看下Call的構(gòu)建
Call的構(gòu)建
Call是個(gè)接口
public interface Call extends Cloneable {
Request request();
Response execute() throws IOException;
void enqueue(Callback responseCallback);
boolean isExecuted();
boolean isCanceled();
Call clone();
interface Factory {
Call newCall(Request request);
}
}
內(nèi)部還定義了一個(gè)Factory接口，OkHttpClient實(shí)現(xiàn)了該接口
QQ20170507-084805@2x.png
@Override
public Call newCall(Request request) {
return new RealCall(this, request, false /* for web socket */);
}
內(nèi)部創(chuàng)建了一個(gè)RealCall，注意RealCall是Call唯一實(shí)現(xiàn)類
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);
}
構(gòu)造函數(shù)中做了以下幾件事
- EventListener.Factory先忽略，這里不關(guān)心；
- 保存OkHttpClient實(shí)例；
- 保存原始的Request實(shí)例；
- WebSocket的標(biāo)識(shí)，傳過(guò)來(lái)的是false，表示沒(méi)用WebSocket；
- 創(chuàng)建了第一個(gè)攔截器RetryAndFollowUpInterceptor，用來(lái)處理請(qǐng)求異常重試；
Call執(zhí)行
這里是個(gè)異步請(qǐng)求
@Override
public void enqueue(Callback responseCallback) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
captureCallStackTrace();
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
- client.dispatcher()返回Dispatcher對(duì)象，它在OkHttpClient構(gòu)建時(shí)創(chuàng)建，用來(lái)處理請(qǐng)求的任務(wù)隊(duì)列，使用了生產(chǎn)者消費(fèi)者模式
- 創(chuàng)建一個(gè)AsyncCall，丟給Dispatcher
synchronized void enqueue(AsyncCall call) {
if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
runningAsyncCalls.add(call);
executorService().execute(call);
} else {
readyAsyncCalls.add(call);
}
}
- maxRequests = 64: 最大并發(fā)請(qǐng)求數(shù)為64
- maxRequestsPerHost = 5: 每個(gè)主機(jī)最大請(qǐng)求數(shù)為5
當(dāng)滿足正在執(zhí)行的請(qǐng)求數(shù)小于64并且每個(gè)主機(jī)最大請(qǐng)求數(shù)小于5，直接把AsyncCall加到runningCalls的隊(duì)列中，并在ExecutorService中執(zhí)行。反之就放入readyAsyncCalls進(jìn)行等待。這里會(huì)進(jìn)入第一個(gè)if塊。
接著執(zhí)行AsyncCall#execute
@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 {
client.dispatcher().finished(this);
}
}
這個(gè)方法做了三件事
- 通過(guò)getResponseWithInterceptorChain執(zhí)行請(qǐng)求
- 處理回調(diào)，onRespons or onFailure
- 釋放Call，將執(zhí)行完畢的Call從runningCalls隊(duì)列中移除，并取出readyAsyncCalls中的一個(gè)Call加入到runningCalls執(zhí)行。
1.責(zé)任鏈分發(fā)請(qǐng)求與響應(yīng)
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()));
interceptors.add(new CacheInterceptor(client.internalCache()));
interceptors.add(new ConnectInterceptor(client));
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);
}
在這個(gè)方法中我們看到了大量的Interceptor，不同的Interceptor賦予不同的職責(zé)，有的處理請(qǐng)求重試、有的處理緩存、有的處理請(qǐng)求壓縮、有的執(zhí)行實(shí)際的請(qǐng)求等等，這些Interceptor被Interceptor.Chain串聯(lián)起來(lái)，順序執(zhí)行。
這里使用了責(zé)任鏈設(shè)計(jì)模式
責(zé)任鏈模式是一種對(duì)象的行為模式。在責(zé)任鏈模式里，很多對(duì)象由每一個(gè)對(duì)象對(duì)其下家的引用而連接起來(lái)形成一條鏈。請(qǐng)求在這個(gè)鏈上傳遞，直到鏈上的某一個(gè)對(duì)象決定處理此請(qǐng)求。發(fā)出這個(gè)請(qǐng)求的客戶端并不知道鏈上的哪一個(gè)對(duì)象最終處理這個(gè)請(qǐng)求，這使得系統(tǒng)可以在不影響客戶端的情況下動(dòng)態(tài)地重新組織和分配責(zé)任。
689802-816d907d9939eb58.png
我們看看責(zé)任鏈的實(shí)現(xiàn)
Interceptor是個(gè)接口，定義如下
public interface Interceptor {
Response intercept(Chain chain) throws IOException;
interface Chain {
Request request();
Response proceed(Request request) throws IOException;
Connection connection();
}
}
內(nèi)部還定義了Chain接口，RealInterceptorChain是唯一實(shí)現(xiàn)類，在getResponseWithInterceptorChain內(nèi)部new了一個(gè)RealInterceptorChain
然后調(diào)用RealInterceptorChain#proceed
@Override
public Response proceed(Request request) throws IOException {
return proceed(request, streamAllocation, httpCodec, connection);
}
public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
RealConnection connection) throws IOException {
if (index >= interceptors.size()) throw new AssertionError();
calls++;
// If we already have a stream, confirm that the incoming request will use it.
if (this.httpCodec != null && !this.connection.supportsUrl(request.url())) {
throw new IllegalStateException("network interceptor " + interceptors.get(index - 1)
+ " must retain the same host and port");
}
// If we already have a stream, confirm that this is the only call to chain.proceed().
if (this.httpCodec != null && calls > 1) {
throw new IllegalStateException("network interceptor " + interceptors.get(index - 1)
+ " must call proceed() exactly once");
}
// 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);
// Confirm that the next interceptor made its required call to chain.proceed().
if (httpCodec != null && index + 1 < interceptors.size() && next.calls != 1) {
throw new IllegalStateException("network interceptor " + interceptor
+ " must call proceed() exactly once");
}
// Confirm that the intercepted response isn't null.
if (response == null) {
throw new NullPointerException("interceptor " + interceptor + " returned null");
}
return response;
}
- 循環(huán)取出攔截器集合中的每個(gè)Interceptpr
- 新建一個(gè)RealInterceptorChain，作為上一步取出的Interceptor#Intercept的參數(shù)
- 調(diào)用Interceptor#Intercept方法，取出Request，在把它轉(zhuǎn)發(fā)給下一個(gè)Interceptor之前做一些加工處理。
- intercept中通過(guò)調(diào)用RealInterceptorChain#proceed，將請(qǐng)求轉(zhuǎn)發(fā)給下一個(gè)Interceptor處理
- 當(dāng)前Interceptor將阻塞等待下一個(gè)Interceptor返回Response，拿到Response后，自己在加工處理一下再交給上一個(gè)Interceptor。
1819140-312c2f839564cfa3.png
后面將具體分析每一個(gè)攔截器的職責(zé)。
2. 請(qǐng)求響應(yīng)回調(diào)
在getResponseWithInterceptorChain中拿到Response后需要回調(diào)Callback了。
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);
}
}
回調(diào)失敗
兩種情況
- 請(qǐng)求被取消；
- getResponseWithInterceptorChain內(nèi)部發(fā)生IO異常；
回調(diào)成功
- onResponse，請(qǐng)求結(jié)果封裝在ResponseBody中
注意：onFailure和onResponse都執(zhí)行在線程池的線程中，而不是UI線程。
如果在以上回調(diào)中操作UI將拋出如下異常
QQ20170507-161133@2x.png
3. 釋放Call
在finally中最后調(diào)用了
client.dispatcher().finished(this);
按照J(rèn)ava語(yǔ)法規(guī)范，無(wú)論發(fā)生異常與否，finally中的代碼必然會(huì)執(zhí)行的。
因此，最終調(diào)用Dispatcher#finished
/** Used by {@code AsyncCall#run} to signal completion. */
void finished(AsyncCall call) {
finished(runningAsyncCalls, call, true);
}
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();
}
}
- promoteCalls為true
- 從runningSyncCalls中移除該Call
- 接著執(zhí)行promoteCalls()函數(shù)
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.
}
}
- 如果正在執(zhí)行的隊(duì)列任務(wù)數(shù)仍大于64，繼續(xù)等待；
- 如果等待隊(duì)列為空，啥也不做，返回；
- 循環(huán)從等待隊(duì)列中取出一個(gè)AsyncCall
- 如果滿足單個(gè)主機(jī)并發(fā)數(shù)小于5，則將該AsyncCall加入執(zhí)行隊(duì)列，并執(zhí)行execute；
- 當(dāng)執(zhí)行隊(duì)列并發(fā)數(shù)大于64，退出循環(huán)；
Refference
http://m.itdecent.cn/p/aad5aacd79bf
https://blog.piasy.com/2016/07/11/Understand-OkHttp/
