Android OKHttp框架的分发器与拦截器源码刨析

1.OKHttp简单使用
2.OkHttp分发器源码解析
2.1.同步请求方式
2.2.异步的方式进行请求
2.3.分发器线程池
3.OkHttp的拦截器
3.1.RetryAndFollowUpInterceptor 重试和重定向拦截器
3.2.桥接拦截器BridgeInterceptor
3.3.缓存拦截器CacheInterceptor
3.4.连接拦截器ConnectInterceptor
3.5.请求服务器拦截器CallServerInterceptor
4.OkHttp总结

1.OKHttp简单使用

 
OkHttpClient okHttpClient=new OkHttpClient.Builder().build();
Request request=new Request.Builder()
        .url("http://www.baidu.com")
        .get().build();
Call call = okHttpClient.newCall(request);
try {
    Response execute = call.execute();
    Log.d("lpf","execute"+execute.body());
} catch (IOException e) {
    e.printStackTrace();
}

这样就使用OkHttp做了一次网络请求，使用流程：

通过建造者的方式创建OkHttpClient
通过建造者的方式创建Request
调用OkHttpClient的newCall方法将Request对象当做参数传进去创建一个Call对象
调用call的execute（这个是同步的方式）方法就完成了一次网络请求，如果调用enqueue方法就是进行的一步请求。

2.OkHttp分发器源码解析

2.1.同步请求方式

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

同步请求的方式很简单，就是通过调用dispatcher分发器的executed方法，将请求call放到同步请求队列runningSyncCalls中，然后执行getResponseWithInterceptorChain，经过各个拦截器的处理就能拿到请求结果，然后进行返回就好。

这里用到来了runningSyncCalls同步请求队列，异步的时候也会有队列可以做比对。

2.2.异步的方式进行请求

 
public void enqueue(Callback responseCallback) {
    synchronized (this) { //不能冲突调用
        if (executed) throw new IllegalStateException("Already Executed");
        executed = true;
    }
    captureCallStackTrace();
    eventListener.callStart(this);
    client.dispatcher().enqueue(new AsyncCall(responseCallback)); //OkHttp会提供默认的分发器
}

通过调用dispatcher的enqueue方法进行异步call分发

Dispatch的enqueue方法

 
synchronized void enqueue(AsyncCall call) { //分发器 分发任务  
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
        runningAsyncCalls.add(call);  //正在执行的请求
        executorService().execute(call); //线程池跑任务
    } else {
        readyAsyncCalls.add(call);   //ready 中的任务什么时候执行
    }
}

如果正在执行的请求小于64，相同host的请求不能超过5个就放入运行队列runningAsyncCalls
否则就放到准备队列readyAsyncCalls

如果在运行队列的call就会马上放入executorService线程池进行执行

AsyncCall的execute方法：

 
 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);  //这里的代码肯定执行
        }
    }
}

getResponseWithInterceptorChain：通过拦截器进行网络请求处理，重试被取消了回调失败，否则正常回调onResponse。

注意finally里边的内容，最后会执行dispatcher().finished(this)

dispatcher的finish方法

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

这个里边就是处理的队列任务的切换，如果任务执行完成之后，将运行队列的call移除，然后查看是否可以将等待队列的任务放入运行队列。

2.3.分发器线程池

分发器就是来调配请求任务的，内部会包含一个线程池。当异步请求时，会将请求任务交给线程池来执行。那分发器中默认的线程池是如何定义的呢

 
 public synchronized ExecutorService executorService() {
        if (executorService == null) {  //这个线程池的特点是高并发 最大吞吐量
            executorService = 
                new ThreadPoolExecutor(, //核心线程数量
                                      Integer.MAX_VALUE,//最大线程数量, 				//空闲线程闲置时间														TimeUnit.SECONDS,   //闲置时间单位
                    				new SynchronousQueue<Runnable>(),//线程等待队列 
                    				Util.threadFactory("OkHttp Dispatcher",false)//线程工厂
                                      );
        }
        return executorService;
    }

首先核心线程为0，表示线程池不会一直为我们缓存线程，线程池中所有线程都是在60s内没有工作就会被回收。而最大线程 Integer.MAX_VALUE 与等待队列 SynchronousQueue 的组合能够得到最大的吞吐量。即当需要线程池执行任务时，如果不存在空闲线程不需要等待，马上新建线程执行任务！等待队列的不同指定了线程池的不同排队机制。一般来说，等待队列 BlockingQueue 有： ArrayBlockingQueue 、 LinkedBlockingQueue 与 SynchronousQueue 。

假设向线程池提交任务时，核心线程都被占用的情况下：

ArrayBlockingQueue ：基于数组的阻塞队列，初始化需要指定固定大小。

当使用此队列时，向线程池提交任务，会首先加入到等待队列中，当等待队列满了之后，再次提交任务，尝试加入

队列就会失败，这时就会检查如果当前线程池中的线程数未达到最大线程，则会新建线程执行新提交的任务。所以

最终可能出现后提交的任务先执行，而先提交的任务一直在等待。

LinkedBlockingQueue ：基于链表实现的阻塞队列，初始化可以指定大小，也可以不指定。

当指定大小后，行为就和 ArrayBlockingQueu 一致。而如果未指定大小，则会使用默认的 Integer.MAX_VALUE 作为队列大小。这时候就会出现线程池的最大线程数参数无用，因为无论如何，向线程池提交任务加入等待队列都会成功。最终意味着所有任务都是在核心线程执行。如果核心线程一直被占，那就一直等待。

