本文参考，感谢大佬

网络上有不少源码解析，参考了部分大佬的解析，本文是日常开发中对自己的总结。

• OkHttp 的使用

OkHttp 依赖 : implementation 'com.squareup.okhttp3:okhttp:3.14.2'

  //获取 OkHttpClient 对象
  OkHttpClient httpClient = new OkHttpClient.Builder().build();
  //获取 Request 对象
  Request request = new Request.Builder().build();
  //获取 Call 对象
  Call call = httpClient.newCall(request);

  //同步方法
  Response response = call.execute();

OkHttpClient-Builder

  public static final class Builder {
    ...
    ...
    public Builder() {
      dispatcher = new Dispatcher();
      protocols = DEFAULT_PROTOCOLS;
      connectionSpecs = DEFAULT_CONNECTION_SPECS;
      eventListenerFactory = EventListener.factory(EventListener.NONE);
      proxySelector = ProxySelector.getDefault();
      if (proxySelector == null) {
        proxySelector = new NullProxySelector();
      }
      cookieJar = CookieJar.NO_COOKIES;
      socketFactory = SocketFactory.getDefault();
      hostnameVerifier = OkHostnameVerifier.INSTANCE;
      certificatePinner = CertificatePinner.DEFAULT;
      proxyAuthenticator = Authenticator.NONE;
      authenticator = Authenticator.NONE;
      connectionPool = new ConnectionPool();
      dns = Dns.SYSTEM;
      followSslRedirects = true;
      followRedirects = true;
      retryOnConnectionFailure = true;
      callTimeout = 0;
      connectTimeout = 10_000;
      readTimeout = 10_000;
      writeTimeout = 10_000;
      pingInterval = 0;
    }
  }

Builder 是 OkHttpClient 中的一个静态内部类，用来初始化参数

Request-Builder

  public final class Request {
  final HttpUrl url;
  final String method;
  final Headers headers;
  final @Nullable RequestBody body;
  final Map<Class<?>, Object> tags;

  private volatile @Nullable CacheControl cacheControl; // Lazily initialized.

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

Builder 是 Request 中的一个静态内部类，也是用来初始化参数

Call-httpClient.newCall()

  @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.transmitter = new Transmitter(client, call);
    return call;
  }

实际上创建的对象是 Call 的实现类 RealCall 对象
(Map map = new HashMap() List list = new ArrayList() 都是如此)

Response-call.execute()

之前拿到的是一个 RealCall 对象，所以看 RealCall 中的 execute（）

  @Override public Response execute() throws IOException {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
    transmitter.timeoutEnter();
    transmitter.callStart();
    try {
      //核心方法
      client.dispatcher().executed(this);
      //重点方法
      return getResponseWithInterceptorChain();
    } finally {
      //核心方法
      client.dispatcher().finished(this);
    }
  }

同步代码块中，首先判断是否执行，如果执行就抛出异常，如果没有执行，则将 executed 置为 true。

在执行 newRealCall 方法时，初始化了一个 Transmitter 对象，分别执行了 Transmitter 的两个方法。分别是超时控制和绑定 call。

核心方法

• client.dispatcher().executed
• client.dispatcher().finished
• getResponseWithInterceptorChain()

executed

  public Dispatcher dispatcher() {
    return dispatcher;
  }

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

