OkHttp源码分析【同步、异步请求流程】

原创

笔头

发布于 2022-03-23 17:24:40

8730

发布于 2022-03-23 17:24:40

文章被收录于专栏：Android记忆

之前我们写过volley源码分析 Volley源码解读，volley相比OkHttp而言简单些，这次我们来看下OkHttp源码吧。

一、demo开场

//同步请求
OkHttpClient okHttpClient=new OkHttpClient();
final Request request=new Request.Builder()
        .url("https://www.wanandroid.com/navi/json")
        .get()
        .build();
final Call call = okHttpClient.newCall(request);
try {
    Response response = call.execute();
    Log.e("同步结果---- ",response.body().string()+"");

} catch (IOException e) {
    e.printStackTrace();
}

通过这代码，我们大概能找出主要类，OkHttpClient、Request、Call、Response，后面我们重点关注这些。

二、同步请求源码

先看下OkHttpClient

OkHttpClient(Builder builder) {
  this.dispatcher = builder.dispatcher;
  this.proxy = builder.proxy;
  this.protocols = builder.protocols;
  this.connectionSpecs = builder.connectionSpecs;
  this.interceptors = Util.immutableList(builder.interceptors);
  this.networkInterceptors = Util.immutableList(builder.networkInterceptors);
  this.eventListenerFactory = builder.eventListenerFactory;
  this.proxySelector = builder.proxySelector;
  this.cookieJar = builder.cookieJar;
  this.cache = builder.cache;
  this.internalCache = builder.internalCache;
  this.socketFactory = builder.socketFactory;

这个主要初始化一些数据，这里我们注意下this.dispatcher和this.interceptors这两个对象，比较重要，后面会用到。

我们继续看

final Request request=new Request.Builder()
        .url("https://www.wanandroid.com/navi/json")
        .get()
        .build();

这里Builder()利用建造者模式创建请求相关url、请求方式、请求头等数据

public Request build() {
  if (url == null) throw new IllegalStateException("url == null");
  return new Request(this);
}

build返回一个Request请求对象。this就是当前Builder对象，

Request(Builder builder) {
  this.url = builder.url;
  this.method = builder.method;
  this.headers = builder.headers.build();
  this.body = builder.body;
  this.tags = Util.immutableMap(builder.tags);
}

这里没有难点，细节可以自行看下源码看。

接下来看下newCall(request)

@Override public Call newCall(Request request) {
  return RealCall.newRealCall(this, request, false /* for web socket */);
}

static RealCall newRealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
  // Safely publish the Call instance to the EventListener.
  RealCall call = new RealCall(client, originalRequest, forWebSocket);
  call.eventListener = client.eventListenerFactory().create(call);
  return call;
}

不复杂Call是一个接口，具体实例是返回的RealCall对象。绑定了当前OkHttpClient对象和Request请求数据

至此，请求相关的准备都做好了。接下来就是执行发起请求操作了。

我们来看下call.execute()

我们知道当前call的具体实例是RealCall，直接看RealCall的execute()方法。

@Override public Response execute() throws IOException {
  synchronized (this) {
    if (executed) throw new IllegalStateException("Already Executed");
    executed = true;
  }
  captureCallStackTrace();
  eventListener.callStart(this);
  try {
    client.dispatcher().executed(this); //1
    Response result = getResponseWithInterceptorChain(); //2
    if (result == null) throw new IOException("Canceled");
    return result;
  } catch (IOException e) {
    eventListener.callFailed(this, e);
    throw e;
  } finally {
    client.dispatcher().finished(this);
  }
}

这里面主要的就是1和2，

1这句话，主要涉及到了dispatcher，Dispatcher是干什么的？我们看下

public final class Dispatcher {
  private int maxRequests = 64;
  private int maxRequestsPerHost = 5;
  private @Nullable Runnable idleCallback;

  /** Executes calls. Created lazily. */
  private @Nullable ExecutorService executorService;

  /** Ready async calls in the order they'll be run. */
  private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();

