转载

源码分析OKHttp的执行过程

OKHttp 是目前 Android 平台主流的网络请求的基础框架。因此我们有必要对其源码进行阅读学习，了解其内部的原理、项目结构、以及请求的执行过程。

它的项目地址为： github.com/square/okht…

0x00 简单使用

先从一个简单的官方示例来看，这是一个同步 GET 请求

public class GetExample {
  //1.http客户端
  OkHttpClient client = new OkHttpClient();
    
  String run(String url) throws IOException {
    //2.构造请求
    Request request = new Request.Builder()
        .url(url)
        .build();
	//3.执行请求，获取响应数据
    try (Response response = client.newCall(request).execute()) {
      return response.body().string();
    }
  }

  public static void main(String[] args) throws IOException {
    GetExample example = new GetExample();
    String response = example.run("https://raw.github.com/square/okhttp/master/README.md");
    System.out.println(response);
  }
}
复制代码

可以看出这个 GET 请求操作是很简单的。有几个很重要的接口

OKHttpClient ：它代表着 http 客户端
Request ：它封装了请求对象，可以构造一个 http 请求对象
Response ：封装了响应结果
Call ： client.newCall 调用后生成一个请求执行对象 Call ，它封装了请求执行过程。

这几个接口是程序员在使用 OKHttp 库中经常遇到的。

接下来将从这个示例开始阅读 OkHttp 的源码

0x01 Call.execute()

跟进源码后发现这个方法是在 Call 中的接口

/**
 * A call is a request that has been prepared for execution. A call can be canceled. As this object
 * represents a single request/response pair (stream), it cannot be executed twice.
 */
public interface Call extends Cloneable {
  //...
  //同步执行请求
  Response execute() throws IOException;

  //将请求加入队列
  void enqueue(Callback responseCallback);

  //...
}
复制代码

从源码注释知道， Call 是一个准备请求的执行对象，它可以被取消，代表一个 “请求/响应” 对，不能执行两次。

RealCall

Call 的实现类是 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);
      Response result = getResponseWithInterceptorChain();
      if (result == null) throw new IOException("Canceled");
      return result;
    } catch (IOException e) {
      eventListener.callFailed(this, e);
      throw e;
    } finally {
      client.dispatcher().finished(this);
    }
  }
复制代码

这个方法也不是很长，逻辑很简单：

同步锁检查该请求是否已经执行，如果没有则标记 executed = ture ，否则抛出异常
调用了回调函数 callStart
okhttp 客户端调用 dispatcher 将执行请求对象
调用了 getResponseWithInterceptorChain 方法获取到响应数据 Response ，这个方法很重要，后面会继续跟进
然后是对请求失败的回调 callFailed
最后还是使用 dispather 对象调用 finished 方法，完成请求

这里的逻辑还是比较清晰的，出现两个重要的方法

dispatcher.execute
getResponseWithInterceptorChain

接下来分别看这两个方法

0x02 Dispatcher

public final class Dispatcher {
 	
  /** 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<>();
  //...

  synchronized void enqueue(AsyncCall call) {
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
      runningAsyncCalls.add(call);
      executorService().execute(call);
    } else {
      readyAsyncCalls.add(call);
    }
  }

  /** Used by {@code Call#execute} to signal it is in-flight. */
  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. */
  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();
    }
  }
}
复制代码

可以看出 Dispatcher 是一个调度器，它内部有一个线程池 executorService ,还有三个队列，分别代表同步请求进行队列、异步请求等待队列、异步请求执行队列。

我们发现调用execute方法时就是将Call对象加入到同步请求进行队列runningSyncCalls中，而调用finished 方法则是将Call请求从队列中移除

0x03 getResponseWithInterceptorChain

现在在回到 RealCall 源码中，这个方法可以说是 OkHttp 最关键的部分了

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()));//处理cookie的拦截器
    interceptors.add(new CacheInterceptor(client.internalCache()));//处理缓存的拦截器
    interceptors.add(new ConnectInterceptor(client));//负责连接的拦截器
    if (!forWebSocket) {
      interceptors.addAll(client.networkInterceptors());//添加程序员自定义的network拦截器
    }
    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);
  }
复制代码

在添加了一系列的拦截器之后，又构造了一个拦截器责任链，这个 RealInterceptorChain 包含了所有的拦截器对象。然后调用 chain.proceed 方法开始执行请求，这时就到了 RealInterceptorChain 这个类中。

0x04 RealInterceptorChain

@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++;

    //省略无关代码...

    //1. 执行拦截器责任链中的下一个拦截器
    RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec,
        connection, index + 1, request, call, eventListener, connectTimeout, readTimeout,
        writeTimeout);
    //2. 获取当前的拦截器
    Interceptor interceptor = interceptors.get(index);
    //3. 执行拦截，并返回响应
    Response response = interceptor.intercept(next);

    //省略...

    return response;
  }
复制代码

可以看到，在 proceed 方法，又构造了 RealInterceptorChain 并且调用了 interceptor.intercept 方法，

而这个方法中又会调用 next.proceed 方法，直至返回 response 。这个过程有点像递归调用。

0x05 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();

    Call call();

    int connectTimeoutMillis();

    Chain withConnectTimeout(int timeout, TimeUnit unit);

    int readTimeoutMillis();

    Chain withReadTimeout(int timeout, TimeUnit unit);

    int writeTimeoutMillis();

    Chain withWriteTimeout(int timeout, TimeUnit unit);
  }
}
复制代码

所有的拦截器都需要实现这个接口。

0x06 异步的情况

public final class AsynchronousGet {
  private final OkHttpClient client = new OkHttpClient();

  public void run() throws Exception {
    Request request = new Request.Builder()
        .url("http://publicobject.com/helloworld.txt")
        .build();
	//调用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 {
       	//这里获取到响应结果数据
      }
    });
  }
复制代码

然后我们再看 RealCall 中的 enqueue 方法

@Override public void enqueue(Callback responseCallback) {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
    captureCallStackTrace();
    eventListener.callStart(this);
    //最终执行了dispatcher的enqueue方法
    client.dispatcher().enqueue(new AsyncCall(responseCallback));
  }
复制代码

其实是执行了 dispatcher 中的 enqueue 方法

synchronized void enqueue(AsyncCall call) {
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
      runningAsyncCalls.add(call);
      executorService().execute(call);
    } else {
      readyAsyncCalls.add(call);
    }
  }
复制代码

在 dispatcher 中通过线程池来执行 AsyncCall 对象，因此跟进到 AsyncCall 中的 execute 方法

@Override protected void execute() {
      boolean signalledCallback = false;
      try {
        //最终还是调用了getResponseWithInterceptorChain()！！！
        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 {
          eventListener.callFailed(RealCall.this, e);
          responseCallback.onFailure(RealCall.this, e);
        }
      } finally {
        client.dispatcher().finished(this);
      }
    }
复制代码

发现最终还是执行了 getResponseWithInterceptorChain ，因此不管是同步还是异步、最终的流程还是一样。