dispatcher 方法返回了一个 Dispatcher 对象，然后调用了 executed 方法，往 runningSyncCalls 中添加了一个 Call 对象，Dispatcher 对象中维护了3中队列

  /** 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<>();

getResponseWithInterceptorChain

  Response getResponseWithInterceptorChain() throws IOException {
    // Build a full stack of interceptors.
    List<Interceptor> interceptors = new ArrayList<>();
    interceptors.addAll(client.interceptors());
    interceptors.add(new RetryAndFollowUpInterceptor(client));
    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, transmitter, null, 0,
        originalRequest, this, client.connectTimeoutMillis(),
        client.readTimeoutMillis(), client.writeTimeoutMillis());

    boolean calledNoMoreExchanges = false;
    try {
      Response response = chain.proceed(originalRequest);
      if (transmitter.isCanceled()) {
        closeQuietly(response);
        throw new IOException("Canceled");
      }
      return response;
    } catch (IOException e) {
      calledNoMoreExchanges = true;
      throw transmitter.noMoreExchanges(e);
    } finally {
      if (!calledNoMoreExchanges) {
        transmitter.noMoreExchanges(null);
      }
    }
  }

创建了一个拦截器集合，放入很多拦截器，然后返回了一个 Response 对象。

finished

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

调用了重载方法，然后调用了 promoteAndExecute 方法

  private <T> void finished(Deque<T> calls, T call) {
    Runnable idleCallback;
    synchronized (this) {
      if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
      idleCallback = this.idleCallback;
    }

    boolean isRunning = promoteAndExecute();

    if (!isRunning && idleCallback != null) {
      idleCallback.run();
    }
  }

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

同步代码块中迭代了 readyAsyncCalls（异步请求就绪队列） ，同步请求流程分析时只往 runningSyncCalls（异步请求执行队列） 中添加了数据，所以不满足当前循环执行的条件，当前返回 false。所以会执行 idleCallback.run() 方法。

  //异步方法
  call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {

            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {

            }
        });

call.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));
  }

同步代码块中，首先判断是否执行，如果执行就抛出异常，如果没有执行，则将 executed 置为 true。

在执行 newRealCall 方法时，初始化了一个 Transmitter 对象，分别执行了 Transmitter 的绑定 call 方法。

executed(RealCall call)
enqueue(AsyncCall call),首先看看 AsyncCall 对象

AsyncCall

  final class AsyncCall extends NamedRunnable {
    private final Callback responseCallback;
    private volatile AtomicInteger callsPerHost = new AtomicInteger(0);

    AsyncCall(Callback responseCallback) {
      super("OkHttp %s", redactedUrl());
      this.responseCallback = responseCallback;
    }

    AtomicInteger callsPerHost() {
      return callsPerHost;
    }

    void reuseCallsPerHostFrom(AsyncCall other) {
      this.callsPerHost = other.callsPerHost;
    }

    String host() {
      return originalRequest.url().host();
    }

    Request request() {
      return originalRequest;
    }

    RealCall get() {
      return RealCall.this;
    }

    /**
     * Attempt to enqueue this async call on {@code executorService}. This will attempt to clean up
     * if the executor has been shut down by reporting the call as failed.
     */
    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!
        }
      }
    }

    @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);
        }
      } finally {
        client.dispatcher().finished(this);
      }
    }
  }

继承了 NamedRunnable ，而 NamedRunnable 实现了 Runnable 接口，我们可以把 AsyncCall 理解为一个线程类

接下来看看 enqueue 方法

enqueue

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

      // Mutate the AsyncCall so that it shares the AtomicInteger of an existing running call to
      // the same host.
      if (!call.get().forWebSocket) {
        AsyncCall existingCall = findExistingCallWithHost(call.host());
        if (existingCall != null) call.reuseCallsPerHostFrom(existingCall);
      }
    }
    promoteAndExecute();
  }

同步代码块中 readyAsyncCalls（异步请求就绪队列）添加 Call 对象。（同步请求中 runningSyncCalls（异步请求执行队列）添加 Call 对象 ）

区别

同步请求 execute() 是 runningSyncCalls（异步请求执行队列）添加对象
异步请求 enqueue() 是 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;
  }

readyAsyncCalls（异步请求就绪队列）刚才添加了对象，所以不为空，我们先看第一个 for 循环

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

• 便利 readyAsyncCalls （异步请求就绪队列）
• (条件1) runningAsyncCalls （异步请求的执行队列）的长度小于最大请求数 64
• (条件2) 当前主机请求数小于最大主机请求数 5（默认）
• 添加到临时队列 executableCalls
• 添加到 runningAsyncCalls （异步请求的执行队列）

接下来看第二个 for 循环

  for (int i = 0, size = executableCalls.size(); i < size; i++) {
      AsyncCall asyncCall = executableCalls.get(i);
      asyncCall.executeOn(executorService());
  }

• 便利临时队列 executableCalls

executorService

  public synchronized ExecutorService executorService() {
    if (executorService == null) {
      executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
          new SynchronousQueue<>(), Util.threadFactory("OkHttp Dispatcher", false));
    }
    return executorService;
  }

• 返回一个线程池对象

所以真正的异步处理的方法是 executeOn，下面我们看看 executeOn方法

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() 可以理解成线程池对象
执行了execute() 方法，之前说过 AsyncCall 可以理解为一个线程类，它继承了 NamedRunnable ，所以我们看看 NamedRunnable这个类。

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

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);
        }
      } finally {
        client.dispatcher().finished(this);
      }
    }

也是调用 getResponseWithInterceptorChain() ，返回一个 Response 对象，最后在 finally 中执行了 finished() 方法。

最后大致的总结一下同步和异步执行的流程。

同步

• RealCall.execute() ➡️ Dispatcher.execute() ---runningSyncCalls（异步请求执行队列）添加 call 对象
• getResponseWithInterceptorChain() 返回 Response 对象
• Dispatcher.finished()

异步

• RealCall. enqueue() ➡️ Dispatcher. enqueue()---readyAsyncCalls（异步请求就绪队列）添加 call 对象
• 便利 readyAsyncCalls（异步请求就绪队列），将满足条件的添加到临时队列和runningAsyncCalls （异步请求的执行队列）
• getResponseWithInterceptorChain() 返回 Response 对象
• Dispatcher.finished()

避免不了有写错的地方，如果有写错的地方，可以通过留言评论。如果您觉得这篇文章对您有帮助，希望您帮忙点个赞，感谢！