Android okhttp源码分析

小咪咪 2023-07-05 14:41 11阅读 0赞

**1.OkHttp同步的使用**

//如果创建创建需要很多的参数，可以使用Builder()方式
     OkHttpClient mOkHttpClient = new OkHttpClient.Builder().readTimeout(5, TimeUnit.SECONDS).build();
      
        public void synRequest(){ 
            Request  request=new Request.Builder().url("http://www.baidu.com").get().build();
            Call call=mOkHttpClient.newCall(request);
            try{ 
                Response response=call.execute();
                System.out.println(response.body().toString());
            }catch(Exception e){ 
                e.printStackTrace();
            }
        }

1看下OkHttpClient的Builder()方法，对象属性的初始化

public Builder() { 
          dispatcher = new Dispatcher(); //请求分发器，由它决定异步请求时直接处理还是缓存等待，对于同步请求没有做太多的操作
          protocols = DEFAULT_PROTOCOLS;
          connectionSpecs = DEFAULT_CONNECTION_SPECS;
          eventListenerFactory = EventListener.factory(EventListener.NONE);
          proxySelector = ProxySelector.getDefault();
          if (proxySelector == null) { 
            proxySelector = new NullProxySelector();
          }
          cookieJar = CookieJar.NO_COOKIES;
          socketFactory = SocketFactory.getDefault();
          hostnameVerifier = OkHostnameVerifier.INSTANCE;
          certificatePinner = CertificatePinner.DEFAULT;
          proxyAuthenticator = Authenticator.NONE;
          authenticator = Authenticator.NONE;
          connectionPool = new ConnectionPool();//它是一个连接池，客户端和服务器之间的连接可以抽象为一个connection，每一个connection都会放在连接池中，由它来进行统一的管理，要是相同的话，可以选择复用，它可以决哪些连接是可以保持打开，哪些是可以复用的

2.看下Request请求报文的Builder()方法

public Builder() { 
          this.method = "GET";
          this.headers = new Headers.Builder();
        }

3.看下Request的build()方法

public Request build() { 
          if (url == null) throw new IllegalStateException("url == null");
          return new Request(this);
        }

4.根据3看下request的构造方法，方法里是有关请求信息的属性

Request(Builder builder) { 
        this.url = builder.url;
        this.method = builder.method;
        this.headers = builder.headers.build();
        this.body = builder.body;
        this.tags = Util.immutableMap(builder.tags);
      }

5.看下newCall(request)方法一直跟踪到call的实现类RealCall构造方法，可以看到RealCall持有了前两步创建的OkHttpClient和Request

private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) { 
        this.client = client;
        this.originalRequest = originalRequest;
        this.forWebSocket = forWebSocket;
      }

6.接下来是call.execute()

@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);//主动回收同步请求
        }
      }

7.看下client.dispatcher()就返回dispatcher实例  
public Dispatcher dispatcher() \{  
return dispatcher;  
\}  
8.看下 client.dispatcher().executed(this); 就是在分发器里同步请求队列了添加请求

/** Used by {@code Call#execute} to signal it is in-flight. */
      synchronized void executed(RealCall call) { 
        runningSyncCalls.add(call);
      }
        /** Running synchronous calls. Includes canceled calls that haven't finished yet. */
      private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

9.看下finished()方法

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

**注意 发送同步请求后，就会进入阻塞状态，直到收到响应**

***3.OkHttp异步的使用***\*