  /** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
  private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();

  /** Running synchronous calls. Includes canceled calls that haven't finished yet. */
  private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

他这个里面主要定义了请求队列，正在执行同步请求队列，正在执行异步请求队列，最大请求数，请求线程池等等。可以说Dispatcher是一个请求管理者。

接着看1这句话，看源码

synchronized void executed(RealCall call) {
  runningSyncCalls.add(call);
}

在正在执行同步请求队列添加一个请求call，这里还不清楚runningSyncCalls有啥用，记住他就好。

接着看2，这句话看起来是返回请求数据，重点看下getResponseWithInterceptorChain()

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, this, eventListener, client.connectTimeoutMillis(),
      client.readTimeoutMillis(), client.writeTimeoutMillis());

  return chain.proceed(originalRequest);
}

原来这里是定义一些拦截器Interceptor，各种各样的拦截器，每个拦截器作用也不一样，这些拦截器就是okhttp的精华。拦截器这里先不展开说了，下篇我们再说。

我们接着看RealInterceptorChain

public RealInterceptorChain(List<Interceptor> interceptors, StreamAllocation streamAllocation,
    HttpCodec httpCodec, RealConnection connection, int index, Request request, Call call,
    EventListener eventListener, int connectTimeout, int readTimeout, int writeTimeout) {
  this.interceptors = interceptors;
  this.connection = connection;
  this.streamAllocation = streamAllocation;
  this.httpCodec = httpCodec;
  this.index = index;
  this.request = request;
  this.call = call;
  this.eventListener = eventListener;
  this.connectTimeout = connectTimeout;
  this.readTimeout = readTimeout;
  this.writeTimeout = writeTimeout;
}

RealInterceptorChain是一个拦截链，里面定义了各种拦截器相关参数。直接看chain.proceed，这里的chain接口具体实例是RealInterceptorChain，我们打开它看里面的proceed方法

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, call, eventListener, connectTimeout, readTimeout,
      writeTimeout);
  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");
  }

  if (response.body() == null) {
    throw new IllegalStateException(
        "interceptor " + interceptor + " returned a response with no body");
  }

  return response;
}

核心代码块命名为代码A

// Call the next interceptor in the chain.
RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec,
    connection, index + 1, request, call, eventListener, connectTimeout, readTimeout,
    writeTimeout);
Interceptor interceptor = interceptors.get(index);
Response response = interceptor.intercept(next);

看样子，好像是遍历拦截链中所有拦截器执行拦截方法。我们看下interceptor.intercept，这里有点烧脑，我们慢慢理顺他。

默认index=0，此时的next我们命名为拦截链1，interceptor命名为拦截器1。【这里的index作用有2个，第一个就是在proceed方法中，我们会判断index值是否大于拦截器数量，是的话就报错。第二个作用就是从拦截器List中获得index位置的拦截器。】

我们先看下interceptors初始数据，他是在RealCall设置的，从getResponseWithInterceptorChain方法中可以知道。我们这篇文章先不细看拦截器了，为了方便，我们假设interceptors中只有BridgeInterceptor、CacheInterceptor、CallServerInterceptor这三个，CallServerInterceptor必须存在，他是最后发起请求的模块。那我们就看interceptor.intercept(next)吧。

默认index=0，那拦截器1就是BridgeInterceptor，然后我们看下他的intercept方法，

@Override public Response intercept(Chain chain) throws IOException {

------------------------------------ 1 -------------------------------
  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());
  }
------------------------------------ 2 -------------------------------
  Response networkResponse = chain.proceed(requestBuilder.build());

------------------------------------ 3 -------------------------------
  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();
}

intercept方法体，内容主要包含3块。

第一块：是对请求头的各种设置，【先不管是什么】

第二块：这里的chain就是拦截链1【上面有定义】。然后调用proceed方法，又走到RealInterceptorChain的代码A【上面有定义】这里，此时index=1，此时的next我们命名为拦截链2，interceptor命名为拦截器2。然后调用CacheInterceptor的intercept方法，大体和BridgeInterceptor的intercept方法差不多，也是分三块，在第二块中，chain是拦截链2，然后调用proceed方法又走到RealInterceptorChain的代码A，此时index=2，此时next我们命名为拦截链3，interceptor命名为拦截器3【CallServerInterceptor】，然后调用CallServerInterceptor的intercept方法，CallServerInterceptor的intercept方法和其他拦截器不一样。拦截器3直接返回请求数据response，结果返回给拦截器2，拦截器2对返回数据response再次处理，把数据返回给拦截器1，

第三块：对response数据进行处理。

