深入理解OkHttp源码——提交请求

Posted 2020-10-01 laughingQing

本篇文章主要介绍OkHttp执行同步和异步请求的大体流程。主要流程如下图：

主要分析到getResponseWidthInterceptorChain方法，该方法为具体的根据请求获取响应部分，留着后面的博客再介绍。

Dispatcher类

Dispatcher类负责异步任务的请求策略。首先看它的部分定义：

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

  /** Executes calls. Created lazily. */
  private 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<>();

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

  ...

}

内部有一个线程池，三个队列，分别是readyAsyncCalls、runningAsyncCalls和runningSyncCalls队列，其中runningSyncCalls用于存储同步请求RealCall对象，runningAsyncCalls队列用于存储进入队列的异步请求AsyncCall对象，而readyAsyncCalls队列用于当runningAsyncCalls的尺寸达到maxRequests参数时(默认64)存储新加的异步请求。至于为什么要什么做呢？
我的理解是为了避免一时间创造大量的线程浪费资源，那么为什么有线程池，还要用到这样一个控制策略呢？这是因为创建默认线程池的参数导致的。默认的executorService的创建类似于Executors.newCachedThreadPool，该线程池的问题在于不会限制线程数量，如果一下子需要开启1000乃至更多的线程，依然会开启，而OkHttp这儿在Dispacther中做了控制。待会儿在下面的分析中可以看到这种控制策略。
其余的参数maxRequestsPerHost表示每个主机的最大请求数，，默认为5，比如说如果这时好多个异步请求都是请求百度上面的图片，如果达到了5，那么新的请求就会被放入到readyAsyncCalls队列中，等该主机的请求数降下去后才会再次执行。
而参数idleCallback是Dispatcher中请求数量为0时的回调，这儿的请求包含同步请求和异步请求，该参数默认为null。
在Dispatcher中，需要明白一点，尽管同步请求自己负责执行请求，但是依然会先加入到Dispatcher的同步队列，完成后从队列中移除，而异步请求则完全属于Dispatcher控制，但是有些方法是对所有请求操作的，有些则是对异步请求操作的，需要特别注意。
一般地，我们会将OkHttpClient作为单例，而Dispatcher是其一个成员，自然也是单例，所以一般整个应用的所有请求都会经过Dispatcher，不论是同步请求还是异步请求。

同步请求的执行流程

在使用OkHttp进行网络同步异步操作中知道了如何进行同步请求，创建一个Request对象，然后再创建一个Call对象，调用Call对象的execute方法即可。那么就从execute方法看起。Call是一个接口，具体实现是RealCall，下面是RealCall的execute方法实现：

@Override public Response execute() throws IOException {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
    try {
      client.dispatcher().executed(this);
      Response result = getResponseWithInterceptorChain();
      if (result == null) throw new IOException("Canceled");
      return result;
    } finally {
      client.dispatcher().finished(this);
    }
  }

首先是设置executed标志为true，同一个Call只允许执行一次，执行多次就会抛出异常。接下来是调用OkHttpClient的dispatcher()方法获得Dispatcher对象，然后调用其executed(RealCall)方法，然后就是调用getResponseWithInterceptorChain方法同步获取响应，最后调用Dispatcher的finished方法，下面先看executed方法：

/** Used by {@code Call#execute} to signal it is in-flight. */
  synchronized void executed(RealCall call) {
    runningSyncCalls.add(call);
  }

从代码中可以看出，Dispatcher的executed方法只是将同步请求加入到了runningSyncCalls队列中。下面再看finished方法：

/** Used by {@code Call#execute} to signal completion. */
  void finished(RealCall call) {
    finished(runningSyncCalls, call, false);
  }

从上面代码中，可以看到finished方法再调用另一个finished方法，并将runningSyncCalls队列传入，具体实现如下：

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

 

首先是从队列中移除请求，如果不能移除，则抛出异常；在上面finished方法调用中看出传入第三个参数为false，所以不会调用promoteCalls方法，该参数用于异步请求时为true，这个下面分析异步请求时再讲。然后调用runningCallsCount统计目前还在运行的请求，最后，如果正在运行的请求数为0表示Dispatcher中没有可运行的请求了，进入Idle状态，此时如果idleCallback不为null，则调用其run方法。下面是runningCallsCount()方法的实现：

