开源库之 OkHttp 源码分析

技术答疑，成长进阶，可以加入我的知识星球：音视频领域专业问答的小圈子

分析一波 OkHttp 的源码实现。

简单使用

官方给出了使用例子，具体详情参考官网。

 1// 创建 OkHttp 请求客户端
 2OkHttpClient client = new OkHttpClient();
 3// 构建一个请求
 4Request request = new Request.Builder()
 5      .url(url)
 6      .build();
 7// 执行网络请求并返回结果      
 8Response response = client.newCall(request).execute();      
 9// 得到结果内容
10response.body().string();
JAVA

可以看到，使用过程还是很容易理解的：

首先创建请求客户端 -> 接着创建网络请求 -> 然后发出网络请求 -> 最后得到请求结果。

把网络请求过程抽象化之后就是上面的流水线步骤了，其中最重要的步骤就是发出网络请求了，在 OkHttp 中用到了责任链模式，对发出的网络请求会按照责任链表的顺序依次进行处理，比如请求重试、连接处理、缓存处理、日志处理等，这个地方再写一篇文章详细分析。

下面针对 OkHttp 的调用流程和结构作分析。

创建请求

首先是创建请求客户端 OkHttpClient 和网络请求 Request 。

对于这种要创建某某对象的，上来就是一个建造者模式，建造者模式可以说是在开源项目中最常见的设计模式了。

在 OkHttpClient 和 Request 内部都有一个 Builder 的内部类用来执行具体的构建操作，一般 Builder 类有很多方法，它们都是用来设置具体构建参数的，顺着哪些方法就可以找到都有哪些配置选项，这些选项也就是对外暴露的接口，这算是阅读源码的一个小技巧了。

OkHttpClient 的 Builder 内部配置选项：

    dispatcher = new Dispatcher();
    // 使用默认的选项
    protocols = DEFAULT_PROTOCOLS;
    connectionSpecs = DEFAULT_CONNECTION_SPECS;
    eventListenerFactory = EventListener.factory(EventListener.NONE);
    proxySelector = ProxySelector.getDefault();
    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;
    connectTimeout = 10_000;
    readTimeout = 10_000;
    writeTimeout = 10_000;
    pingInterval = 0;
JAVA

重点分析 Dispatcher 类用来管理和分发请求的，其他的如果没有指定的话都是默认选项，等用到时再具体分析。

在 Request 内部的 Builder 也有一些配置选项：

  HttpUrl url;
  String method;
  Headers.Builder headers;
  RequestBody body;
  /** A mutable map of tags, or an immutable empty map if we don't have any. */
  Map<Class<?>, Object> tags = Collections.emptyMap();
JAVA

像 url、body、method 这些都是 HTTP 请求相关的内容，用一个 Request 类将它们进行封装，到具体执行请求时，都会将它们取出来使用。

同步执行

1val response = client.newCall(request).execute()
KOTLIN

同步执行的代码如上，首先是 newCall 方法将 request 转换成 RealCall 对象，它实现了 Call 的接口。

Request 类只是封装了请求的信息，比如 GET、POST 方法之类的，但具体的执行通过 Call 这一层来实现了。

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.eventListener = client.eventListenerFactory().create(call);
  return call;
}
JAVA

然后是 Call 的 execute 方法。

@Override public Response execute() throws IOException {
  // 保证线程安全
  synchronized (this) {
    if (executed) throw new IllegalStateException("Already Executed");
    executed = true;
  }
  captureCallStackTrace();
  eventListener.callStart(this);
  try {
    // 把请求添加到队列中
    client.dispatcher().executed(this);
    // 具体执行网络请求
    Response result = getResponseWithInterceptorChain();
    if (result == null) throw new IOException("Canceled");
    return result;
  } catch (IOException e) {
    // 请求失败的回调
    eventListener.callFailed(this, e);
    throw e;
  } finally {
    // 将请求从队列中移除
    client.dispatcher().finished(this);
  }
}
JAVA

首先是 synchronized 加锁保证线程安全，然后是通过 Dispatcher 类的 executed 方法将请求 RealCall 添加到请求的同步队列中去。

1private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
2
3synchronized void executed(RealCall call) {
4  runningSyncCalls.add(call);
5}
JAVA

runningSyncCalls 是一个 Deque 类型的队列。Deque 是一个双端队列，它具有队列和栈的特性，队列中的元素可以从两端弹出，而插入和删除操作只能在队列的两端进行。

