OkHttp源码解析(二):异步请求

2018-09-06 本文已影响4人 zYoung_Tang

1.内容

Okhttp 异步请求 enqueue() 源码分析
任务调度器 Dispatcher

依赖

implementation 'com.squareup.okhttp3:okhttp:3.11.0

2.例子

OkHttpClient client = new OkHttpClient();
Request request = new Request
                .Builder()
                .url("https://www.baidu.com")
                .method("GET", null)
                .build();
Call call = client.newCall(request);
call.enqueue(new Callback() {
    @Override
    public void onFailure(Call call, IOException e) {
    }

    @Override
    public void onResponse(Call call, Response response) throws IOException {
        System.out.println(response.body().string());
    }
});

OkHttp基本使用步骤如下:

创建 OkHttpClient 实例
创建 Request 实例
通过 OkHttpClient 和 Request 封装成一个请求 Call
同步请求调用 Call.execute()，异步请求调用 Call.enqueue()

3.源码分析

直接查看步骤 4 中的 enqueue()

// RealCall.java
@Override public void enqueue(Callback responseCallback) {
    synchronized (this) { // 如果这个 call 已经被执行过，抛异常
        if (executed) throw new IllegalStateException("Already Executed");
        executed = true;
    }
    captureCallStackTrace();
    eventListener.callStart(this);
    client.dispatcher().enqueue(new AsyncCall(responseCallback));
}

核心代码为最后一行，可以分解为三步:

AsyncCall call = new AsyncCall(responseCallback);
Dispatcher dispatcher = client.dispatcher();
dispatcher.enqueue(call);

把 responseCallback 封装成 AsyncCall (实际上是一个runnable，提供给线程池执行)
返回了一个 Dispatcher

// OkHttpClient.java
public Dispatcher dispatcher() {
    return dispatcher;
}

调用任务调度器 Dispatcher 的 enqueue() 异步执行 call

首先了解一下 Dispatcher

// Dispatcher.java
public final class Dispatcher {
    // 同步请求和异步请求之和最大值
    private int maxRequests = 64;
    // 同一个 host 请求数最大值
    private int maxRequestsPerHost = 5;
    private @Nullable Runnable idleCallback;

    /** Executes calls. Created lazily. */
    /** 用于执行请求的线程池，且是懒加载的 */
    private @Nullable ExecutorService executorService;

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

   ...
       
    public synchronized ExecutorService executorService() {
        if (executorService == null) {
             // 核心线程数为0，最大线程数为 Integer.MAX_VALUE ，空闲线程最多存活60秒
            executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
        }
        return executorService;
    }
    
    synchronized void enqueue(AsyncCall call) {
      if (runningAsyncCalls.size() < maxRequests 
          && runningCallsForHost(call) < maxRequestsPerHost) {
        // 如果正在执行的异步请求数没有达到最大值
        // 就放到 runningAsyncCalls 标记为正在执行
        runningAsyncCalls.add(call);
        // 用 executorService 执行这个异步请求
        executorService().execute(call);
      } else { 
        // 如果正在执行的异步请求数达到最大值
        // 就放到 readyAsyncCalls 标记为等待执行
        readyAsyncCalls.add(call);
      }
    }
    
  ...
}

Dispatcher 是任务调度器，内部建立了一个线程池 ExecutorService ，而且维护了三个集合:

readyAsyncCalls : 等待被执行的异步请求集合
runningAsyncCalls : 正在执行的异步请求集合，包括已经被取消但未完成的请求
runningSyncCalls : 正在执行的同步请求集合，包括已经被取消但未完成的请求

所有异步请求都交由线程池 ExecutorService 来执行。

线程池其实是 ThreadPoolExecutor ，且核心线程数为 0 、最大线程数为Integer.MAX_VALUE、空闲线程存活最大时间为60秒，即所有线程执行完之后空闲60秒就会被销毁，而且存在线程过多导致内存溢出问题等问题，但是在 Dispatcher 的调度下是不会发生线程过多情况的，因为 Dispatcher 限制了正在执行的请求数(同步和异步之和)最大为64，同一个host下请求同时存在数最大值为 5 。

