談?wù)凮kHttp源碼的同步異步原理

2023-03-13 19:19:38來(lái)源：Android開(kāi)發(fā)編程

1、okhttp工作的大致流程當(dāng)我們通過(guò)OkhttpClient創(chuàng)建一個(gè)Call，并發(fā)起同步或異步請(qǐng)求時(shí)。okhttp會(huì)通過(guò)Dispatcher對(duì)我們所有的RealCall（Call的具體實(shí)現(xiàn)類）進(jìn)行統(tǒng)一管理，并通過(guò)execute()及enqueue()方法對(duì)同步或異步請(qǐng)求進(jìn)行處理。execute()及enqueue()這兩個(gè)方法會(huì)最終調(diào)用RealCall中的getResponseWithInterceptorChain()方法，從攔截器鏈中獲取返回結(jié)果。攔截器鏈中，依次通過(guò)RetryAndFollowUpInterceptor（重定向攔截器）、BridgeInterceptor（橋接攔截器）、CacheInterceptor（緩存攔截器）、ConnectInterceptor（連接攔截器）、CallServerInterceptor（網(wǎng)絡(luò)攔截器）對(duì)請(qǐng)求依次處理，與服務(wù)的建立連接后，獲取返回?cái)?shù)據(jù)，再經(jīng)過(guò)上述攔截器依次處理后，最后將結(jié)果返回給調(diào)用方。同步請(qǐng)求就是執(zhí)行請(qǐng)求的操作是阻塞式，直到 HTTP 響應(yīng)返回。它對(duì)應(yīng) OKHTTP 中的 execute 方法。異步請(qǐng)求就類似于非阻塞式的請(qǐng)求，它的執(zhí)行結(jié)果一般都是通過(guò)接口回調(diào)的方式告知調(diào)用者。它對(duì)應(yīng) OKHTTP 中的 enqueue 方法。

OkHttpClient okHttpClient = new OkHttpClient();Request request = new Request.Builder()        .url("http://www.test.com")        .build();Call call = okHttpClient.newCall(request);//1.異步請(qǐng)求，通過(guò)接口回調(diào)告知用戶 http 的異步執(zhí)行結(jié)果call.enqueue(new Callback() {    @Override    public void onFailure(Call call, IOException e) {        System.out.println(e.getMessage());    }    @Override    public void onResponse(Call call, Response response) throws IOException {        if (response.isSuccessful()) {            System.out.println(response.body().string());        }    }});//2.同步請(qǐng)求Response response = call.execute();if (response.isSuccessful()) {    System.out.println(response.body().string());}

2、異步請(qǐng)求實(shí)現(xiàn)流程

//get的異步請(qǐng)求    public void getAsync(View view) {        //定義okhttp對(duì)象        OkHttpClient okHttpClient = new OkHttpClient();        Request request = new Request.Builder().url("http://www.test.com").build();        Call call = okHttpClient.newCall(request);        //異步請(qǐng)求：不用創(chuàng)建子線程        //enqueue()并不會(huì)阻塞代碼的執(zhí)行,不需要與服務(wù)器請(qǐng)求完成之后,才會(huì)執(zhí)行后面的代碼        //而且enqueue內(nèi)部會(huì)為我們創(chuàng)建子線程        call.enqueue(new Callback() {            @Override            public void onFailure(@NotNull Call call, @NotNull IOException e) {            }            @Override            public void onResponse(@NotNull Call call, @NotNull Response response) throws IOException {             Log.i("TAG", "onResponse: " + (Looper.getMainLooper().getThread() == Thread.currentThread()));//為false 表示這是在子線程，需要切換到主線程才能操作UI                if (response.isSuccessful()){                    Log.i(TAG,"getAsync："+response.body().string());                }            }        });    }

(資料圖片)

call的異步調(diào)用是通過(guò)RealCall.enqueue()實(shí)現(xiàn)的。而請(qǐng)求結(jié)果通過(guò)Callback回調(diào)到主線程。

（1）RealCall.enqueue()

@Override public void enqueue(Callback responseCallback) {    synchronized (this) {      if (executed) throw new IllegalStateException("Already Executed");      executed = true;    }    transmitter.callStart();    client.dispatcher().enqueue(new AsyncCall(responseCallback));  }

將用戶創(chuàng)建的callback作為參數(shù)傳入AsyncCall()構(gòu)造函數(shù)。AsyncCall 繼承于Runnable。

（2）AsyncCall