SynchronousQueue : 无容量的队列。

使用此队列意味着希望获得最大并发量。因为无论如何，向线程池提交任务，往队列提交任务都会失败。而失败后

如果没有空闲的非核心线程，就会检查如果当前线程池中的线程数未达到最大线程，则会新建线程执行新提交的任

务。完全没有任何等待，唯一制约它的就是最大线程数的个数。因此一般配合 Integer.MAX_VALUE 就实现了真正的无等待。

但是需要注意的时，我们都知道，进程的内存是存在限制的，而每一个线程都需要分配一定的内存。所以线程并不

能无限个数。那么当设置最大线程数为 Integer.MAX_VALUE 时，OkHttp同时还有最大请求任务执行个数: 64的限

堂制。这样即解决了这个问题同时也能获得最大吞吐。

3.OkHttp的拦截器

RealCall的getResponseWithInterceptorChain方法，是OkHttp最核心的部分

 
Response getResponseWithInterceptorChain() throws IOException {
    // Build a full stack of interceptors.
    List<Interceptor> interceptors = new ArrayList<>();
    interceptors.addAll(client.interceptors()); //自定义拦截器加入到集合
    interceptors.add(retryAndFollowUpInterceptor);
    interceptors.add(new BridgeInterceptor(client.cookieJar()));
    interceptors.add(new CacheInterceptor(client.internalCache()));
    interceptors.add(new ConnectInterceptor(client));
    if (!forWebSocket) {//如果不是webSocket
        interceptors.addAll(client.networkInterceptors());
    }
    interceptors.add(new CallServerInterceptor(forWebSocket));
    Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null,,
            originalRequest, this, eventListener, client.connectTimeoutMillis(),
            client.readTimeoutMillis(), client.writeTimeoutMillis());
    return chain.proceed(originalRequest);
}

这个方法就是进行网络请求的入口方法，进过这个方法之后就拿到网络请求的Response结果了。

这个方法里边就是创建了一个List网络拦截器的集合，将拦截器添加到集合里边去。

retryAndFollowUpInterceptor：重试和重定向拦截器

client.interceptors()：用户自定义的拦截器
BridgeInterceptor：桥接拦截器
CacheInterceptor：缓存拦截器
ConnectInterceptor：连接拦截器
CallServerInterceptor：请求服务拦截器

创建链条RealInterceptorChain，调用链条的proceed方法

通过责任链设计模式，就会逐次访问每个拦截器，拦截器是一个U型的数据传递过程。

PS：责任链设计模式，对象行为型模式，为请求创建了一个接收者对象的链，在处理请求的时候执行过滤(各司职)。

责任链上的处理者负责处理请求，客户只需要将请求发送到责任链即可，无须关心请求的处理细节和请求的传递所以职责链将请求的发送者和请求的处理者解耦了。

3.1.RetryAndFollowUpInterceptor 重试和重定向拦截器

RetryAndFollowUpInterceptor的intercept方法

 
@Override
public Response intercept(Chain chain) throws IOException {
    Request request = chain.request();
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Call call = realChain.call();
    EventListener eventListener = realChain.eventListener();
    /**
     *管理类，维护了 与服务器的连接、数据流与请求三者的关系。真正使用的拦截器为 Connect
     */
    StreamAllocation streamAllocation = new StreamAllocation(client.connectionPool(),
            createAddress(request.url()), call, eventListener, callStackTrace);
    this.streamAllocation = streamAllocation;
    int followUpCount =;
    Response priorResponse = null;
    while (true) {
        if (canceled) {
            streamAllocation.release();
            throw new IOException("Canceled");
        }
        Response response;
        boolean releaseConnection = true;
        try {
            //请求出现了异常，那么releaseConnection依旧为true。
            response = realChain.proceed(request, streamAllocation, null, null);
            releaseConnection = false;
        } catch (RouteException e) {
            //路由异常，连接未成功，请求还没发出去
            //The attempt to connect via a route failed. The request will not have been sent.
            if (!recover(e.getLastConnectException(), streamAllocation, false, request)) {
                throw e.getLastConnectException();
            }
            releaseConnection = false;
            continue;
        } catch (IOException e) {
            //请求发出去了，但是和服务器通信失败了。(socket流正在读写数据的时候断开连接)
            // HTTP才会抛出ConnectionShutdownException。所以对于HTTP1 requestSendStarted一定是true
            //An attempt to communicate with a server failed. The request may have been sent.
            boolean requestSendStarted = !(e instanceof ConnectionShutdownException);
            if (!recover(e, streamAllocation, requestSendStarted, request)) throw e;
            releaseConnection = false;
            continue;
        } finally {
            // We're throwing an unchecked exception. Release any resources.
            //不是前两种的失败，那直接关闭清理所有资源
            if (releaseConnection) {
                streamAllocation.streamFailed(null);
                streamAllocation.release();
            }
        }
        //如果进过重试/重定向才成功的，则在本次响应中记录上次响应的情况
        //Attach the prior response if it exists. Such responses never have a body.
        if (priorResponse != null) {
            response = response.newBuilder()
                    .priorResponse(
                            priorResponse.newBuilder()
                                    .body(null)
                                    .build()
                    )
                    .build();
        }
        //处理和4xx的一些状态码，如301 302重定向
        Request followUp = followUpRequest(response, streamAllocation.route());
        if (followUp == null) {
            if (!forWebSocket) {
                streamAllocation.release();
            }
            return response;
        }
        closeQuietly(response.body());
        //限制最大 followup 次数为次
        if (++followUpCount > MAX_FOLLOW_UPS) {
            streamAllocation.release();
            throw new ProtocolException("Too many follow-up requests: " + followUpCount);
        }
        if (followUp.body() instanceof UnrepeatableRequestBody) {
            streamAllocation.release();
            throw new HttpRetryException("Cannot retry streamed HTTP body", response.code());
        }
        //todo 判断是不是可以复用同一份连接
        if (!sameConnection(response, followUp.url())) {
            streamAllocation.release();
            streamAllocation = new StreamAllocation(client.connectionPool(),
                    createAddress(followUp.url()), call, eventListener, callStackTrace);
            this.streamAllocation = streamAllocation;
        } else if (streamAllocation.codec() != null) {
            throw new IllegalStateException("Closing the body of " + response
                    + " didn't close its backing stream. Bad interceptor?");
        }
        request = followUp;
        priorResponse = response;
    }
}