线程池会调用线程执行 AsyncCall 的 execute()，下面直接查看源码:

// RealCall.java
final class AsyncCall extends NamedRunnable {
  
  ...
        
  @Override protected void execute() {
    boolean signalledCallback = false;
    try {
      // 通过拦截器链得到从服务器获取的响应 Response
      Response response = getResponseWithInterceptorChain();
      // 如果 retryAndFollowUpInterceptor.cancel() 被调用过就报异常
      if (retryAndFollowUpInterceptor.isCanceled()) {
        signalledCallback = true; // 标记 callback 回调函数已被调用
        responseCallback.onFailure(RealCall.this, new IOException("Canceled"));
      } else {
        // 到这里表示获取响应成功
        signalledCallback = true; // 标记 callback 回调函数已被调用
        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 {
      // 最后要通知 dispatcher 标记该任务已完成
      client.dispatcher().finished(this);
    }
  }
}

AsyncCall 的 execute() 逻辑很简单，getResponseWithInterceptorChain() 我们已经在上篇文章中了解过了，获取 Response 之后只需要判断是回调 responseCallback 的 onFailure() 还是 onResponse()，所以 enqueue() 中的回调方法是在子线程中被调用的，当然最后还要调用 finished() 通知 Dispatcher 该任务已经完成了，需要从runningAsyncCalls中移除该任务。

// Dispatcher
void finished(AsyncCall call) {
  finished(runningAsyncCalls, call, true);
}

private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
  int runningCallsCount;
  Runnable idleCallback;
  synchronized (this) {
    // 把已经完成的异步任务从 runningAsyncCalls 中移除
    if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
    if (promoteCalls) promoteCalls();
    runningCallsCount = runningCallsCount();
    idleCallback = this.idleCallback;
  }

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

readyAsyncCalls 和 runningAsyncCalls 之间的调度

Dispatcher 是如何判断 runningAsyncCalls 有空余位置，并从 readyAsyncCalls 中获取添加到 runningAsyncCalls 的呢？

Dispatcher 的判断就在每个异步任务结束时调用的 finish(call) 内

// Dispatcher.java
// 异步请求
void finished(AsyncCall call) {
    finished(runningAsyncCalls, call, true);
}
// 同步请求
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!");
        // 异步请求 promoteCalls 为 true，同步请求为 false
        if (promoteCalls) promoteCalls();
        runningCallsCount = runningCallsCount();
        idleCallback = this.idleCallback;
    }

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

public synchronized int runningCallsCount() {
    // 返回所有正在运行的同步异步请求总数
    return runningAsyncCalls.size() + runningSyncCalls.size();
}

同步请求和异步请求执行后最终都会调用 dispatcher.finish()，内部执行的区别是异步请求会调用 promoteCalls()，目的就是对 readyAsyncCalls 和 runningAsyncCalls 进行调度

// Dispatcher.java
private void promoteCalls() {
    // 1.当正在运行的同步请求异步请求数大于64时直接 return
    if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity.
    // 2.没有等待执行的异步请求的时候 return
    if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote.
    // 3.可以从 readyAsyncCalls 中取任务放到 runningAsyncCalls 中并执行
    for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
        // 从迭代器获取下一个等待的异步任务
        AsyncCall call = i.next();
       // 对同一个 host 请求数不能超过 5
        if (runningCallsForHost(call) < maxRequestsPerHost) {
            // 从 readyAsyncCalls 中删除 call
            i.remove();
            // 把 call 添加到 runningAsyncCalls
            runningAsyncCalls.add(call);
            // 使用线程池执行 call
            executorService().execute(call);
        }
        // 一直执行 for 循环直到 runningAsyncCalls 数达到 64 个
        if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
    }
}

总结

异步请求的回调方法在子线程中.
异步请求其实就是把 Call 封装成 Runnable ，交给线程池调用子线程执行.
最多同时请求数为 64.
相同 Host 的请求最多同时执行 5 个.

OkHttp源码解析(二):异步请求

1.内容

依赖

2.例子

3.源码分析

readyAsyncCalls 和 runningAsyncCalls 之间的调度

总结

猜你喜欢

热点阅读