OkHttpClient mOkHttpClient = new OkHttpClient.Builder().readTimeout(5, TimeUnit.SECONDS).build();
    
    public void synRequest(){ 
        Request  request=new Request.Builder().url("http://www.baidu.com").get().build();
        Call call=mOkHttpClient.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().toString());
            }
        });
    }
    1.看下enqueue()方法
      @Override public void enqueue(Callback responseCallback) { 
        synchronized (this) { 
          if (executed) throw new IllegalStateException("Already Executed");
          executed = true;//call只能被执行一次
        }
        transmitter.callStart();
        client.dispatcher().enqueue(new AsyncCall(responseCallback));
      }
     
    2.看下client.dispatcher().enqueue(new AsyncCall(responseCallback));的enqueue()方法
    
      void enqueue(AsyncCall call) { 
        synchronized (this) { 
          readyAsyncCalls.add(call);//缓存等待的队列添加异步请求
    
          // 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();
      }
      /** 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<>();  //异步队列

3.看下promoteAndExecute() 方法

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; // 异步请求队列里请求数量要小于64.
            if (asyncCall.callsPerHost().get() >= maxRequestsPerHost) continue; // Host max capacity.
    
            i.remove();//遍历缓存队列，移除，添加到正在运行的队列
            asyncCall.callsPerHost().incrementAndGet();
            executableCalls.add(asyncCall);
            runningAsyncCalls.add(asyncCall);//小于64就添加到异步请求对ie
          }
          isRunning = runningCallsCount() > 0;
        }
    for (int i = 0, size = executableCalls.size(); i < size; i++) { 
          AsyncCall asyncCall = executableCalls.get(i);
          asyncCall.executeOn(executorService());//执行
        }
    
        return isRunning;
      }
    
     private int maxRequests = 64;
      private int maxRequestsPerHost = 5;
    4.看下runningCallsCount()方法
    public synchronized int runningCallsCount() { 
        return runningAsyncCalls.size() + runningSyncCalls.size();//同步请求队列里请求的数量和异步请求队列里请求的数量
      }

5.看下 asyncCall.executeOn(executorService());里的executorService

public synchronized ExecutorService executorService() {  //同步
        if (executorService == null) {  //保证单例
          executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,  //线程池 Integer.MAX_VALUE前面有maxRequests=64限制 参数的意义,60s后，将无关的线程关闭
              new SynchronousQueue<>(), Util.threadFactory("OkHttp Dispatcher", false));
        }
        return executorService;
      }

6.看下AsyncCall，AsyncCall就是Runnable

final class AsyncCall extends NamedRunnable { 
      public abstract class NamedRunnable implements Runnable {

7.看下asyncCall.executeOn()方法

@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);//1.从正在执行的异步队列中，移除当前请求 2.重新计算正在运行请求的数量
          }
        }
      }

**一个异步请求执行完后，要将此请求从异步执行队列中移除**  
**注意onFailure，onResponse方法都是在子线程中进行的**

**dispatcher**  
dispatcher的作用为维护请求的状态，并维护一个线程池，用于执行请求  
**dispatcher里为什么要维护两个异步请求队列**  
dispatcher运行在主线程的生产者  
ExecutorService消费者池  
Deque readyAsyncCalls缓存异步队列  
Deque readyAsyncCalls正在执行的异步队列

**拦截器是okhttp中提供一种强大机制，它可以实现网监听，请求以及响应重写，请求失败重功能（不区分同步异步）**

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);
    1.看下 chain.proceed(originalRequest)方法
    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); //从index + 1开始访问，也就是下一个拦截器开始访问，构成拦截器链
        Interceptor interceptor = interceptors.get(index);
        Response response = interceptor.intercept(next);

1.RetryAndFollowUpInterceptor拦截器 此拦截器根据需要执行重定向  
看RetryAndFollowUpInterceptor的intercept方法

@Override public Response intercept(Chain chain) throws IOException { 
                                       .
                                       .
                                       .
                                         if (++followUpCount > MAX_FOLLOW_UPS) {   
            throw new ProtocolException("Too many follow-up requests: " + followUpCount);
          }
    
          request = followUp;
          priorResponse = response;
        }
         private static final int MAX_FOLLOW_UPS = 20; //重连次数20次

2.BridgeInterceptor拦截器  
从应用程序代码到网络代码的桥梁。首先，它构建来自用户的网络请求  
\*请求。然后它继续呼叫网络。最后，它从网络构建用户响应回应。

*  看BridgeInterceptor的intercept()方法

*  `@Override public Response intercept(Chain chain) throws IOException { `
                                                      .
                                                      .
                                                      .
           if (userRequest.header("Connection") == null) { 
          requestBuilder.header("Connection", "Keep-Alive");//保持连接
                                                       .
                                                       .
                                                       .
          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")) //支持gzip
            && 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, -1L, Okio.buffer(responseBody)));
        }
        }
    }

3.okhttp的缓存  
使用

OkHttpClient mOkHttpClient = new OkHttpClient.Builder().cache(new Cache(new File("catch"),24*1024*1024)).readTimeout(5, TimeUnit.SECONDS).build();

看下cache的put方法

@Nullable CacheRequest put(Response response) { 
        String requestMethod = response.request().method();
    
        if (HttpMethod.invalidatesCache(response.request().method())) { 
          try { 
            remove(response.request());
          } catch (IOException ignored) { 
            // The cache cannot be written.
          }
          return null;
        }
        if (!requestMethod.equals("GET")) {   //不缓存get方法
          // Don't cache non-GET responses. We're technically allowed to cache
          // HEAD requests and some POST requests, but the complexity of doing
          // so is high and the benefit is low.
          return null;
        }
    
        if (HttpHeaders.hasVaryAll(response)) { 
          return null;
        }
    
        Entry entry = new Entry(response); //它的实例就是要写入缓存的部分（它的源码里都是有关需要缓存的属性，被封装成了Entry）
        DiskLruCache.Editor editor = null; //okhttp的缓存技术就是使用DiskLruCache
        try { 
          editor = cache.edit(key(response.request().url()));//将网络请求的url转换为对应的key（源码里是MD5加密的十六进制的表现形式）
          if (editor == null) { 
            return null;
          }
          entry.writeTo(editor);//开始缓存
          return new CacheRequestImpl(editor);
        } catch (IOException e) { 
          abortQuietly(editor);
          return null;
        }
      }
    看下writeTo()方法
    
    ```java
        public void writeTo(DiskLruCache.Editor editor) throws IOException { 
          BufferedSink sink = Okio.buffer(editor.newSink(ENTRY_METADATA));
                                      .
                                      .
                                      .
    
    ```if (isHttps()) {   //判断了是否是https请求，写入了握手的信息
            sink.writeByte('\n');
            sink.writeUtf8(handshake.cipherSuite().javaName())
                .writeByte('\n');
            writeCertList(sink, handshake.peerCertificates());
            writeCertList(sink, handshake.localCertificates());
            sink.writeUtf8(handshake.tlsVersion().javaName()).writeByte('\n');
          }
          sink.close();
        }

还有主体信息在哪缓存？看下CacheRequestImpl(final DiskLruCache.Editor editor)方法

CacheRequestImpl(final DiskLruCache.Editor editor) {  //这个方法写入body信息
          this.editor = editor;
          this.cacheOut = editor.newSink(ENTRY_BODY);
          this.body = new ForwardingSink(cacheOut) {  
            @Override public void close() throws IOException { 
              synchronized (Cache.this) { 
                if (done) { 
                  return;
                }
                done = true;
                writeSuccessCount++;
              }
              super.close();
              editor.commit();
            }
          };
        }

CacheInterceptor缓存拦截器  
看下intercept方法

@Override public Response intercept(Chain chain) throws IOException { 
        Response cacheCandidate = cache != null
            ? cache.get(chain.request()) //获取缓存
            : null;
    
        long now = System.currentTimeMillis();
    
        CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();//工厂方法，实例化CacheStrategy ，写入请求头的操作 
        /* /** CacheStrategy 缓存策略，看是从networkRequest网络请求还是cacheResponse缓存中获取Response；还可以同时使用，进行对比，选择最好的策略 The request to send on the network, or null if this call doesn't use the network. */
      public final @Nullable Request networkRequest;
    
      /** The cached response to return or validate; or null if this call doesn't use a cache. */
      public final @Nullable Response cacheResponse;
        **/
        Request networkRequest = strategy.networkRequest;
        Response cacheResponse = strategy.cacheResponse;
    
        if (cache != null) { 
          cache.trackResponse(strategy);//更新缓存指标，主要是命中率，在okhttp3.Cache的trackResponse()方法里，hitCount++表示命中了缓存
        }
    
        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) {   //网络不好，没有缓存，构建Response，抛出504错误
          return new Response.Builder()
              .request(chain.request())
              .protocol(Protocol.HTTP_1_1)
              .code(504)
              .message("Unsatisfiable Request (only-if-cached)")
              .body(Util.EMPTY_RESPONSE)
              .sentRequestAtMillis(-1L)
              .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) {  //如果响应码HTTP_NOT_MODIFIED为304，直接从缓存中读取数据
            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())) { //无效的缓存方法，移除request 
            try { 
              cache.remove(networkRequest);
            } catch (IOException ignored) { 
              // The cache cannot be written.
            }
          }
        }
    
        return response;
      }

