分析Okhttp源码的一篇笔记

1.OkHttp的请求网络流程

使用就不说了 Okhttp官网:http://square.github.io/okhttp/
(1) 从请求处理开始分析

OkHttpClient.newCall(request) 进行 execute 或者 enqueue操作 当调用newCall方法时

 @Override public Call newCall(Request request) {
    return new RealCall(this, request);
  }

看到返回的是一个RealCall类, 调用enqueue 异步请求网络实际调用的是RealCall的enqueue方法，接下来看一下RealCall的enqueue里面干了什么

  void enqueue(Callback responseCallback, boolean forWebSocket) {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
//最终的请求是dispatcher来完成
    client.dispatcher().enqueue(new AsyncCall(responseCallback, forWebSocket));
  }

So 接下来分析一下dispatcher

(2) Dispatcher任务调度
Dispatcher 主要是控制并发的请求，它主要维护一下变量：

//最大并发请求数
  private int maxRequests = 64;  
//每个主机的最大请求数
  private int maxRequestsPerHost = 5;
  /**  消费者线程 */
  private ExecutorService executorService;
  /** 将要运行的异步请求队列 */
  private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
  /**  正在运行的异步请求队列*/
  private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
  /** 正在运行的同步请求队列 */
  private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

Dispatcher构造方法

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

两个构造方法，可以使用自己设定的线程池，如果没有设定，则会请求网络前自己创建默认线程池。这个线程池类似CachedThreadPool,比较适合执行大量的耗时比较少的任务，其实当调用RealCall的enqueue方法实际上调用的Dispatcher里面的enqueue方法

  synchronized void enqueue(AsyncCall call) {
//如果正在运行的异步请求队列小于64且 正在运行的请求主机数小于5 则把请求加载到runningAsyncCalls
//中并在线程池中执行，否则就加入到readyAsyncCalls中进行缓存等待。
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
      runningAsyncCalls.add(call);
      executorService().execute(call);
    } else {
      readyAsyncCalls.add(call);
    }
  }

线程池中传进来的参数AsyncCall，它是RealCall的内部类，其内部也实现了execute方法

  @Override protected void execute() {
      boolean signalledCallback = false;
      try {
       ...
      } catch (IOException e) {
       ...
      } finally {
  //无论这个请求的结果如何，都会执行   client.dispatcher().finished(this);
        client.dispatcher().finished(this);
      }
    }
  }

finished方法如下

 
  synchronized void finished(AsyncCall call) {
 //将此次请求 runningAsyncCalls移除后还执行promoteCalls方法
    if (!runningAsyncCalls.remove(call)) throw new AssertionError("AsyncCall wasn't running!");
    promoteCalls();
  }

promoteCalls方法如下

  private void promoteCalls() {
//如果正在运行的异步请求队列数大于最大请求数 return
    if (runningAsyncCalls.size() >= maxRequests) return;
   //若果将要运行的异步请求队列为空  return
    if (readyAsyncCalls.isEmpty()) return;  

    for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
      AsyncCall call = i.next();//取出下一个请求 

      if (runningCallsForHost(call) < maxRequestsPerHost) {
        i.remove();
        runningAsyncCalls.add(call);//加入runningAsyncCalls中并交由线程池处理
        executorService().execute(call);
      }
      if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
    }
  }

在回AsyncCall的execute方法

 @Override protected void execute() {

      try {
 //getResponseWithInterceptorChain() 返回Response   请求网络
        Response response = getResponseWithInterceptorChain(forWebSocket);
        ...
      } catch (IOException e) {
       ...
      } finally {
     ...
      }
    }
  }

(3) Interceptor拦截器
getResponseWithInterceptorChain方法如下

private Response getResponseWithInterceptorChain(boolean forWebSocket) throws IOException {
    Interceptor.Chain chain = new ApplicationInterceptorChain(0, originalRequest, forWebSocket);
    return chain.proceed(originalRequest);
  }

