揭秘Android常用三方框架源码-okhttp

Posted 2021-07-14 码农 小生

okhttp是一个处理网络请求的开源项目,是安卓端最火热的轻量级框架,由移动支付Square公司贡献(该公司还贡献了Picasso) 用于替代HttpUrlConnection和Apache HttpClient。作为常用框架，无论是从使用还是面试方面，我们都有必要了解其源码实现，不能仅停留在api的调用层面。

在讲解其源码之前，先来简单回顾一下okhttp的使用。

一、基本使用

1.1 导入

通过gradle引入依赖

implementation 'com.squareup.okhttp3:okhttp:3.14.0'
implementation 'com.squareup.okio:okio:2.8.0'

okhttp目前最新可用版本为4.9.1，位于maven中心仓库，使用kotlin语言编写，本篇文章以公司某个项目所用版本进行分析，为java语言。

1.2 使用

//同步请求，需放在子线程
lifecycleScope.launch {
            withContext(Dispatchers.IO) {
                val syncCall =
                    OkHttpClient().newCall(Request.Builder().url("https://www.baidu.com/").get().build()).execute()
                toast("sync call result is ${syncCall.body()?.string()}")
            }
        }
 //异步请求
 OkHttpClient().newCall(Request.Builder().url("https://www.baidu.com/").get().build())
                .enqueue(object : Callback {
                    override fun onFailure(call: Call, e: IOException) {
                    }

                    override fun onResponse(call: Call, response: Response) {
                        toast("sync call result is ${response.body()?.string()}")
                    }
                })

通过上面代码，我们就能以最快速度使用okhttp(代码中toast为项目中扩展出来的方法，当做原生toast的使用即可)。

二、源码中关键类

2.1 OkHttpClient

用于发送http请求并读取其响应，调用newcall()创建call对象，对于一个项目来说，okhttpclient应该以单例模式提供实例，通过复用连接池和线程池减少延迟并节省内存，使okhttp性能最佳。

okhttpclient采用构建者模式创建实例的同时，为其属性赋值。

• 通过OkHttpClient()获取实例采用默认设置
• 通过new OkHttpClient.Builder()获取实例时可以自定义配置（连接超时、读取超时、写入超时、拦截器、缓存信息等）

关于构建者模式，我们在平常开发中也可以使用，比如封装页面titleBar时，标题栏左、中、右三个区域即可指定文本、又可指定图片、或者文本图片混合，以及三个区域各个元素左上右下间距、文本大小、样式、点击事件等等。通过构建者模式，可以将大量的属性聚合在一起，通过链式调用设置属性，将调用与实现分离，既美观又便于维护。

2.2 Request

一个request就代表一个Http请求，内部封装了url、请求方式、请求头、请求体等信息，也是通过构建者模式创建。

2.3 Call

一个call代表一个准备好要执行的request，可以被取消掉，不能被多次执行。其实现类为RealCall，call提供同步调用方法execute()、异步调用方法enqueue(Callback responseCallback)。

2.4 AsyncCall

AsyncCall是RealCall的内部类，是NamedRunnable接口的实现类，我们只要知道它是一个Runnable类型的实现类即可，异步请求的处理实际由该类的executeOn()方法完成。

2.5 Dispatcher

调度请求的类，内部有最大请求数（默认64）、每个主机最大请求数（默认5）、线程池、三个双端队列（准备执行的异步队列readyAsyncCalls、正在执行的异步队列runningAsyncCalls、正在执行的同步队列runningSyncCalls）等属性。

关于线程池,其实现如下：

  Dispatcher.java

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

• 核心线程数量corePoolSize = 0
• 最大线程数量maximumPoolSize = Integer.MAX_VALUE，即线程数量几乎无限制
• keepAliveTime = 60s，线程空闲60s后自动结束。
• workQueue为SynchronousQueue 同步队列，这个队列类似于一个接力棒，入队出队必须同时传递，因为线程创建无限制，不会有队列等待，所以使用SynchronousQueue。

关于双端队列Deque(double ended queue):

可以作为FIFO(First In First Out)的queue,也可以当做LIFO(Last In First Out)的stack使用，push和pop方法只在stack的形式下可以使用。不像list一样，双端队列不支持根据索引获取元素。

其常用api总结如下：

• boolean add(E e)：在尾部添加元素，如果队列有数量限制，超出数量添加抛出IllegalStateException
• boolean offer(E e)：在尾部添加元素，如果队列有数量限制，超出数量返回false，不抛异常
• E remove()：检索并移除队列头部元素，如果队列为空，抛出NoSuchElementException
• E poll()：检索并移除队列头部元素，如果队列为空，返回null,不抛异常
• E element()：检索不移除队列头部元素，如果队列为空，抛出NoSuchElementException
• E peek()：检索不移除队列头部元素，如果队列为空，返回null，不抛出异常
• void push(E e)：将元素压入双端队列代表的栈中（队列头部，栈专用方法），如果队列有数量限制，超出数量添加抛出IllegalStateException
• E pop()：从此双端队列代表的堆栈中弹出一个元素（移除并返回队列头部，栈专用方法），如果队列为空，抛出NoSuchElementException

三、源码分析

3.1 同步请求方式分析

同步请求由execute()方法开始，先看一下RealCall的execute()实现

  RealCall.java

  @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);//请求完成
    }
  }

其中，执行请求的方法实现如下：

  Dispatcher.java

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

就是把当前的同步请求call放入了Dispatcher的runningSyncCalls中。

看下请求完成的方法实现：

Dispatcher.java

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

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