看下CacheStrategy 的getCandidate()方法

private CacheStrategy getCandidate() { 
                                     .
                                     .
                                     . 
             //写入请求头的操作 
           if (!responseCaching.noCache() && ageMillis + minFreshMillis < freshMillis + maxStaleMillis) { 
            Response.Builder builder = cacheResponse.newBuilder();
            if (ageMillis + minFreshMillis >= freshMillis) { 
              builder.addHeader("Warning", "110 HttpURLConnection \"Response is stale\"");
            }
            long oneDayMillis = 24 * 60 * 60 * 1000L;
            if (ageMillis > oneDayMillis && isFreshnessLifetimeHeuristic()) { 
              builder.addHeader("Warning", "113 HttpURLConnection \"Heuristic expiration\"");
            }
            return new CacheStrategy(null, builder.build());
          }
          String conditionName;
          String conditionValue;
          if (etag != null) { 
            conditionName = "If-None-Match";
            conditionValue = etag;
          } else if (lastModified != null) { 
            conditionName = "If-Modified-Since";
            conditionValue = lastModifiedString;
          } else if (servedDate != null) { 
            conditionName = "If-Modified-Since";
            conditionValue = servedDateString;
          } else { 
            return new CacheStrategy(request, null); // No condition! Make a regular request.
          }
    
          Headers.Builder conditionalRequestHeaders = request.headers().newBuilder();
          Internal.instance.addLenient(conditionalRequestHeaders, conditionName, conditionValue);
    
          Request conditionalRequest = request.newBuilder()
              .headers(conditionalRequestHeaders.build())
              .build();
          return new CacheStrategy(conditionalRequest, cacheResponse);                                                                    
        }