public synchronized int runningCallsCount() {
    return runningAsyncCalls.size() + runningSyncCalls.size();
  }

可以看到这个方法返回的请求包括同步请求和异步请求。
至此，同步请求的执行流程分析完成，可以看到Dispatcher只是保存了一下同步请求和移除同步请求，而对于异步请求，Dispatcher的工作就不只是这么简单了。

异步请求的执行流程

我们知道如果要发起异步请求，那么就调用Call的enqueue方法并传入回调，依然从RealCall的enqueue方法看起：

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

可以看到，依然是首先将executed参数设为true，同样地，异步请求也不可以被执行两次，然后调用Dispatcher的enqueue方法，但是这儿涉及到了一个新的类，AsyncCall。AsyncCall是RealCall的一个内部类并且继承NamedRunnable，那么首先看NamedRunnable类是什么样的，如下：

/**
 * Runnable implementation which always sets its thread name.
 */
public abstract class NamedRunnable implements Runnable {
  protected final String name;

  public NamedRunnable(String format, Object... args) {
    this.name = Util.format(format, args);
  }

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

  protected abstract void execute();
}

可以看到NamedRunnable实现了Runnbale接口并且是个抽象类，其抽象方法是execute()，该方法是在run方法中被调用的，这也就意味着NamedRunnable是一个任务，并且其子类应该实现execute方法。下面再看AsyncCall的实现：

final class AsyncCall extends NamedRunnable {
    private final Callback responseCallback;

    private AsyncCall(Callback responseCallback) {
      super("OkHttp %s", redactedUrl().toString());
      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 {
        client.dispatcher().finished(this);
      }
    }
  }

AsyncCall实现了execute方法，首先是调用getResponseWithInterceptorChain()方法获取响应，然后获取成功后，就调用回调的onReponse方法，如果失败，就调用回调的onFailure方法。最后，调用Dispatcher的finished方法。
由于AsyncCall的execute()方法是在run中被调用的，所以getResponseWithInterceptorChain是在非调用线程中被调用的，然后得到响应后再交给Callback。
从上面的流程看出，与Dispatcher的交互主要涉及enqueue方法和finished方法，与同步请求类似。下面先看enqueue方法：

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

首先如果正在运行的异步请求的数量小于maxRequests并且与该请求相同的主机数量小于maxRequestsPerHost，也就是说符合放入runningAsyncCalls队列的要求，那么放入队列，然后将AsyncCall交给线程池；如果不符合，那么就放入到readyAsyncCalls队列中。
当线程池执行AsyncCall任务时，它的execute方法会被调用，getResponseWithInterceptorChain()会去获取响应，最后调用Dispatcher的finished方法，下面看finished方法：

/** Used by {@code AsyncCall#run} to signal completion. */
  void finished(AsyncCall call) {
    finished(runningAsyncCalls, call, true);
  }

从上面的代码可以看出，与同步请求的finished方法不同的是第一个参数传入的是正在运行的异步队列，第三个参数为true，下面再看有是三个参数的finished方法：

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

与同步请求相同的是，移除请求，获取运行数量判断是否进入了Idle状态，不同的是会调用promoteCalls()方法，下面是promoteCalls()方法：

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.
    }
  }
promoteCalls方法主要负责从异步等待队列中将请求移步到异步运行队列中。主要就是遍历等待队列，并且需要满足同一主机的请求小于maxRequestsPerHost时，就移到运行队列中并交给线程池运行。

至此，分析完了同步请求和异步请求的提交流程，Dispatcher负责异步请求是放入运行队列还是等待队列中，并且在每个异步请求执行完后，需要判断是否需要把等待队列中的请求移到运行队列中并运行。不管是同步请求还是异步请求，最终都会调用getResponseWithInterceptorChain()方法进行具体的网络请求，该方法下篇博客深入理解OkHttp源码（二）——获取响应会具体介绍。