OkHttpOkHttp 源码分析 ( 同步 / 异步 Request 请求执行原理分析 )

Posted 2021-06-30 韩曙亮

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了OkHttpOkHttp 源码分析 ( 同步 / 异步 Request 请求执行原理分析 )相关的知识，希望对你有一定的参考价值。

OkHttp 系列文章目录

【OkHttp】OkHttp 源码分析 ( 网络框架封装 | OkHttp 4 迁移 | OkHttp 建造者模式 )
【OkHttp】OkHttp 源码分析 ( OkHttpClient.Builder 构造器源码分析 )
【OkHttp】OkHttp 源码分析 ( 同步 / 异步 Request 请求执行原理分析 )

一、分析 OkHttp 执行原理

以 OkHttp 同步 / 异步 Get 请求为例 , 分析底层的运行细节 ;

    /**
     * OkHttp 异步 Get 请求
     */
    private void httpAsynchronousGet() {
    	// 初始化 OkHttp 
		OkHttpClient mOkHttpClient = new OkHttpClient();

        // Request 中封装了请求相关信息
        Request request = new Request.Builder()
                .url("https://www.baidu.com")   // 设置请求地址
                .get()                          // 使用 Get 方法
                .build();

        // 异步 Get 请求
        mOkHttpClient.newCall(request).enqueue(new Callback(){

            @Override
            public void onFailure(Call call, IOException e) {
                // 请求失败的情况
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                // 请求成功 , 获取
                String result = response.body().string();
                Log.i(TAG, "result : " + result);
                runOnUiThread(new Runnable() {
                    @Override
                    public void run() {
                        // 主线程中执行相关代码
                    }
                });
            }
        });
    }

1、创建 OkHttpClient

创建 OkHttpClient : 调用者调用 OkHttpClient 构造函数 , 创建 OkHttpClient , 然后返回给调用者 ;

OkHttpClient mOkHttpClient = new OkHttpClient();

OkHttpClient 构造函数中 , 实际上创建了自身的创建者 ;

  public OkHttpClient() {
    this(new 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;
    }
  }

2、创建 Request

创建 Request 时 , 使用 Request 的创建者 Request.Builder 创建该 Request 对象 ;

先调用 Request.Builder 的构造函数 , 创建 Request.Builder 对象 , 然后调用 Request.Builder 的 build 方法 , 创建 Request 对象 ;

// Request 中封装了请求相关信息
Request request = new Request.Builder()
        .url("https://www.baidu.com")   // 设置请求地址
        .get()                          // 使用 Get 方法
        .build();

3、获取 RealCall

调用 OkHttpClient 对象的 newCall 方法 , 发起新的请求调用 , 返回 $1$ 个 RealCall 类型对象 ;

mOkHttpClient.newCall(request)

在 OkHttpClient 的 newCall 方法中 , 创建了 RealCall , 并返回给了调用者 ;

  /**
   * Prepares the {@code request} to be executed at some point in the future.
   */
  @Override public Call newCall(Request request) {
    return RealCall.newRealCall(this, request, false /* for web socket */);
  }

RealCall 实现了 Call 接口 ;

final class RealCall implements Call {
  final OkHttpClient client;
}

Call 接口提供的功能 : execute 是同步请求 , enqueue 是异步请求 ;

public interface Call extends Cloneable {
  Request request();

  Response execute() throws IOException;

  void enqueue(Callback responseCallback);

  void cancel();

  boolean isExecuted();

  boolean isCanceled();

  Timeout timeout();

  Call clone();

  interface Factory {
    Call newCall(Request request);
  }
}

4、通过 RealCall 发送同步 / 异步 Request 请求

RealCall 实现了上述 Call 接口的各项功能 , 主要关注其实现 Call 接口的 execute 同步请求方法 , enqueue 异步请求方法 ;

final class RealCall implements Call {
  @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);
    }
  }

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

( 1 ) 、同步 Request 请求

同步请求方法 , 返回一个责任链 , 在该方法中可以清楚的看到 OkHttp 的 Get 请求具体做了哪些步骤 ;
在该方法中通过添加不同功能的拦截器 , 实现相关业务路基 ;

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

( 2 ) 、异步 Request 请求

在 RealCall 的 enqueue 异步请求方法中 , 最终调用的还是 OkHttpClient 的 dispatcher 进行调度 ;

在上一篇博客【OkHttp】OkHttp 源码分析 ( OkHttpClient.Builder 构造器源码分析 ) 已经提到过 OkHttpClient 的 Dispatcher dispatcher 成员 , 是 Get / Post 方法的请求线程调度器 ;

final class RealCall implements Call {
  final OkHttpClient client;
  @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));
  }
}

二、OkHttp 异步 Request 请求源码分析

异步 Request 请求涉及到线程调度 , 比较复杂 ;

OKHttpClient 调用 newCall 获取 RealCall , 然后调用 RealCall 的 enqueue 方法进行异步 Get/Post 请求 , 在该方法中最终调用 OKHttpClient 对象中的 Dispatcher dispatcher 线程调度器的 enqueue 方法 , 进行异步请求 ;

1、Dispatcher 调度器 enqueue 方法分析

在 Dispatcher 的 enqueue 方法中 , 调用了 findExistingCallWithHost 方法获取 AsyncCall , 然后在方法最后调用了 promoteAndExecute 进行后续执行异步任务操作 ;

/**
 * Policy on when async requests are executed.
 *
 * <p>Each dispatcher uses an {@link ExecutorService} to run calls internally. If you supply your
 * own executor, it should be able to run {@linkplain #getMaxRequests the configured maximum} number
 * of calls concurrently.
 */
public final class Dispatcher {
  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();
  }
  
  @Nullable private AsyncCall findExistingCallWithHost(String host) {
    for (AsyncCall existingCall : runningAsyncCalls) {
      if (existingCall.host().equals(host)) return existingCall;
    }
    for (AsyncCall existingCall : readyAsyncCalls) {
      if (existingCall.host().equals(host)) return existingCall;
    }
    return null;
  }
}

AsyncCall 继承了 NamedRunnable , NamedRunnable 实现了 Runnable 接口 , AsyncCall 本质是 Runnable ;

final class AsyncCall extends NamedRunnable
public abstract class NamedRunnable implements Runnable

2、Dispatcher 调度器 promoteAndExecute 方法分析

分析 promoteAndExecute 方法 : 将符合条件的调用从 readyAsyncCalls 提升为 runningAsyncCalls , 并且在线程池中调用它们 ; 这些操作必须同步调用 , 因为执行这些调用需要调用用户代码 ;

最终的异步请求执行调用的是 AsyncCall 的 executeOn 方法 ;

      AsyncCall asyncCall = executableCalls.get(i);
      asyncCall.executeOn(executorService());

Dispatcher | promoteAndExecute 方法源码 :

  /**
   * Promotes eligible calls from {@link #readyAsyncCalls} to {@link #runningAsyncCalls} and runs
   * them on the executor service. Must not be called with synchronization because executing calls
   * can call into user code.
   *
   * @return true if the dispatcher is currently running calls.
   */
  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;
  }

3、AsyncCall 的 executeOn 方法分析

AsyncCall 的 executeOn 方法中 , 主要使用了传入的 ExecutorService executorService 线程池 , 执行异步请求任务 ;

RealCall $ AsyncCall | executeOn 方法代码 :

final class RealCall implements Call {
  final class AsyncCall extends NamedRunnable {
    /**
     * 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!
        }
      }
    }
  }
}