executed 方法只是将请求添加到了队列中，具体的执行在 getResponseWithInterceptorChain 方法中，在这里就会将请求通过OkHttp 的各种拦截器，按照责任链模式进行调用，并得到请求返回的结果，用 Response 类封装。

当请求结束后，会调用 Dispatcher 类的 finished 方法。

void finished(RealCall call) {
  finished(runningSyncCalls, call, false);
}

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;
  }
15
  if (runningCallsCount == 0 && idleCallback != null) {
    idleCallback.run();
  }
}  
JAVA

finished 方法主要是将请求 RealCall 从队列中 runningSyncCalls 移除。

这里还有个小知识点就是 try、catch、finally 三者的执行顺序，在 try 语句里面执行了 return 语句，但是返回的结果会保存在一个临时区域里面，然后执行 finally 语句，移除队列请求后再继续返回。

网络请求

下面重点看一下 getResponseWithInterceptorChain 方法。

Response getResponseWithInterceptorChain() throws IOException {
  // Build a full stack of interceptors.
  List<Interceptor> interceptors = new ArrayList<>();
  interceptors.addAll(client.interceptors());
  // OkHttpClient 提供的默认拦截器
  interceptors.add(retryAndFollowUpInterceptor);
  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));
  // 重点参数是 0 ，表示责任链的开始
  Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
      originalRequest, this, eventListener, client.connectTimeoutMillis(),
      client.readTimeoutMillis(), client.writeTimeoutMillis());
18
  return chain.proceed(originalRequest);
}
JAVA

有一个拦截器 Interceptor 的集合，在创建 OkHttpClient 的时候可以通过 Builder 去添加我们自定义的拦截器，另外 OkHttp 也提供了几个默认的拦截器。

把拦截器添加到集合中，作为 RealInterceptorChain 类的参数，它实现了 Chain 接口，表示为一条链。在这条链上，每个拦截器都是它的一个节点，并且以链上任何一个拦截器为起点，又可以开始一条新的链。

具体来看看 proceed 方法：

@Override public Response proceed(Request request) throws IOException {
  return proceed(request, streamAllocation, httpCodec, connection);
}

public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
    RealConnection connection) throws IOException {
  if (index >= interceptors.size()) throw new AssertionError();
 8
  calls++;
  // 删除一些判断和抛出异常的方法
  
  // Call the next interceptor in the chain.
  // 以拦截器队列中的下一个拦截器为起点，构建新的请求链
  RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec,
      connection, index + 1, request, call, eventListener, connectTimeout, readTimeout,
      writeTimeout);
  // 取出当前请求链中的第一个
  Interceptor interceptor = interceptors.get(index);
  // 执行当前拦截器的功能，并且开始子链的请求流程
  Response response = interceptor.intercept(next);
21
  // 删除一些判断和抛出异常的方法
  return response;
}
JAVA

整个拦截器的集合 interceptors 是不会变的，而 index 就对应从拦截器集合中取出拦截器的索引，为 0 表示取出第一个来作为整条 链 的起点，从而构建一条请求链。

当一个拦截器执行具体功能时，也就是 intercept 方法，会把下一个链作为参数传递过去，这样就又会以这条 链 为起始继续下一步的执行。

比如，看 HttpLoggingInterceptor 内部的实现:

  Request request = chain.request();
  if (level == Level.NONE) {
    return chain.proceed(request);
  }
JAVA

如果要打印的日志即为为 NONE，也就是不打印，直接就开始 链 的下一个执行了。

简单地看，这里就是一个递归的调用流程。

异步执行

1client.newCall(request).enqueue(object : okhttp3.Callback {
2
3    override fun onFailure(call: okhttp3.Call?, e: IOException?) {
4    }
5    
6    override fun onResponse(call: okhttp3.Call?, response: okhttp3.Response?) {
7    }
8})
KOTLIN

异步请求的代码如上，和同步请求的区别在于后面是 enqueue 方法。