这个拦截器：第一个接触到请求，最后接触到响应；负责判断是否需要重新发起整个请求

3.2.桥接拦截器BridgeInterceptor

BridgeInterceptor的intercept方法

 
public Response intercept(Chain chain) throws IOException {
    Request userRequest = chain.request();
    Request.Builder requestBuilder = userRequest.newBuilder();
    RequestBody body = userRequest.body();
    if (body != null) {
        MediaType contentType = body.contentType();
        if (contentType != null) {
            requestBuilder.header("Content-Type", contentType.toString());
        }
        long contentLength = body.contentLength();
        if (contentLength != -) {
            requestBuilder.header("Content-Length", Long.toString(contentLength));
            requestBuilder.removeHeader("Transfer-Encoding");
        } else {
            requestBuilder.header("Transfer-Encoding", "chunked");
            requestBuilder.removeHeader("Content-Length");
        }
    }
    if (userRequest.header("Host") == null) {
        requestBuilder.header("Host", hostHeader(userRequest.url(), false));
    }
    if (userRequest.header("Connection") == null) {
        requestBuilder.header("Connection", "Keep-Alive");
    }
    // If we add an "Accept-Encoding: gzip" header field we're responsible for also
  // decompressing
    // the transfer stream.
    boolean transparentGzip = false;
    if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
        transparentGzip = true;
        requestBuilder.header("Accept-Encoding", "gzip");
    }
    List<Cookie> cookies = cookieJar.loadForRequest(userRequest.url());
    if (!cookies.isEmpty()) {
        requestBuilder.header("Cookie", cookieHeader(cookies));
    }
    if (userRequest.header("User-Agent") == null) {
        requestBuilder.header("User-Agent", Version.userAgent());
    }
    Response networkResponse = chain.proceed(requestBuilder.build());
    HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());
    Response.Builder responseBuilder = networkResponse.newBuilder()
            .request(userRequest);
    if (transparentGzip
            && "gzip".equalsIgnoreCase(networkResponse.header("Content-Encoding"))
            && HttpHeaders.hasBody(networkResponse)) {
        GzipSource responseBody = new GzipSource(networkResponse.body().source());
        Headers strippedHeaders = networkResponse.headers().newBuilder()
                .removeAll("Content-Encoding")
                .removeAll("Content-Length")
                .build();
        responseBuilder.headers(strippedHeaders);
        String contentType = networkResponse.header("Content-Type");
        responseBuilder.body(new RealResponseBody(contentType, -L, Okio.buffer(responseBody)));
    }
    return responseBuilder.build();
}

BridgeInterceptor ，连接应用程序和服务器的桥梁，我们发出的请求将会经过它的处理才能发给服务器，比如设

置请求内容长度，编码，gzip压缩，cookie等，获取响应后保存Cookie等操作。

桥接拦截器的责任是：补全请求头，并默认使用gzip压缩，同时将响应体重新设置为gzip读取。

3.3.缓存拦截器CacheInterceptor