把该请求从同步请求队列runningSyncCalls中移除。如此看来，对同步请求的处理就发生在了RealCall$getResponseWithInterceptorChain()中。

总结一下：

调用execute()进行同步请求时大致分三步：

• 请求加入runningSyncCalls队列
• RealCall$getResponseWithInterceptorChain()调用
• 请求从runningSyncCalls队列移除

关于getResponseWithInterceptorChain()的分析后面会讲。

3.2 异步请求方式分析

异步请求由enqueue(CallBack callBack)方法开始，看一下RealCall的enqueue(CallBack callBack)后会执行哪些操作

RealCall.java

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后，会用我们创建的响应回调生成一个AsyncCall，并且调用Dispatcher的enqueue()

Dispatcher.java

 void enqueue(AsyncCall call) {
    synchronized (this) {
      readyAsyncCalls.add(call);//将异步请求加入异步请求准备队列readyAsyncCalls

      // 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中可执行请求移除并加入到runningAsyncCalls
  }  

在Dispatcher的enqueue()中，首先会将异步请求放入异步请求准备队列readyAsyncCalls，让后调用promoteAndExecute()方法，将readyAsyncCalls中可执行请求移除并加入到runningAsyncCalls。

 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;//正在请求的同步请求、异步请求数量是否大于0
    }

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

    return isRunning;
  }

在遍历readyAsyncCalls的时候，依次检测正在执行的请求数是否达到阈值、同一主机下正在请求数是否达到阈值，通过前面检测后，从readyAsyncCalls中移除请求，并加入到局部变量executableCalls、正在执行队列runningAsyncCalls中。遍历executableCalls，调用AsyncCall$executeOn()。

RealCall.java

   RealCall$AsyncCall

    void executeOn(ExecutorService executorService) {
      assert (!Thread.holdsLock(client.dispatcher()));
      boolean success = false;
      try {
        executorService.execute(this);//线程池执行Runnable
        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!
        }
      }
    }

在executeOn()中，线程池会触发AsyncCall的run(),AsyncCall内没有run()方法，看下其父类NameRunnable中run()的定义

  NamedRunnable.java

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

run()中调用了execute()，AsyncCall实现了该抽象方法，所以我们还是得看AsyncCall中execute()的实现。

RealCall.java

RealCall$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);//请求完成
      }
    }

在AsyncCall的execute()中，才是异步请求真正开始的地方。和同步请求一样，异步请求最终也是通过RealCall$getResponseWithInterceptorChain()获取到响应，并且在请求成功或失败后回调给应用层。我们看下异步请求结束后调用的finished()。

 Dispatcher.java

 /** Used by {@code AsyncCall#run} to signal completion. */
  void finished(AsyncCall call) {
    call.callsPerHost().decrementAndGet();//主机请求数量递减
    finished(runningAsyncCalls, call);
  }

异步请求结束后，会使当前请求对应主机请求数减一，然后调用Dispatcher$finished(),不同于同步请求，这里的第一个参数传的是runningAsyncCalls，会将该请求从runningAsyncCalls中移除。

总结一下：

调用enqueue(Callback callback)进行异步请求时大致分以下几步：

• 使用回调callback生成一个AsyncCall,将AsyncCall放入readyAsyncCalls中。
• 遍历readyAsyncCalls，将符合条件的请求从readyAsyncCalls移除并加入runningAsyncCalls，局部变量executableCalls中
• 遍历executableCalls，调用AsyncCall.executeOn(),在线程池中触发AsyncCall.execute()
• RealCall.getResponseWithInterceptorChain()调用
• 回调结果，请求从runningAsyncCalls队列移除

至此，同步请求和异步请求大体流程我们已经梳理出来了，殊途同归，二者的核心实现都是RealCall.getResponseWithInterceptorChain()，下面分析下该方法。

3.3 getResponseWithInterceptorChain()分析

getResponseWithInterceptorChain()代码如下：

  RealCall.java

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

该方法主要做了两件事

• 把自定义拦截器和框架提供的拦截器按照顺序加入到list中，形成拦截器链。
• 拦截器链从index=0的地方开始，取出RealInterceptorChain类型的实例，依次调用proceed(),生成Response，逐层返回。

我们在看下RealInterceptorChain$proceed()

  RealInterceptorChain.java

  public Response proceed(Request request, Transmitter transmitter, @Nullable Exchange exchange)
      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.exchange != null && !this.exchange.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.exchange != 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, transmitter, exchange,
        index + 1, request, call, 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 (exchange != 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;
  }

RealInterceptorChain$proceed()里面做了很多校验，比如：同一请求域名和端口是否改变、请求是否执行多次、响应是否为空、响应体是否为空，在该方法里最重要的是获取到当前拦截器的下一个拦截器，并且进入下一拦截器的proceed()中，也就是说该拦截器在未获取到响应时，下发了请求给下一个拦截器，直到拦截器链的最后一个，响应从最后一个逐层返回到index=0的拦截器。

拦截器添加的顺序与其执行完的顺序刚好相反，最后添加的CallServerInterceptor是第一个执行完产生Response的，而最先添加的拦截器确实最后一个执行完的。

拦截器链的请求下发类似于android系统中的事件传递模型，但是两者又有不同，事件下发顺序二者一致，都是自顶向下，直至树最底层或list最后一个元素，但是对于事件传递来讲，子view消费了事件，事件就不会在各父view处理了，而拦截器最后一个执行完后，Response会逐层包装返给上一个拦截器处理，直到链index=0拦截器proceed执行完为止。

拦截器详解传送门

四、总结

一图以蔽之