ConnectInterceptor网络连接拦截器  
看下intercept方法

@Override public Response intercept(Chain chain) throws IOException { 
                                       .
                                       .
                                       .
        Exchange exchange = transmitter.newExchange(chain, doExtensiveHealthChecks);
    
        return realChain.proceed(request, transmitter, exchange);
      }

看下newExchange()方法

Exchange newExchange(Interceptor.Chain chain, boolean doExtensiveHealthChecks) { 
                                                                         .
                                                                         .
                                                                         .
     ExchangeCodec codec = exchangeFinder.find(client, chain, doExtensiveHealthChecks);
    }
    ``
    看下 exchangeFinder.find(client, chain, doExtensiveHealthChecks)方法
    
    ```java
     public ExchangeCodec find(
                                                        .
                                                        .
                                                        .
                                                        try { 
          RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,//RealConnection 是核心
              writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
          return resultConnection.newCodec(client, chain);
        } catch (RouteException e) { 
          trackFailure();
          throw e;
        } catch (IOException e) { 
          trackFailure();
          throw new RouteException(e);
        }
      }
    }

看下findHealthyConnection()方法

private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
                                                                        .
                                                                        .
                                                                        .
          synchronized (connectionPool) { 
            if (candidate.successCount == 0) {   //如果successCount =0，就返回，这时候循环已结束
              return candidate;
            }
              if (!candidate.isHealthy(doExtensiveHealthChecks)) {  //检查candidate是否健康（比如Socket连接没有关闭，输入输出流没有关闭，就是不健康的）
            candidate.noNewExchanges(); //如果不健康，进行资源销毁的动作
            continue;//循环调用RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
              pingIntervalMillis, connectionRetryEnabled);去寻找RealConnection 
          }
    
          return candidate;
        }
      }

看下 findConnection(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis, connectionRetryEnabled);方法

private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
          int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException { 
                                                  .
                                                  .
                                                  .
               if (transmitter.connection != null) { 
    
            result = transmitter.connection;  //尝试复用，赋值给result，result就是RealConnection
            releasedConnection = null;
          }
                                                  .
                                                  .
                                                  .
          if (result == null) { 
            // 如果不能复用从connectionPool连接池中获取
            if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, null, false)) { 
              foundPooledConnection = true;
              result = transmitter.connection;
            } else if (nextRouteToTry != null) { 
              selectedRoute = nextRouteToTry;
              nextRouteToTry = null;
            } else if (retryCurrentRoute()) { 
              selectedRoute = transmitter.connection.route();
            }
          }
        }
                                                  .
                                                  .
                                                  .
           result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,//核心，获取到了RealConnection，就进行网络连接
          connectionRetryEnabled, call, eventListener);
        connectionPool.routeDatabase.connected(result.route());
    
        Socket socket = null;
      synchronized (connectionPool) { 
          connectingConnection = null;
          // Last attempt at connection coalescing, which only occurs if we attempted multiple
          // concurrent connections to the same host.
          if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, routes, true)) { 
            // We lost the race! Close the connection we created and return the pooled connection.
            result.noNewExchanges = true;
            socket = result.socket();
            result = transmitter.connection;
          } else { 
            connectionPool.put(result);//获取到了RealConnection，添加到连接池中
            transmitter.acquireConnectionNoEvents(result);
          }
        }
    }

看下esult.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,connectionRetryEnabled, call, eventListener);方法

public void connect(int connectTimeout, int readTimeout, int writeTimeout,
          int pingIntervalMillis, boolean connectionRetryEnabled, Call call,
          EventListener eventListener) { 
        if (protocol != null) throw new IllegalStateException("already connected");//核心判断有没有连接
            RouteException routeException = null;
        List<ConnectionSpec> connectionSpecs = route.address().connectionSpecs();//指定Socket连接的配置
        ConnectionSpecSelector connectionSpecSelector = new ConnectionSpecSelector(connectionSpecs);//用于选择链接（隧道链接，Socket链接）
                                                  .
                                                  .
                                                  .
         }

RealConnectionPool连接池 作用：在它的时间范围内复用connection  
看下 transmitterAcquirePooledConnection(Address address, Transmitter transmitter,@Nullable List routes, boolean requireMultiplexed) 获取链接的方法

boolean transmitterAcquirePooledConnection(Address address, Transmitter transmitter,
          @Nullable List<Route> routes, boolean requireMultiplexed) { 
        assert (Thread.holdsLock(this));
        for (RealConnection connection : connections) { 
          if (requireMultiplexed && !connection.isMultiplexed()) continue;
          if (!connection.isEligible(address, routes)) continue;//判断链接是否可以使用
          transmitter.acquireConnectionNoEvents(connection);
          return true;
        }
        return false;
      }

看下transmitter.acquireConnectionNoEvents(connection)方法

void acquireConnectionNoEvents(RealConnection connection) { 
        assert (Thread.holdsLock(connectionPool));
    
        if (this.connection != null) throw new IllegalStateException();
        this.connection = connection; //赋值
        connection.transmitters.add(new TransmitterReference(this, callStackTrace));//添加到transmitters，transmitters就是一个集合，通过集合的大小判断是否超过负载量
      }

看下put(RealConnection connection) 方法

void put(RealConnection connection) { 
        assert (Thread.holdsLock(this));
        if (!cleanupRunning) { 
          cleanupRunning = true;
          executor.execute(cleanupRunnable);//执行
        }
        connections.add(connection);//添加到connections集合中
      }

看下清理线程池的代码

private final Runnable cleanupRunnable = () -> { 
        while (true) { 
          long waitNanos = cleanup(System.nanoTime());//下一次需要清理间隔的时间
          if (waitNanos == -1) return;
          if (waitNanos > 0) { 
            long waitMillis = waitNanos / 1000000L;
            waitNanos -= (waitMillis * 1000000L);
            synchronized (RealConnectionPool.this) { 
              try { 
                RealConnectionPool.this.wait(waitMillis, (int) waitNanos);//等待释放锁和时间片，当等待的时间过了之后，它会再次调用Runnable去清理，同时会返回下次需要清理的间隔时间
              } catch (InterruptedException ignored) { 
              }
            }
          }
        }
      };

看下 cleanup(System.nanoTime())方法 相当于虚拟机的标记清除算法，就是标记不活跃或者长时间为使用的链接，然后进行清除

long cleanup(long now) { 
        int inUseConnectionCount = 0;
        int idleConnectionCount = 0;
        RealConnection longestIdleConnection = null;
        long longestIdleDurationNs = Long.MIN_VALUE;
    
        // Find either a connection to evict, or the time that the next eviction is due.
        synchronized (this) { 
          for (Iterator<RealConnection> i = connections.iterator(); i.hasNext(); ) { 
            RealConnection connection = i.next();
    
            // If the connection is in use, keep searching.
            if (pruneAndGetAllocationCount(connection, now) > 0) { //找到正在使用的链接的方法
              inUseConnectionCount++;
              continue;
            }
    
            idleConnectionCount++;
    
            // If the connection is ready to be evicted, we're done.
            long idleDurationNs = now - connection.idleAtNanos;
            if (idleDurationNs > longestIdleDurationNs) { 
              longestIdleDurationNs = idleDurationNs;
              longestIdleConnection = connection;
            }
          }
    
          if (longestIdleDurationNs >= this.keepAliveDurationNs
              || idleConnectionCount > this.maxIdleConnections) { 
            // We've found a connection to evict. Remove it from the list, then close it below (outside
            // of the synchronized block).
            connections.remove(longestIdleConnection);//空闲链接数量超过数量，就进行清除
          } else if (idleConnectionCount > 0) { 
            // A connection will be ready to evict soon.
            return keepAliveDurationNs - longestIdleDurationNs;
          } else if (inUseConnectionCount > 0) { 
            // All connections are in use. It'll be at least the keep alive duration 'til we run again.
            return keepAliveDurationNs;
          } else { 
            // No connections, idle or in use.
            cleanupRunning = false;
            return -1;//没有需要清理的链接，跳出循环
          }
        }
    
        closeQuietly(longestIdleConnection.socket());
    
        // Cleanup again immediately.
        return 0;
      }

CallServerInterceptor拦截器，主要是向服务器发起一个真正的网络请求，接收到服务器给我们的返回响应  
看下@Override public Response intercept(Chain chain) throws IOException 方法

@Override public Response intercept(Chain chain) throws IOException { 
        RealInterceptorChain realChain = (RealInterceptorChain) chain;
        Exchange exchange = realChain.exchange(); //编码request，解码response
        Request request = realChain.request();
    
        long sentRequestMillis = System.currentTimeMillis();
    
        exchange.writeRequestHeaders(request);//向Socket里写入请求的头部信息
    
        boolean responseHeadersStarted = false;
        Response.Builder responseBuilder = null;
    
      if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {  //特殊处理，头部是否可以包含Expect: 100-continue
            exchange.flushRequest();
            responseHeadersStarted = true;
            exchange.responseHeadersStart();
            responseBuilder = exchange.readResponseHeaders(true);
          }
           if (responseBuilder == null) { 
            if (request.body().isDuplex()) { 
              // Prepare a duplex body so that the application can send a request body later.
              exchange.flushRequest();
              BufferedSink bufferedRequestBody = Okio.buffer(
                  exchange.createRequestBody(request, true));
              request.body().writeTo(bufferedRequestBody);//向Socket里写入body信息
                                                   .
                                                   .
                                                   .
                 if (request.body() == null || !request.body().isDuplex()) { 
          exchange.finishRequest();//完成整个网络请求的写入信息
        }
       
        if (responseBuilder == null) { 
                                                   .
                                                   .
                                                   .
          responseBuilder = exchange.readResponseHeaders(false);//读取网络请求的头部信息
                                                   .
                                                   .
                                                   .
    if (forWebSocket && code == 101) {  //判断响应码
          // 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()  //构建response
              .body(exchange.openResponseBody(response))
              .build();
        }
            if ("close".equalsIgnoreCase(response.request().header("Connection"))
            || "close".equalsIgnoreCase(response.header("Connection"))) { 
          exchange.noNewExchangesOnConnection();//关闭流，关闭链接
    
    
        }