@Override public void enqueue(Callback responseCallback) {
  synchronized (this) {
    if (executed) throw new IllegalStateException("Already Executed");
    executed = true;
  }
  captureCallStackTrace();
  eventListener.callStart(this);
  // 通过 Dispatcher 管理请求
  // 
  client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
JAVA

enqueue 需要传入一个 Callback 的回调接口，用来处理请求成功和失败。另外，在 enqueue 的内部还是通过 Dispatcher 进行请求的管理。

这里传入的参数是 AsyncCall 类，不再是 RealCall 类了，AsyncCall 类继承自 NamedRunnable 类，NamedRunnable 实现了 Runnable 接口。

NamedRunnable 意思就是有名字的线程，会在线程执行时临时改变线程的名字，执行结束后再改回来。

@Override public final void run() {
  // 保存当前线程的名字
  String oldName = Thread.currentThread().getName();
  // 设置新的名字
  Thread.currentThread().setName(name);
  try {
    execute();
  } finally {
    // 执行后再改为原来的名字
    Thread.currentThread().setName(oldName);
  }
}
JAVA

另外，AsyncCall 实现了 NamedRunnable 的抽象方法 execute ，该方法也是网络请求的具体执行部分，问题在于这些异步请求是如何分发和管理的，还是回到 Dispatcher 类中来。

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

首先是判断异步请求队列 runningAsyncCalls 的个数是否超过最大请求数了，另外是判断同一主机的网络请求个数是否超过限制。

如果都不超过，就将请求添加到异步请求队列中，并在线程池中去执行。

如果超过了，就把请求添加到准备就绪的队列 readyAsyncCalls 中，等待后续再去执行。

executorService() 方法的执行就是初始化线程池的。

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

ThreadPoolExecutor 用来提供一个线程池，它的方法原型如下：

  public ThreadPoolExecutor(int corePoolSize,             // 核心线程池的大小
                            int maximumPoolSize,        // 线程池的最大大小
                            long keepAliveTime,         // 线程池空闲时，线程的存活时间
                            TimeUnit unit,             // 存活时间的时间单位
                            BlockingQueue<Runnable> workQueue, // 存放线程任务的队列
                            ThreadFactory threadFactory,      // 创建线程的工厂
                            RejectedExecutionHandler handler) {     // 任务拒绝策略
          // 省略
  }
JAVA

使用线程池的好处是显而易见的，统一管理网络请求，减少线程频繁创建、销毁带来的开销。

通过线程池的 execute 方法去执行请求，具体就是 Runnable 接口中的 run 方法。

  // 异步请求中的方法
  @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 {
        // 请求失败的回调
        eventListener.callFailed(RealCall.this, e);
        responseCallback.onFailure(RealCall.this, e);
      }
    } finally {
      client.dispatcher().finished(this);
    }
  }
JAVA

异步请求执行的线程变了，肯定就得有相应的回调通知，另外，具体是网络请求部分还是在 getResponseWithInterceptorChain 方法中。

在 finally 中还是会将异步的请求从队列中移除，不同的是内部最后一个参数为 false 。

void finished(AsyncCall call) {
  // 与同步请求不同的是，最后一个参数为 True, 同步请求为 false
  finished(runningAsyncCalls, call, true);
}
JAVA

因为 True，所以 promoteCalls 就会执行。

// 异步请求才会执行该方法
private void promoteCalls() {
  // 优先把 readyAsyncCalls 队列中的请求执行完
  if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity.
  if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote.
 6
  for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
    AsyncCall call = i.next();
 9
    if (runningCallsForHost(call) < maxRequestsPerHost) {
      i.remove();
      runningAsyncCalls.add(call);
      executorService().execute(call);
    }
15
    if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
  }
}
JAVA

之前提到，如果异步请求队列已经满了，就会把请求放到准备就绪的队列 readyAsyncCalls 中去，那么 promoteCalls 方法就是来处理 readyAsyncCalls 队列中的请求的。

每次从队列中移除异步请求后 size 都会减一，然后再判断队列的数量是否还是超过了最大请求数量，如果是就返回，表示优先把异步队列中的请求消耗完，如果不是，表示有空位，那就从准备就绪的队列中取出请求来执行。

把 readyAsyncCalls 队列中的请求放到 runningAsyncCalls 队列中去，再通过线程池去执行，并且还是会判断 runningAsyncCalls 的数量是否超过了最大请求，超过了就返回，表示 OkHttp 中整个异步请求的数量不能超过 maxRequests 表示的数量。

就这样实现了 OkHttp 的异步请求管理流程，并且还有一个类似请求排队的机制，

小结

OkHttp 是用来执行网络请求，包括同步和异步的请求。

对于同步请求，直接就执行了，同步阻塞直到返回取得请求结果。

对于异步请求，对请求进行管理，限制请求最大数量，如果超出数量就排队候选，每次执行完一次异步请求，有空位就去处理排队的请求。

对于具体的请求过程，通过责任链的模式，把请求分成多个过程，递归地对请求进行处理。

欢迎关注微信公众号：音视频开发进阶