CacheInterceptor的intercept方法，缓存的判断条件比较复杂

 
public Response intercept(Chain chain) throws IOException {
    //通过url的md数据 从文件缓存查找 （GET请求才有缓存）
    Response cacheCandidate = cache != null
            ? cache.get(chain.request())
            : null;
    long now = System.currentTimeMillis();
    //缓存策略:根据各种条件(请求头)组成 请求与缓存
    CacheStrategy strategy =
            new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();
    //
    Request networkRequest = strategy.networkRequest;
    Response cacheResponse = strategy.cacheResponse;
    if (cache != null) {
        cache.trackResponse(strategy);
    }
    if (cacheCandidate != null && cacheResponse == null) {
        closeQuietly(cacheCandidate.body()); // The cache candidate wasn't applicable. Close it.
    }
    //没有网络请求也没有缓存
    //If we're forbidden from using the network and the cache is insufficient, fail.
    if (networkRequest == null && cacheResponse == null) {
        return new Response.Builder()
                .request(chain.request())
                .protocol(Protocol.HTTP__1)
                .code()
                .message("Unsatisfiable Request (only-if-cached)")
                .body(Util.EMPTY_RESPONSE)
                .sentRequestAtMillis(-L)
                .receivedResponseAtMillis(System.currentTimeMillis())
                .build();
    }
    //没有请求，肯定就要使用缓存
    //If we don't need the network, we're done.
    if (networkRequest == null) {
        return cacheResponse.newBuilder()
                .cacheResponse(stripBody(cacheResponse))
                .build();
    }
    //去发起请求
    Response networkResponse = null;
    try {
        networkResponse = chain.proceed(networkRequest);
    } finally {
        // If we're crashing on I/O or otherwise, don't leak the cache body.
        if (networkResponse == null && cacheCandidate != null) {
            closeQuietly(cacheCandidate.body());
        }
    }
    // If we have a cache response too, then we're doing a conditional get.
    if (cacheResponse != null) {
        //服务器返回无修改，那就使用缓存的响应修改了时间等数据后作为本次请求的响应
        if (networkResponse.code() == HTTP_NOT_MODIFIED) {
            Response response = cacheResponse.newBuilder()
                    .headers(combine(cacheResponse.headers(), networkResponse.headers()))
                    .sentRequestAtMillis(networkResponse.sentRequestAtMillis())
                    .receivedResponseAtMillis(networkResponse.receivedResponseAtMillis())
                    .cacheResponse(stripBody(cacheResponse))
                    .networkResponse(stripBody(networkResponse))
                    .build();
            networkResponse.body().close();
            // Update the cache after combining headers but before stripping the
            // Content-Encoding header (as performed by initContentStream()).
            cache.trackConditionalCacheHit();
            cache.update(cacheResponse, response);
            return response;
        } else {
            closeQuietly(cacheResponse.body());
        }
    }
    //走到这里说明缓存不可用 那就使用网络的响应
    Response response = networkResponse.newBuilder()
            .cacheResponse(stripBody(cacheResponse))
            .networkResponse(stripBody(networkResponse))
            .build();
    //进行缓存
    if (cache != null) {
        if (HttpHeaders.hasBody(response) && CacheStrategy.isCacheable(response,
                networkRequest)) {
            // Offer this request to the cache.
            CacheRequest cacheRequest = cache.put(response);
            return cacheWritingResponse(cacheRequest, response);
        }
        if (HttpMethod.invalidatesCache(networkRequest.method())) {
            try {
                cache.remove(networkRequest);
            } catch (IOException ignored) {
                // The cache cannot be written.
            }
        }
    }
    return response;
}

CacheInterceptor ，在发出请求前，判断是否命中缓存。如果命中则可以不请求，直接使用缓存的响应。 (只会存

在Get请求的缓存)

缓存是否存在，整个方法中的第一个判断是缓存是不是存在，cacheResponse 是从缓存中找到的响应，如果为null，那就表示没有找到对应的缓存，创建的 CacheStrategy 实例对象只存在 networkRequest ，这代表了需要发起网络请求。

https请求的缓存，继续往下走意味着 cacheResponse 必定存在，但是它不一定能用，如果本次请求是HTTPS，但是缓存中没有对应的握手信息，那么缓存无效

响应码以及响应头，整个逻辑都在 isCacheable 中，缓存响应中的响应码为 200, 203, 204, 300, 301, 404, 405, 410, 414, 501, 308 的情况下,只判断服务器是不是给了Cache-Control: no-store (资源不能被缓存)，所以如果服务器给到了这个响应头，那就和前面两个判定一致(缓存不可用)。否则继续进一步判断缓存是否可用

响应码不为 200, 203, 204, 300, 301, 404, 405, 410, 414, 501, 308，302，307 缓存不可用
当响应码为302或者307时，未包含某些响应头，则缓存不可用
当存在 Cache-Control: no-store 响应头则缓存不可用

用户的请求配置，OkHttp需要先对用户本次发起的 Request 进行判定，如果用户指定了 Cache-Control: no-cache (不使用缓存)的请求头或者请求头包含 If-Modified-Since 或 If-None-Match (请求验证)，那么就不允许使用缓存

资源是否不变，如果缓存的响应中包含 Cache-Control: immutable ，这意味着对应请求的响应内容将一直不会改变。此时就可以直接使用缓存。否则继续判断缓存是否可用

3.4.连接拦截器ConnectInterceptor

ConnectInterceptor的intercept方法

 
public Response intercept(Chain chain) throws IOException {
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Request request = realChain.request();
    StreamAllocation streamAllocation = realChain.streamAllocation();
    // We need the network to satisfy this request. Possibly for validating a conditional GET.
    boolean doExtensiveHealthChecks = !request.method().equals("GET");
    HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
    RealConnection connection = streamAllocation.connection();
    return realChain.proceed(request, streamAllocation, httpCodec, connection);
}

虽然代码量很少，实际上大部分功能都封装到其它类去了，这里只是调用而已。

首先我们看到的 StreamAllocation 这个对象是在第一个拦截器：重定向拦截器创建的，但是真正使用的地方却在

" 当一个请求发出，需要建立连接，连接建立后需要使用流用来读写数据 "；

而这个StreamAllocation就是协调请求、连接与数据流三者之间的关系，它负责为一次请求寻找连接，然后获得流来实现网络通信。

这里使用的 newStream 方法实际上就是去查找或者建立一个与请求主机有效的连接，返回的 HttpCodec 中包含了输入输出流，并且封装了对HTTP请求报文的编码与解码，直接使用它就能够与请求主机完成HTTP通信。

StreamAllocation 中简单来说就是维护连接： RealConnection ——封装了Socket与一个Socket连接池。可复用