getResponseWithInterceptorChain方法中创建了ApplicationInterceptorChain，他是一个拦截器链，这个类也是RealCall的内部类，接下来执行了它的proceed方法

    @Override public Response proceed(Request request) throws IOException {
      
      if (index < client.interceptors().size()) {
        Interceptor.Chain chain = new ApplicationInterceptorChain(index + 1, request, forWebSocket);
       //从拦截器列表中取出拦截器
        Interceptor interceptor = client.interceptors().get(index);
        Response interceptedResponse = interceptor.intercept(chain);//1

        if (interceptedResponse == null) {
          throw new NullPointerException("application interceptor " + interceptor
              + " returned null");
        }

        return interceptedResponse;
      }

      // 如果没有更多拦截器的话 执行网络请求
      return getResponse(request, forWebSocket);
    }
  }

proceed方法每次从拦截器列表中去除拦截器，当存在多个拦截器时都会在上面注释1处阻塞，并等待下一个拦截器的调用返回，下面分别以拦截器中有一个、两个拦截器的场景加以模拟

拦截器模拟场景.png

拦截器是一种能够监控、重写、重试调用的机制。通常情况下用来添加、移除、转换请求和响应的头部信息，比如将域名替换为IP地址，在请求中添加host属性，也可以添加我们应用中的一下公共参数，比如设备id、版本号等，回到代码上来， 最后一行返回getResponse(request, forWebSocket)
来看getResponse做了什么

 Response getResponse(Request request, boolean forWebSocket) throws IOException {

   engine = new HttpEngine(client, request, false, false, forWebSocket, null, null, null);

  int followUpCount = 0;
    while (true) {
      if (canceled) {
        engine.releaseStreamAllocation();
        throw new IOException("Canceled");
      }
     boolean releaseConnection = true;
      try {
        engine.sendRequest();
        engine.readResponse();
        releaseConnection = false;
      } catch (RequestException e) {
 
        throw e.getCause();
      } catch (RouteException e) {

      } catch (IOException e) {
       ... 
      }
      }

}

创建了HttpEngine并且调用了HttpEngine的sendRequest方法和readResponse方法

(4) 缓存策略
查看一下sendRequest方法

 public void sendRequest() throws RequestException, RouteException, IOException {
    if (cacheStrategy != null) return; // Already sent.
    if (httpStream != null) throw new IllegalStateException();

    Request request = networkRequest(userRequest);
  // 获取client中的Cache，同时Cache在初始化时会读取缓存目录中曾经请求过的所有信息
    InternalCache responseCache = Internal.instance.internalCache(client);
    Response cacheCandidate = responseCache != null
        ? responseCache.get(request)//1
        : null;

    long now = System.currentTimeMillis();
    cacheStrategy = new CacheStrategy.Factory(now, request, cacheCandidate).get();
   //网络请求
    networkRequest = cacheStrategy.networkRequest;
 //缓存的响应
    cacheResponse = cacheStrategy.cacheResponse;

    if (responseCache != null) {
   //记录当前请求是网络发起还是缓存发起
      responseCache.trackResponse(cacheStrategy);
    }


   

    // 不进行网络请求并且缓存不存在或者过期，则返回504
    if (networkRequest == null && cacheResponse == null) {
      userResponse = new Response.Builder()
          .request(userRequest)
          .priorResponse(stripBody(priorResponse))
          .protocol(Protocol.HTTP_1_1)
          .code(504)
          .message("Unsatisfiable Request (only-if-cached)")
          .body(EMPTY_BODY)
          .build();
      return;
    }

    // 不进行网络请求而且缓存可以使用，则直接返回缓存
    if (networkRequest == null) {
      userResponse = cacheResponse.newBuilder()
          .request(userRequest)
          .priorResponse(stripBody(priorResponse))
          .cacheResponse(stripBody(cacheResponse))
          .build();
      userResponse = unzip(userResponse);
      return;
    }

    // 需要访问网络时
    boolean success = false;
    try {
      httpStream = connect();
      httpStream.setHttpEngine(this);

      ...
    }  
  }