final class AsyncCall extends NamedRunnable {    private volatile AtomicInteger callsPerHost = new AtomicInteger(0);    ...    /**     * 該方法是在dispather需要執(zhí)行此請(qǐng)求的時(shí)候，分配給它線程池，此異步請(qǐng)求便在這個(gè)線程池中執(zhí)行網(wǎng)絡(luò)請(qǐng)求。     */    void executeOn(ExecutorService executorService) {      ...      boolean success = false;      try {        //異步的關(guān)鍵：將請(qǐng)求放到線程池中執(zhí)行。        executorService.execute(this);        success = true;      } catch (RejectedExecutionException e) {       ...       success = false;      } finally {        if (!success) {          client.dispatcher().finished(this); // 執(zhí)行失敗會(huì)通過(guò)Dispatcher進(jìn)行finished，以后再也不會(huì)用此AsyncCall。        }      }    }    @Override protected void execute() {      boolean signalledCallback = false;      transmitter.timeoutEnter();      try {        Response response = getResponseWithInterceptorChain();        signalledCallback = true;        //請(qǐng)求成功時(shí)，回調(diào)Response給到用戶        responseCallback.onResponse(RealCall.this, response);      } catch (IOException e) {        ...        //請(qǐng)求錯(cuò)誤時(shí)，回調(diào)錯(cuò)誤接口給到用戶          responseCallback.onFailure(RealCall.this, e);      } finally {       //結(jié)束一次請(qǐng)求。        client.dispatcher().finished(this);      }    }  }

AsyncCall繼承于Runnable，它提供了將Call放到線程池執(zhí)行的能力，實(shí)現(xiàn)了請(qǐng)求的異步流程。

（3）Dispatcher.enqueue()

//準(zhǔn)備進(jìn)行異步調(diào)用的請(qǐng)求。private final Deque readyAsyncCalls = new ArrayDeque<>(); //正在執(zhí)行的異步請(qǐng)求。private final Deque runningAsyncCalls = new ArrayDeque<>();void enqueue(AsyncCall call) {    synchronized (this) {      //將異步請(qǐng)求加入到雙端隊(duì)列中      readyAsyncCalls.add(call);      // 尋找是否有同Host的請(qǐng)求，如果有進(jìn)行復(fù)用      if (!call.get().forWebSocket) {        AsyncCall existingCall = findExistingCallWithHost(call.host());        if (existingCall != null) call.reuseCallsPerHostFrom(existingCall);      }    }    //將符合條件的Ready的異步請(qǐng)求轉(zhuǎn)入runningAsyncCalls，并執(zhí)行    promoteAndExecute();  }

將此請(qǐng)求登記到Dispatcher的預(yù)備雙端隊(duì)列中。以此次的請(qǐng)求的Host來(lái)查找可服用的異步請(qǐng)求，如果存在，進(jìn)行復(fù)用。嘗試將剛剛加入預(yù)備隊(duì)的請(qǐng)求執(zhí)行。（4）Dipatcher.finish()

private  void finished(Deque calls, T call) {    Runnable idleCallback;    synchronized (this) {    ...    //一個(gè)請(qǐng)求完成后，檢查此時(shí)是否有在等待執(zhí)行的請(qǐng)求，并處理。    boolean isRunning = promoteAndExecute();    if (!isRunning && idleCallback != null) {      //通知此時(shí)已經(jīng)沒(méi)有異步請(qǐng)求任務(wù)      idleCallback.run();    }  }

調(diào)度器結(jié)束一次請(qǐng)求。當(dāng)一個(gè)異步任務(wù)完成后，調(diào)度器會(huì)觸發(fā)一次預(yù)備任務(wù)執(zhí)行流程。讓之前因?yàn)樽畲笳?qǐng)求數(shù)等限制而不能執(zhí)行的請(qǐng)求有機(jī)會(huì)得到執(zhí)行。通過(guò)idleCallback.run()通知此時(shí)的調(diào)度器空閑狀態(tài)。（5）Dipatcher.promoteAndExecute()

private boolean promoteAndExecute() {    ...    List executableCalls = new ArrayList<>();    boolean isRunning;    synchronized (this) {      for (Iterator i = readyAsyncCalls.iterator(); i.hasNext(); ) {        AsyncCall asyncCall = i.next();        //檢查最大請(qǐng)求數(shù)限制和        if (runningAsyncCalls.size() >= maxRequests) break;        if (asyncCall.callsPerHost().get() >= maxRequestsPerHost) continue;        //滿足條件，便把預(yù)備隊(duì)的請(qǐng)求提升到執(zhí)行隊(duì)列。        i.remove();        asyncCall.callsPerHost().incrementAndGet();        executableCalls.add(asyncCall);        runningAsyncCalls.add(asyncCall);      }      isRunning = runningCallsCount() > 0;    }    //將可執(zhí)行的異步請(qǐng)求放進(jìn)線程池執(zhí)行    for (int i = 0, size = executableCalls.size(); i < size; i++) {      AsyncCall asyncCall = executableCalls.get(i);      //      asyncCall.executeOn(executorService());    }    return isRunning;

該方法是對(duì)預(yù)備隊(duì)列里的請(qǐng)求提升至執(zhí)行隊(duì)列并執(zhí)行的一次嘗試。如果不能執(zhí)行，他啟動(dòng)時(shí)機(jī)將會(huì)延后到其他請(qǐng)求結(jié)束。3、同步請(qǐng)求

public static final MediaType JSON    = MediaType.get("application/json; charset=utf-8");OkHttpClient client = new OkHttpClient();String post(String url, String json) throws IOException {  RequestBody body = RequestBody.create(json, JSON);  Request request = new Request.Builder()      .url(url)      .post(body)      .build();  try (Response response = client.newCall(request).execute()) {    return response.body().string();  }}

通過(guò)Builder模式構(gòu)建Request。調(diào)用client.newCall()，通過(guò)request生成一個(gè)Call對(duì)象。他的實(shí)現(xiàn)類是RealCall。隨后調(diào)用RealCall.execute()，進(jìn)行同步請(qǐng)求。（1）RealCall.execute()

@Override public Response execute() throws IOException {    synchronized (this) {      // 如果該請(qǐng)求已經(jīng)執(zhí)行過(guò)，報(bào)錯(cuò)。      if (executed) throw new IllegalStateException("Already Executed");      executed = true;    }    transmitter.timeoutEnter();    transmitter.callStart();    try {      //獲取 client 里面的調(diào)度器 Dispatcher 并記錄這個(gè)請(qǐng)求。      client.dispatcher().executed(this);      //通過(guò)責(zé)任鏈的方式將發(fā)起請(qǐng)求并返回結(jié)果。這里是同步動(dòng)作，會(huì)阻塞。      return getResponseWithInterceptorChain();    } finally {      //請(qǐng)求完后需要把這個(gè)請(qǐng)求從調(diào)度器中移走      client.dispatcher().finished(this);    }  }

判斷Call的合法性。將RealCall傳進(jìn)Client里面的Dispatcher.executed()里，而Dispatcher是在 OkHttpClient 被的構(gòu)建函數(shù)里被創(chuàng)建并作為成員變量的。開(kāi)啟責(zé)任鏈模式，進(jìn)行請(qǐng)求相關(guān)邏輯。執(zhí)行完成后，調(diào)度器對(duì)這個(gè)請(qǐng)求進(jìn)行收尾工作。（2）Dispatcher.executed()

/** Running asynchronous calls. Includes canceled calls that haven"t finished yet. */  private final Deque runningAsyncCalls = new ArrayDeque<>();  synchronized void executed(RealCall call) {    runningSyncCalls.add(call);  }

（3）RealCall.getResponseWithInterceptorChain()

Response getResponseWithInterceptorChain() throws IOException {    // 將請(qǐng)求的具體邏輯進(jìn)行分層，并采用責(zé)任鏈的方式進(jìn)行構(gòu)造。    List interceptors = new ArrayList<>();    // 用戶自已的請(qǐng)求攔截器    interceptors.addAll(client.interceptors());    //重試和重定向攔截器    interceptors.add(new RetryAndFollowUpInterceptor(client));    //橋攔截器    interceptors.add(new BridgeInterceptor(client.cookieJar()));    //緩存邏輯攔截器    interceptors.add(new CacheInterceptor(client.internalCache()));    //網(wǎng)絡(luò)連接邏輯攔截器    interceptors.add(new ConnectInterceptor(client));    if (!forWebSocket) {      interceptors.addAll(client.networkInterceptors());    }    // 網(wǎng)絡(luò)請(qǐng)求攔截器，真正網(wǎng)絡(luò)通行的地方，這個(gè)攔截器處理過(guò)后會(huì)生成一個(gè)Response    interceptors.add(new CallServerInterceptor(forWebSocket));    //依照如上配置，構(gòu)建出一個(gè)請(qǐng)求的處理邏輯責(zé)任鏈，特別注意：這條鏈開(kāi)始于下標(biāo)位置為的0攔截器。    Interceptor.Chain chain = new RealInterceptorChain(interceptors, transmitter, null, 0,        originalRequest, this, client.connectTimeoutMillis(),        client.readTimeoutMillis(), client.writeTimeoutMillis());    boolean calledNoMoreExchanges = false;    try {      //按下處理邏輯鏈條的開(kāi)關(guān)。      Response response = chain.proceed(originalRequest);      if (transmitter.isCanceled()) {        closeQuietly(response);        throw new IOException("Canceled");      }      //返回請(qǐng)求結(jié)果      return response;    } catch (IOException e) {      calledNoMoreExchanges = true;      throw transmitter.noMoreExchanges(e);    } finally {      if (!calledNoMoreExchanges) {        transmitter.noMoreExchanges(null);      }    }  }

將用戶自定義的攔截器先加入集合中。加入一次請(qǐng)求中需要用的攔截器，這些攔截器代表一次完整的網(wǎng)絡(luò)請(qǐng)求邏輯被分了幾層以及他們的先后順序。從代碼中我們不難看出他們的流程是：重試/重定向邏輯->網(wǎng)絡(luò)橋邏輯->緩存邏輯->建立網(wǎng)絡(luò)連接邏輯->網(wǎng)絡(luò)通行邏輯。用以上攔截器集合構(gòu)建出一條邏輯處理的攔截鏈，并將這條鏈需要使用的攔截器下標(biāo)賦值為0，從第一個(gè)開(kāi)始。調(diào)用chain.proceed()啟動(dòng)這條鏈的處理流程。使用責(zé)任鏈的設(shè)計(jì)模式來(lái)處理一次網(wǎng)絡(luò)請(qǐng)求中的邏輯可以有效的劃分邏輯層。而前一個(gè)攔截器可以根據(jù)實(shí)際的處理情況來(lái)決定下一攔截器是否應(yīng)該繼續(xù)處理。（4）RealInterceptorChain.proceed()

RealInterceptorChain.javapublic Response proceed(Request request, Transmitter transmitter, @Nullable Exchange exchange)      throws IOException {    if (index >= interceptors.size()) throw new AssertionError();    calls++;    ...    /**    * index+1：構(gòu)建出新的攔截鏈，不過(guò)新的攔截鏈的處理攔截器是下標(biāo)為index+1的    * 實(shí)現(xiàn)了責(zé)任鏈中，處理邏輯的流轉(zhuǎn)。    */    RealInterceptorChain next = new RealInterceptorChain(interceptors, transmitter, exchange,        index + 1, request, call, connectTimeout, readTimeout, writeTimeout);    //此時(shí)index = 0;所以拿到了第一個(gè)攔截器，并且調(diào)用他的intercept 方法進(jìn)行具體邏輯處理。    Interceptor interceptor = interceptors.get(index);    //當(dāng)前攔截器對(duì)網(wǎng)絡(luò)請(qǐng)求進(jìn)行處理。    Response response = interceptor.intercept(next);    // Confirm that the next interceptor made its required call to chain.proceed().    if (exchange != null && index + 1 < interceptors.size() && next.calls != 1) {      throw new IllegalStateException("network interceptor " + interceptor          + " must call proceed() exactly once");    }    // 省略對(duì)response合法性的檢查代碼    ...    return response;  }

先index+1，構(gòu)建index指向下一個(gè)攔截器的責(zé)任鏈。在鏈中取出下標(biāo)當(dāng)前index（此時(shí)為0）的攔截器，并調(diào)用intercept(next)進(jìn)行此次攔截器邏輯的真正處理。這里注意，傳遞進(jìn)去的參數(shù)正是1中構(gòu)建出來(lái)的next責(zé)任鏈。在Interceptor.intercept()方法內(nèi)部進(jìn)行自身的邏輯處理以后，會(huì)調(diào)用next.proceed()進(jìn)行一次傳遞，由下一個(gè)攔截器進(jìn)行處理。而此次請(qǐng)求也是這樣在各個(gè)攔截器中被處理->傳遞->處理->...->返回結(jié)果。4、異步和同步總結(jié)

同步

RealCall.execute() ?? Dispatcher.execute() ---runningSyncCalls（異步請(qǐng)求執(zhí)行隊(duì)列）添加 call 對(duì)象。getResponseWithInterceptorChain() 返回 Response 對(duì)象。Dispatcher.finished()。

異步

RealCall. enqueue() ?? Dispatcher. enqueue()---readyAsyncCalls（異步請(qǐng)求就緒隊(duì)列）添加 call 對(duì)象。readyAsyncCalls（異步請(qǐng)求就緒隊(duì)列），將滿足條件的添加到臨時(shí)隊(duì)列和runningAsyncCalls （異步請(qǐng)求的執(zhí)行隊(duì)列）。getResponseWithInterceptorChain() 返回 Response 對(duì)象；。Dispatcher.finished()。

關(guān)鍵詞：