的 RealConnection 需要：

3.5.请求服务器拦截器CallServerInterceptor

CallServerInterceptor的intercept方法

 
public Response intercept(Chain chain) throws IOException {
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    HttpCodec httpCodec = realChain.httpStream();
    StreamAllocation streamAllocation = realChain.streamAllocation();
    RealConnection connection = (RealConnection) realChain.connection();
    Request request = realChain.request();
    long sentRequestMillis = System.currentTimeMillis();
    realChain.eventListener().requestHeadersStart(realChain.call());
    httpCodec.writeRequestHeaders(request);
    realChain.eventListener().requestHeadersEnd(realChain.call(), request);
    Response.Builder responseBuilder = null;
    if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
        // If there's a "Expect:-continue" header on the request, wait for a "HTTP/1.1 100
        // Continue" response before transmitting the request body. If we don't get that, return
        // what we did get (such as axx response) without ever transmitting the request body.
         //这个请求头代表了在发送请求体之前需要和服务器确定是否愿意接受客户端发送的请求体
        //但是如果服务器不同意接受请求体，那么我们就需要标记该连接不能再被复用，调用 noNewStreams() 关闭相关的Socket。
        if ("-continue".equalsIgnoreCase(request.header("Expect"))) {
            httpCodec.flushRequest();
            realChain.eventListener().responseHeadersStart(realChain.call());
            responseBuilder = httpCodec.readResponseHeaders(true);
        }
        if (responseBuilder == null) {
            // Write the request body if the "Expect:-continue" expectation was met.
            realChain.eventListener().requestBodyStart(realChain.call());
            long contentLength = request.body().contentLength();
            CountingSink requestBodyOut =
                    new CountingSink(httpCodec.createRequestBody(request, contentLength));
            BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
            request.body().writeTo(bufferedRequestBody);
            bufferedRequestBody.close();
            realChain.eventListener()
                    .requestBodyEnd(realChain.call(), requestBodyOut.successfulCount);
        } else if (!connection.isMultiplexed()) {
            // If the "Expect:-continue" expectation wasn't met, prevent the HTTP/1
          // connection
            // from being reused. Otherwise we're still obligated to transmit the request
          // body to
            // leave the connection in a consistent state.
            streamAllocation.noNewStreams();
        }
    }
    httpCodec.finishRequest();
    if (responseBuilder == null) {
        realChain.eventListener().responseHeadersStart(realChain.call());
        responseBuilder = httpCodec.readResponseHeaders(false);
    }
    Response response = responseBuilder
            .request(request)
            .handshake(streamAllocation.connection().handshake())
            .sentRequestAtMillis(sentRequestMillis)
            .receivedResponseAtMillis(System.currentTimeMillis())
            .build();
    int code = response.code();
    if (code ==) {
        // server sent a-continue even though we did not request one.
        // try again to read the actual response
        responseBuilder = httpCodec.readResponseHeaders(false);
        response = responseBuilder
                .request(request)
                .handshake(streamAllocation.connection().handshake())
                .sentRequestAtMillis(sentRequestMillis)
                .receivedResponseAtMillis(System.currentTimeMillis())
                .build();
        code = response.code();
    }
    realChain.eventListener()
            .responseHeadersEnd(realChain.call(), response);
    if (forWebSocket && code ==) {
        // Connection is upgrading, but we need to ensure interceptors see a non-null
      // response body.
        response = response.newBuilder()
                .body(Util.EMPTY_RESPONSE)
                .build();
    } else {
        response = response.newBuilder()
                .body(httpCodec.openResponseBody(response))
                .build();
    }
    if ("close".equalsIgnoreCase(response.request().header("Connection"))
            || "close".equalsIgnoreCase(response.header("Connection"))) {
        streamAllocation.noNewStreams();
    }
    if ((code == || code == 205) && response.body().contentLength() > 0) {
        throw new ProtocolException(
                "HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
    }
    return response;
}

调用 httpCodec.writeRequestHeaders(request); 将请求头写入到缓存中(直到调用 flushRequest() 才真正发

送给服务器)。

CallServerInterceptor ，利用 HttpCodec 发出请求到服务器并且解析生成 Response,在这个拦截器中就是完成HTTP协议报文的封装与解析。

4.OkHttp总结

整个OkHttp功能的实现就在这五个默认的拦截器中，所以先理解拦截器模式的工作机制是先决条件。

这五个拦截器分别为:

重试拦截器

重试拦截器在交出(交给下一个拦截器)之前，负责判断用户是否取消了请求；在获得了结果之后，会根据响应码判断是否需要重定向，如果满足条件那么就会重启执行所有拦截器

桥接拦截器

桥接拦截器在交出之前，负责将HTTP协议必备的请求头加入其中(如：Host)并添加一些默认的行为(如：GZIP压缩)；在获得了结果后，调用保存cookie接口并解析GZIP数据

缓存拦截器

缓存拦截器顾名思义，交出之前读取并判断是否使用缓存；获得结果后判断是否缓存

连接拦截器

连接拦截器在交出之前，负责找到或者新建一个连接，并获得对应的socket流；在获得结果后不进行额外的处理。

请求服务拦截器

请求服务器拦截器进行真正的与服务器的通信，向服务器发送数据，解析读取的响应数据。