显然是发送请求，但是最主要的是做了缓存的策略，上面注释1处cacheCandidate 是上次与服务器交互缓存的Response,这里缓存均基于Map,key是请求中的url的md5,value是在文件中查询到的缓存，页面置换基于LRU算法，现在只需知道cachCandidate是一个可以读取缓存Header的Response即可，根据cacheStrategy的处理得到了networkRequest和cacheResponse都是为nulld的情况下，也就是不进行网络请求并且缓存不存在或者过期，这个是返回504错误，当networkRequest为null时也就是不进行网络请求，如果缓存可以使用时则直接返回缓存，其他则请求网络。

接下来查看readResponse方法

 public void readResponse() throws IOException {

     ... 
    Response networkResponse;

    if (forWebSocket) {
    //读取网络响应
      networkResponse = readNetworkResponse();
    }  

 
    receiveHeaders(networkResponse.headers());

  //检查缓存是否可用，如果可用，就用当前缓存的Response，关闭网络连接，释放连接
    if (cacheResponse != null) {
      if (validate(cacheResponse, networkResponse)) {//1
        userResponse = cacheResponse.newBuilder()
            .request(userRequest)
            .priorResponse(stripBody(priorResponse))
            .headers(combine(cacheResponse.headers(), networkResponse.headers()))
            .cacheResponse(stripBody(cacheResponse))
            .networkResponse(stripBody(networkResponse))
            .build();
        networkResponse.body().close();
        releaseStreamAllocation();

        // Update the cache after combining headers but before stripping the
        // Content-Encoding header (as performed by initContentStream()).
        InternalCache responseCache = Internal.instance.internalCache(client);
        responseCache.trackConditionalCacheHit();
        responseCache.update(cacheResponse, stripBody(userResponse));
        userResponse = unzip(userResponse);
        return;
      } else {
        closeQuietly(cacheResponse.body());
      }
    }

);

  }

这个方法主要用来解析HTTP响应报文，如果有缓存并且可用，则用缓存的数据并更新缓存，否则就用网络 请求返回的数据，查看注释1处validate方法是如何判断缓存是否可用。

  private static boolean validate(Response cached, Response network) {
 //如果服务器返回304，则缓存有效
    if (network.code() == HTTP_NOT_MODIFIED) {
      return true;
    }

   
//通过缓存和网络请求响应中的Last-Modified来计算是否是最新数据，如果是，则缓存有效
    Date lastModified = cached.headers().getDate("Last-Modified");
    if (lastModified != null) {
      Date networkLastModified = network.headers().getDate("Last-Modified");
      if (networkLastModified != null
          && networkLastModified.getTime() < lastModified.getTime()) {
        return true;
      }
    }

    return false;
  }

如果缓存有效，则返回304 Not Modified，否者直接返回body。如果缓存过期或者强制放弃缓存，则缓存策略全部交给服务器判断，客服端只需要发送条件GET请求即可。条件GET请求有两种方式:一种是Last-Modified-Data，另一种ETag,这里采用Last-Modified-Data，通过缓存和网络请求响应中的Last-Modified来计算是否是最新数据 ，如果是 则缓存有效

(5) 失败重连
重回RealCall的getResponse方法

 Response getResponse(Request request, boolean forWebSocket) throws IOException {
    
      ...
   
      try {
   
      }catch (RouteException e) {
       
       HttpEngine retryEngine = engine.recover(e.getLastConnectException(), null);//1
      } catch (IOException e) {
    
        HttpEngine retryEngine = engine.recover(e, null);//2
     
      }
  }
}

当发生IOException或者RouteException时都会执行HttpEngine的recover方法 ，代码如下:

  public HttpEngine recover(IOException e, Sink requestBodyOut) {
    if (!streamAllocation.recover(e, requestBodyOut)) {
      return null;
    }

    if (!client.retryOnConnectionFailure()) {
      return null;
    }

    StreamAllocation streamAllocation = close();

    //重新创建HttpEngine并返回，用来完成重连
    return new HttpEngine(client, userRequest, bufferRequestBody, callerWritesRequestBody,
        forWebSocket, streamAllocation, (RetryableSink) requestBodyOut, priorResponse);
  }

Okhttp请求流程图

image.png