最后把拦截器1对返回数据response，返回给getResponseWithInterceptorChain方法中，最后把数据返回给前端，至此整个请求完成。

拦截器大致流程是：拦截器1把请求相关设置好了之后，调用拦截器2，再次对请求相关配置进行设置，调用拦截器3，处理请求，返回response给拦截器2，拦截器2对response数据进行处理后，再返回给拦截器1，拦截器1对response数据进行处理后，返回给getResponseWithInterceptorChain方法。整个拦截过程完成。

============================以上是同步请求源码=======================

我们再看下异步请求源码

三、异步请求源码

demo

call.enqueue(new Callback() {
    @Override
    public void onFailure(Call call, IOException e) {
    }
    @Override
    public void onResponse(Call call, Response response) throws IOException {
    }
});

@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));
}

来看下enqueue方法

void enqueue(AsyncCall call) {
  synchronized (this) {
    readyAsyncCalls.add(call);
    ......
  }
  promoteAndExecute();
}

把当前的call请求添加到readyAsyncCalls，后面会使用到他。

接下来看下promoteAndExecute

private boolean promoteAndExecute() {
  assert (!Thread.holdsLock(this));

  List<AsyncCall> executableCalls = new ArrayList<>();
  boolean isRunning;
  synchronized (this) {
    for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
      AsyncCall asyncCall = i.next();

      if (runningAsyncCalls.size() >= maxRequests) break; // Max capacity.
      if (asyncCall.callsPerHost().get() >= maxRequestsPerHost) continue; // Host max capacity.

      i.remove();
      asyncCall.callsPerHost().incrementAndGet();
      executableCalls.add(asyncCall);
      runningAsyncCalls.add(asyncCall);
    }
    isRunning = runningCallsCount() > 0;
  }
  for (int i = 0, size = executableCalls.size(); i < size; i++) {
    AsyncCall asyncCall = executableCalls.get(i);
    asyncCall.executeOn(executorService());
  }
  return isRunning;
}

第一个for，把等待执行的请求放到正在进行执行的请求list中，同时放在executableCalls List中。

第二个for，遍历获取每一个AsyncCall执行executeOn

void executeOn(ExecutorService executorService) {
  assert (!Thread.holdsLock(client.dispatcher()));
  boolean success = false;
  try {
    executorService.execute(this);
    success = true;
  } catch (RejectedExecutionException e) {
    InterruptedIOException ioException = new InterruptedIOException("executor rejected");
    ioException.initCause(e);
    transmitter.noMoreExchanges(ioException);
    responseCallback.onFailure(RealCall.this, ioException);
  } finally {
    if (!success) {
      client.dispatcher().finished(this); // This call is no longer running!
    }
  }
}

直接看 executorService.execute(this);开启线程池执行this任务，当前this是AsyncCall继承NamedRunnable，我们直接看他的run方法，AsyncCall没有重写run方法，我们直接看他的基类NamedRunnable，看他的run方法。

@Override public final void run() {
  String oldName = Thread.currentThread().getName();
  Thread.currentThread().setName(name);
  try {
    execute();
  } finally {
    Thread.currentThread().setName(oldName);
  }
}

execute()方法具体实现在AsyncCall中

@Override protected void execute() {
  boolean signalledCallback = false;
  transmitter.timeoutEnter();
  try {
    Response response = getResponseWithInterceptorChain();
    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);
    }
  } catch (Throwable t) {
    cancel();
    if (!signalledCallback) {
      IOException canceledException = new IOException("canceled due to " + t);
      canceledException.addSuppressed(t);
      responseCallback.onFailure(RealCall.this, canceledException);
    }
    throw t;
  } finally {
    client.dispatcher().finished(this);
  }
}

我们看到，这个里面执行了getResponseWithInterceptorChain，这个在我们看同步请求源码中也遇到这个，那我们就直接看同步请求源码。拿到请求结果后通过callback把数据回调给Callback。

至此，异步请求流程也结束了。

四、总结

异步请求和同步请求相比，除了加了Callback回调。异步请求，面对不断的请求，先将其存放在readyAsyncCalls，然后把readyAsyncCalls中所有请求转移到executableCalls和runningAsyncCalls中【转移过程中，需要判断runningAsyncCalls size大小，默认大小是64】，然后遍历executableCalls，开启线程池执行每个请求。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

http