每一个拦截器负责的工作不一样，就好像工厂流水线，最终经过这五道工序，就完成了最终的产品。

	OkHttpClient okHttpClient=new OkHttpClient.Builder().build();
	Request request=new Request.Builder()
	.url("http://www.baidu.com")
	.get().build();
	Call call = okHttpClient.newCall(request);
	try {
	Response execute = call.execute();
	Log.d("lpf","execute"+execute.body());
	} catch (IOException e) {
	e.printStackTrace();
	}

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

	public void enqueue(Callback responseCallback) {
	synchronized (this) { //不能冲突调用
	if (executed) throw new IllegalStateException("Already Executed");
	executed = true;
	}
	captureCallStackTrace();
	eventListener.callStart(this);
	client.dispatcher().enqueue(new AsyncCall(responseCallback)); //OkHttp会提供默认的分发器
	}

	synchronized void enqueue(AsyncCall call) { //分发器分发任务
	if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
	runningAsyncCalls.add(call); //正在执行的请求
	executorService().execute(call); //线程池跑任务
	} else {
	readyAsyncCalls.add(call); //ready 中的任务什么时候执行
	}
	}

	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); //这里的代码肯定执行
	}
	}
	}

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

	public synchronized ExecutorService executorService() {
	if (executorService == null) { //这个线程池的特点是高并发最大吞吐量
	executorService =
	new ThreadPoolExecutor(, //核心线程数量
	Integer.MAX_VALUE,//最大线程数量, //空闲线程闲置时间 TimeUnit.SECONDS, //闲置时间单位
	new SynchronousQueue<Runnable>(),//线程等待队列
	Util.threadFactory("OkHttp Dispatcher",false)//线程工厂
	);
	}
	return executorService;
	}

	Response getResponseWithInterceptorChain() throws IOException {
	// Build a full stack of interceptors.
	List<Interceptor> interceptors = new ArrayList<>();
	interceptors.addAll(client.interceptors()); //自定义拦截器加入到集合
	interceptors.add(retryAndFollowUpInterceptor);
	interceptors.add(new BridgeInterceptor(client.cookieJar()));
	interceptors.add(new CacheInterceptor(client.internalCache()));
	interceptors.add(new ConnectInterceptor(client));
	if (!forWebSocket) {//如果不是webSocket
	interceptors.addAll(client.networkInterceptors());
	}
	interceptors.add(new CallServerInterceptor(forWebSocket));
	Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null,,
	originalRequest, this, eventListener, client.connectTimeoutMillis(),
	client.readTimeoutMillis(), client.writeTimeoutMillis());
	return chain.proceed(originalRequest);
	}

	@Override
	public Response intercept(Chain chain) throws IOException {
	Request request = chain.request();
	RealInterceptorChain realChain = (RealInterceptorChain) chain;
	Call call = realChain.call();
	EventListener eventListener = realChain.eventListener();
	/**
	*管理类，维护了与服务器的连接、数据流与请求三者的关系。真正使用的拦截器为 Connect
	*/
	StreamAllocation streamAllocation = new StreamAllocation(client.connectionPool(),
	createAddress(request.url()), call, eventListener, callStackTrace);
	this.streamAllocation = streamAllocation;
	int followUpCount =;
	Response priorResponse = null;
	while (true) {
	if (canceled) {
	streamAllocation.release();
	throw new IOException("Canceled");
	}
	Response response;
	boolean releaseConnection = true;
	try {
	//请求出现了异常，那么releaseConnection依旧为true。
	response = realChain.proceed(request, streamAllocation, null, null);
	releaseConnection = false;
	} catch (RouteException e) {
	//路由异常，连接未成功，请求还没发出去
	//The attempt to connect via a route failed. The request will not have been sent.
	if (!recover(e.getLastConnectException(), streamAllocation, false, request)) {
	throw e.getLastConnectException();
	}
	releaseConnection = false;
	continue;
	} catch (IOException e) {
	//请求发出去了，但是和服务器通信失败了。(socket流正在读写数据的时候断开连接)
	// HTTP才会抛出ConnectionShutdownException。所以对于HTTP1 requestSendStarted一定是true
	//An attempt to communicate with a server failed. The request may have been sent.
	boolean requestSendStarted = !(e instanceof ConnectionShutdownException);
	if (!recover(e, streamAllocation, requestSendStarted, request)) throw e;
	releaseConnection = false;
	continue;
	} finally {
	// We're throwing an unchecked exception. Release any resources.
	//不是前两种的失败，那直接关闭清理所有资源
	if (releaseConnection) {
	streamAllocation.streamFailed(null);
	streamAllocation.release();
	}
	}
	//如果进过重试/重定向才成功的，则在本次响应中记录上次响应的情况
	//Attach the prior response if it exists. Such responses never have a body.
	if (priorResponse != null) {
	response = response.newBuilder()
	.priorResponse(
	priorResponse.newBuilder()
	.body(null)
	.build()
	)
	.build();
	}
	//处理和4xx的一些状态码，如301 302重定向
	Request followUp = followUpRequest(response, streamAllocation.route());
	if (followUp == null) {
	if (!forWebSocket) {
	streamAllocation.release();
	}
	return response;
	}
	closeQuietly(response.body());
	//限制最大 followup 次数为次
	if (++followUpCount > MAX_FOLLOW_UPS) {
	streamAllocation.release();
	throw new ProtocolException("Too many follow-up requests: " + followUpCount);
	}
	if (followUp.body() instanceof UnrepeatableRequestBody) {
	streamAllocation.release();
	throw new HttpRetryException("Cannot retry streamed HTTP body", response.code());
	}
	//todo 判断是不是可以复用同一份连接
	if (!sameConnection(response, followUp.url())) {
	streamAllocation.release();
	streamAllocation = new StreamAllocation(client.connectionPool(),
	createAddress(followUp.url()), call, eventListener, callStackTrace);
	this.streamAllocation = streamAllocation;
	} else if (streamAllocation.codec() != null) {
	throw new IllegalStateException("Closing the body of " + response
	+ " didn't close its backing stream. Bad interceptor?");
	}
	request = followUp;
	priorResponse = response;
	}
	}

	public Response intercept(Chain chain) throws IOException {
	Request userRequest = chain.request();
	Request.Builder requestBuilder = userRequest.newBuilder();
	RequestBody body = userRequest.body();
	if (body != null) {
	MediaType contentType = body.contentType();
	if (contentType != null) {
	requestBuilder.header("Content-Type", contentType.toString());
	}
	long contentLength = body.contentLength();
	if (contentLength != -) {
	requestBuilder.header("Content-Length", Long.toString(contentLength));
	requestBuilder.removeHeader("Transfer-Encoding");
	} else {
	requestBuilder.header("Transfer-Encoding", "chunked");
	requestBuilder.removeHeader("Content-Length");
	}
	}
	if (userRequest.header("Host") == null) {
	requestBuilder.header("Host", hostHeader(userRequest.url(), false));
	}
	if (userRequest.header("Connection") == null) {
	requestBuilder.header("Connection", "Keep-Alive");
	}
	// If we add an "Accept-Encoding: gzip" header field we're responsible for also
	// decompressing
	// the transfer stream.
	boolean transparentGzip = false;
	if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
	transparentGzip = true;
	requestBuilder.header("Accept-Encoding", "gzip");
	}
	List<Cookie> cookies = cookieJar.loadForRequest(userRequest.url());
	if (!cookies.isEmpty()) {
	requestBuilder.header("Cookie", cookieHeader(cookies));
	}
	if (userRequest.header("User-Agent") == null) {
	requestBuilder.header("User-Agent", Version.userAgent());
	}
	Response networkResponse = chain.proceed(requestBuilder.build());
	HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());
	Response.Builder responseBuilder = networkResponse.newBuilder()
	.request(userRequest);
	if (transparentGzip
	&& "gzip".equalsIgnoreCase(networkResponse.header("Content-Encoding"))
	&& HttpHeaders.hasBody(networkResponse)) {
	GzipSource responseBody = new GzipSource(networkResponse.body().source());
	Headers strippedHeaders = networkResponse.headers().newBuilder()
	.removeAll("Content-Encoding")
	.removeAll("Content-Length")
	.build();
	responseBuilder.headers(strippedHeaders);
	String contentType = networkResponse.header("Content-Type");
	responseBuilder.body(new RealResponseBody(contentType, -L, Okio.buffer(responseBody)));
	}
	return responseBuilder.build();
	}

	public Response intercept(Chain chain) throws IOException {
	//通过url的md数据从文件缓存查找（GET请求才有缓存）
	Response cacheCandidate = cache != null
	? cache.get(chain.request())
	: null;
	long now = System.currentTimeMillis();
	//缓存策略:根据各种条件(请求头)组成请求与缓存
	CacheStrategy strategy =
	new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();
	//
	Request networkRequest = strategy.networkRequest;
	Response cacheResponse = strategy.cacheResponse;
	if (cache != null) {
	cache.trackResponse(strategy);
	}
	if (cacheCandidate != null && cacheResponse == null) {
	closeQuietly(cacheCandidate.body()); // The cache candidate wasn't applicable. Close it.
	}
	//没有网络请求也没有缓存
	//If we're forbidden from using the network and the cache is insufficient, fail.
	if (networkRequest == null && cacheResponse == null) {
	return new Response.Builder()
	.request(chain.request())
	.protocol(Protocol.HTTP__1)
	.code()
	.message("Unsatisfiable Request (only-if-cached)")
	.body(Util.EMPTY_RESPONSE)
	.sentRequestAtMillis(-L)
	.receivedResponseAtMillis(System.currentTimeMillis())
	.build();
	}
	//没有请求，肯定就要使用缓存
	//If we don't need the network, we're done.
	if (networkRequest == null) {
	return cacheResponse.newBuilder()
	.cacheResponse(stripBody(cacheResponse))
	.build();
	}
	//去发起请求
	Response networkResponse = null;
	try {
	networkResponse = chain.proceed(networkRequest);
	} finally {
	// If we're crashing on I/O or otherwise, don't leak the cache body.
	if (networkResponse == null && cacheCandidate != null) {
	closeQuietly(cacheCandidate.body());
	}
	}
	// If we have a cache response too, then we're doing a conditional get.
	if (cacheResponse != null) {
	//服务器返回无修改，那就使用缓存的响应修改了时间等数据后作为本次请求的响应
	if (networkResponse.code() == HTTP_NOT_MODIFIED) {
	Response response = cacheResponse.newBuilder()
	.headers(combine(cacheResponse.headers(), networkResponse.headers()))
	.sentRequestAtMillis(networkResponse.sentRequestAtMillis())
	.receivedResponseAtMillis(networkResponse.receivedResponseAtMillis())
	.cacheResponse(stripBody(cacheResponse))
	.networkResponse(stripBody(networkResponse))
	.build();
	networkResponse.body().close();
	// Update the cache after combining headers but before stripping the
	// Content-Encoding header (as performed by initContentStream()).
	cache.trackConditionalCacheHit();
	cache.update(cacheResponse, response);
	return response;
	} else {
	closeQuietly(cacheResponse.body());
	}
	}
	//走到这里说明缓存不可用那就使用网络的响应
	Response response = networkResponse.newBuilder()
	.cacheResponse(stripBody(cacheResponse))
	.networkResponse(stripBody(networkResponse))
	.build();
	//进行缓存
	if (cache != null) {
	if (HttpHeaders.hasBody(response) && CacheStrategy.isCacheable(response,
	networkRequest)) {
	// Offer this request to the cache.
	CacheRequest cacheRequest = cache.put(response);
	return cacheWritingResponse(cacheRequest, response);
	}
	if (HttpMethod.invalidatesCache(networkRequest.method())) {
	try {
	cache.remove(networkRequest);
	} catch (IOException ignored) {
	// The cache cannot be written.
	}
	}
	}
	return response;
	}

	public Response intercept(Chain chain) throws IOException {
	RealInterceptorChain realChain = (RealInterceptorChain) chain;
	Request request = realChain.request();
	StreamAllocation streamAllocation = realChain.streamAllocation();
	// We need the network to satisfy this request. Possibly for validating a conditional GET.
	boolean doExtensiveHealthChecks = !request.method().equals("GET");
	HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
	RealConnection connection = streamAllocation.connection();
	return realChain.proceed(request, streamAllocation, httpCodec, connection);
	}

	public Response intercept(Chain chain) throws IOException {
	RealInterceptorChain realChain = (RealInterceptorChain) chain;
	HttpCodec httpCodec = realChain.httpStream();
	StreamAllocation streamAllocation = realChain.streamAllocation();
	RealConnection connection = (RealConnection) realChain.connection();
	Request request = realChain.request();
	long sentRequestMillis = System.currentTimeMillis();
	realChain.eventListener().requestHeadersStart(realChain.call());
	httpCodec.writeRequestHeaders(request);
	realChain.eventListener().requestHeadersEnd(realChain.call(), request);
	Response.Builder responseBuilder = null;
	if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
	// If there's a "Expect:-continue" header on the request, wait for a "HTTP/1.1 100
	// Continue" response before transmitting the request body. If we don't get that, return
	// what we did get (such as axx response) without ever transmitting the request body.
	//这个请求头代表了在发送请求体之前需要和服务器确定是否愿意接受客户端发送的请求体
	//但是如果服务器不同意接受请求体，那么我们就需要标记该连接不能再被复用，调用 noNewStreams() 关闭相关的Socket。
	if ("-continue".equalsIgnoreCase(request.header("Expect"))) {
	httpCodec.flushRequest();
	realChain.eventListener().responseHeadersStart(realChain.call());
	responseBuilder = httpCodec.readResponseHeaders(true);
	}
	if (responseBuilder == null) {
	// Write the request body if the "Expect:-continue" expectation was met.
	realChain.eventListener().requestBodyStart(realChain.call());
	long contentLength = request.body().contentLength();
	CountingSink requestBodyOut =
	new CountingSink(httpCodec.createRequestBody(request, contentLength));
	BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
	request.body().writeTo(bufferedRequestBody);
	bufferedRequestBody.close();
	realChain.eventListener()
	.requestBodyEnd(realChain.call(), requestBodyOut.successfulCount);
	} else if (!connection.isMultiplexed()) {
	// If the "Expect:-continue" expectation wasn't met, prevent the HTTP/1
	// connection
	// from being reused. Otherwise we're still obligated to transmit the request
	// body to
	// leave the connection in a consistent state.
	streamAllocation.noNewStreams();
	}
	}
	httpCodec.finishRequest();
	if (responseBuilder == null) {
	realChain.eventListener().responseHeadersStart(realChain.call());
	responseBuilder = httpCodec.readResponseHeaders(false);
	}
	Response response = responseBuilder
	.request(request)
	.handshake(streamAllocation.connection().handshake())
	.sentRequestAtMillis(sentRequestMillis)
	.receivedResponseAtMillis(System.currentTimeMillis())
	.build();
	int code = response.code();
	if (code ==) {
	// server sent a-continue even though we did not request one.
	// try again to read the actual response
	responseBuilder = httpCodec.readResponseHeaders(false);
	response = responseBuilder
	.request(request)
	.handshake(streamAllocation.connection().handshake())
	.sentRequestAtMillis(sentRequestMillis)
	.receivedResponseAtMillis(System.currentTimeMillis())
	.build();
	code = response.code();
	}
	realChain.eventListener()
	.responseHeadersEnd(realChain.call(), response);
	if (forWebSocket && code ==) {
	// Connection is upgrading, but we need to ensure interceptors see a non-null
	// response body.
	response = response.newBuilder()
	.body(Util.EMPTY_RESPONSE)
	.build();
	} else {
	response = response.newBuilder()
	.body(httpCodec.openResponseBody(response))
	.build();
	}
	if ("close".equalsIgnoreCase(response.request().header("Connection"))
	\|\| "close".equalsIgnoreCase(response.header("Connection"))) {
	streamAllocation.noNewStreams();
	}
	if ((code == \|\| code == 205) && response.body().contentLength() > 0) {
	throw new ProtocolException(
	"HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
	}
	return response;
	}