OkHttp源码解析06-宏观

ゝ一世哀愁。 2023-10-10 10:28 38阅读 0赞

#### OKHTTP的使用 ####

OkHttpClient okHttpClient = new OkHttpClient.Builder().build();
    
    Request request = new Request.Builder().url("https://www.baidu.com").get().build(); //用到了建造者模式
    
    Call call = okHttpClient.newCall(request);
    
    call.enqueue(new Callback() {
        
        @Override
        public void onFailure(Call call, IOException e) {
        
            System.out.println("请求失败.. E:" + e.toString());
        }
    
        @Override
        public void onResponse(Call call, Response response) throws IOException {
        
            String string = response.body().string();
            System.out.println("请求完成.. result:" + string);
    		//response.body().byteStream();
    		//response.body().charStream();
        }
    });

#### 主流程源码分析: ####

@Override public void enqueue(Callback responseCallback) {
        
    	synchronized (this) {
        
    	  //不能执行大于一次enqueue
    	  if (executed) throw new IllegalStateException("Already Executed");
    	  executed = true;
    	}
    	captureCallStackTrace();
    	eventListener.callStart(this);
    	//拿到调度器dispatcher.enqueue方法
    	client.dispatcher().enqueue(new AsyncCall(responseCallback));
    }

接着来看`client.dispatcher.enqueue`

synchronized void enqueue(AsyncCall call) {
        
    	/*同时运行的异步任务小于64 && 同时访问同一个服务器 不能超过5个*/
    	if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
        
    		//加入到执行队列
    	  	runningAsyncCalls.add(call);
    	  	executorService().execute(call);
    	} else {
        
    		//加入到等待队列
    	  	readyAsyncCalls.add(call);
    	}
     }

这里有两个队列，会根据如上的判断，添加到指定队列里去

/** 等待执行的队列 */
      private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
    
      /** 异步运行的队列 enqueue */
      private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
    
      /** 同步运行的队列 execute */
      private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

> Deque: 双端队列  
> AsyncCall: 执行耗时任务

来看下AsyncCall，这里getResponseWithInterceptorChain()是责任链模式，最终会请求到服务器

final class AsyncCall extends NamedRunnable {
        
        private final Callback responseCallback;
    	//...
        @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) {
        
              // signalledCallback代表已经回调过CallBack，说明这个异常是用户造成的，不处理
              // 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);
          }
        }
      }

#### 分析OKHTTP的线程池 ####

在`executorService().execute(call);`这行代码，就是调用线程池去执行。

先来看`executorService()`，位于Dispatcher类

public synchronized ExecutorService executorService() {
        
      if (executorService == null) {
        
         executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
             new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
       }
       return executorService;
     }

可以看到，这里创建了一个核心线程数为0，最大线程数为MAX\_VAUE，缓存时间为60秒的一个线程池，

> 这里的配置和Executors.newCachedThreadPool()一致; //创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。这种线程池适合执行大量耗时较少的任务。

结论:Okhttp里面的线程池，采用的是缓存方案

#### OkHttp责任链模式 ####

`getResponseWithInterceptorChain()`用到了责任链模式，来看下相关代码，在RealCall

Response getResponseWithInterceptorChain() throws IOException {
        
        // Build a full stack of interceptors.
        List<Interceptor> interceptors = new ArrayList<>();
        //用户自定义的拦截器，用在与服务器建立链接之前进行拦截
        interceptors.addAll(client.interceptors());
        //重试和失败重定向拦截器
        interceptors.add(retryAndFollowUpInterceptor);
        //桥接和适配拦截器。主要补充用户创建请求当中的一些请求头Content-Type等
        interceptors.add(new BridgeInterceptor(client.cookieJar()));
        //主要处理缓存
        interceptors.add(new CacheInterceptor(client.internalCache()));
        //与服务器建立链接。创建可以用的RealConnection(对java.io和java.nio进行了封装)
        interceptors.add(new ConnectInterceptor(client));
        if (!forWebSocket) {
        
          //用户自定义的拦截器，用在与服务器建立链接之后进行拦截。只有非socket进行设置
          interceptors.addAll(client.networkInterceptors());
        }
        //向服务器发送请求和接收数据。将请求写入IO流，再从IO流中读取响应数据
        interceptors.add(new CallServerInterceptor(forWebSocket));
    
        Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
            originalRequest, this, eventListener, client.connectTimeoutMillis(),
            client.readTimeoutMillis(), client.writeTimeoutMillis());
    
        return chain.proceed(originalRequest);
      }

可以看到，Interceptor.Chain的实现类是RealInterceptorChain，最终调用的是RealInterceptorChain.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();
       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);
      // ...
       return response;
     }

来看`interceptor.intercept(next);`，这里以interceptor的其中一个实现类`BridgeInterceptor`为例

@Override 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 != -1) {
        
            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, -1L, Okio.buffer(responseBody)));
        }
    
        return responseBuilder.build();
      }

主要来看`chain.proceed(requestBuilder.build())`，可以看到，这里创建了`requestBuilder`，并调用了下一个`chain`，从而最终调用到向服务器请求的`chain`

> 源码分析中的版本  
> com.squareup.okhttp3:okhttp:3.10.0