手写OkHttp核心代码与责任链详细分析

OkHttp源码看了好多遍,时间长了还是记不住,怎么破? 从头手写一遍OkHttp的核心代码,你就再也不会忘记. 手写一遍对知识进行梳理,更加深入的去了解,当时作者为什么会这样写,这样写的好处是什么?

Ok,要想手写okhttp,就需要对okhttp的源码以及架构有一定的了解,可以去看这篇文章OkHttp 源码解析及OkHttp的设计思想 .我们虽然了解了okhttp的源码,但是并没有真正的掌握okhttp的核心代码,死扣细节才能真正的掌握,哪怕以后使用其他的网络框架也是同样的原理.

本篇文章较长,建议根据文章步骤手敲代码更容易理解.

基本的包装类

请求URL的包装类,主要包装了,host file protocol(是http还是https) port (端口).
主要就是对请求url地址的分解,如对http://www.kuaidi100.com/query?type=yuantong&postid=222222222的分解,host:www.kuaidi100.com/query
file:query?type=yuantong&postid=222222222, protocol:http,prot:80这些数据都要在socket发送请求时用到,详细的使用在下面讲解.

  1. public class HttpUrl {
  2.     private String host;
  3.     private String file;
  4.     private String protocol;
  5.     private int port;
  7.     public HttpUrl(String url) throws MalformedURLException {
  8.         URL urls = new URL(url);
  9.         host = urls.getHost();//host
  10.         file = urls.getFile();// /query?.....
  11.         file = TextUtils.isEmpty(file) ? "/" : file;
  12.         protocol = urls.getProtocol();//http/https
  13.         port = urls.getPort();//端口 如：80
  14.         port = port == -1 ? urls.getDefaultPort() : port;
  15.     }
  17.     public String getHost() {
  18.         return host;
  19.     }
  21.     public String getFile() {
  22.         return file;
  23.     }
  25.     public String getProtocol() {
  26.         return protocol;
  27.     }
  29.     public int getPort() {
  30.         return port;
  31.     }
  32. }

请求的包装类,主要是包装了请求的方式(GET/POST)、请求头、请求URL、请求体的包装,通过builder的方式代码很简单如下:

  1. public class Request {
  2.     //请求头
  3.     private Map<String, String> headers;
  4.     //请求方式 GET/POST
  5.     private Method method;
  6.     //请求URL
  7.     private HttpUrl url;
  8.     //请求体 post请求方式需要用到
  9.     private RequestBody body;
  11.     public Map<String, String> getHeaders() {
  12.         return headers;
  13.     }
  15.     public Method getMethod() {
  16.         return method;
  17.     }
  19.     public HttpUrl getUrl() {
  20.         return url;
  21.     }
  23.     public RequestBody getBody() {
  24.         return body;
  25.     }
  27.     public Request(Builder builder) {
  28.         this.headers = builder.headers;
  29.         this.method = builder.method;
  30.         this.url = builder.url;
  31.         this.body = builder.body;
  32.     }
  34.     public final static class Builder {
  35.         //请求头
  36.         Map<String, String> headers = new HashMap<>();
  37.         //请求方式 GET/POST
  38.         Method method = Method.GET;
  39.         HttpUrl url;
  40.         RequestBody body;
  42.         public Builder url(String url) {
  43.             try {
  44.                 this.url = new HttpUrl(url);
  45.             } catch (MalformedURLException e) {
  46.                 e.printStackTrace();
  47.             }
  48.             return this;
  49.         }
  51.         public Builder addHeader(String name, String value) {
  52.             headers.put(name, value);
  53.             return this;
  54.         }
  56.         public Builder removeHeader(String name) {
  57.             headers.remove(name);
  58.             return this;
  59.         }
  61.         public Builder get() {
  62.             method = Method.GET;
  63.             return this;
  64.         }
  66.         public Builder post(RequestBody body) {
  67.             this.body = body;
  68.             method = Method.POST;
  69.             return this;
  70.         }
  72.         public Request build() {
  73.             if (url == null) {
  74.                 throw new IllegalStateException("HttpUrl this is url,not null");
  75.             }
  77.             return new Request(this);
  78.         }
  79.     }
  80. }

关于请求体RequestBody,这里只用了表单的提交方式,代码如下: 只有post请求的时候才会使用RequestBody, 真正的请求体 就是一个key=value & …字符串

  1. public class RequestBody {
  2.     /**
  3.      * 表单提交 使用urlencoded编码 只模拟表单提交
  4.      */
  5.     private final static String CONTENT_TYPE = "application/x-www-form-urlencoded";
  7.     private final static String CHARSET = "utf-8";
  9.     private Map<String,String> encodedBodys = new HashMap<>();
  11.     public String contentType(){
  12.         return CONTENT_TYPE;
  13.     }
  15.     public long contentLength(){
  16.         return body().getBytes().length;
  17.     }
  19.     /**
  20.      * 真正的请求体 就是一个key:value & ...字符串
  21.      * @return
  22.      */
  23.     public String body(){
  24.         StringBuffer sb = new StringBuffer();
  25.         for (Map.Entry<String, String> entry : encodedBodys.entrySet()) {
  26.             sb.append(entry.getKey())
  27.                     .append("=")
  28.                     .append(entry.getValue())
  29.                     .append("&");
  30.         }
  31.         if (sb.length() != 0){
  32.             sb.deleteCharAt(sb.length()-1);
  33.         }
  34.         return sb.toString();
  35.     }
  37.     public RequestBody add(String name,String value){
  38.         try {
  39.             encodedBodys.put(URLEncoder.encode(name,CHARSET),URLEncoder.encode(value,CHARSET));
  40.         } catch (UnsupportedEncodingException e) {
  41.             e.printStackTrace();
  42.         }
  43.         return this;
  44.     }
  45. }

调度器(Dispatcher)的实现

上述我们实现了最简单的几个基本的包装类,这些其实没有什么可讲解的,下面我们来重点讲解调度器的实现.
新建一个Dispatcher类,主要负责并发调度请求和控制最大的并发请求数.首先我们先要定义最大同时进行的请求数和执行队列和等待队列.队列使用双端队列,新数据添加到尾部,移除的是头部数据.

  1. //最大同时进行的请求数
  2.     private int maxRequests = 64;
  4.     //同时请求的相同的host的最大数
  5.     private int maxRequestsPreHost = 5;
  7.     //等待执行双端队列
  8.     private Deque<Call.AsyncCall> readyAsyncCalls = new ArrayDeque<>();
  10.     //正在执行双端队列
  11.     private Deque<Call.AsyncCall> runningAsyncCalls = new ArrayDeque<>();
  13.    public Dispatcher() {
  14.         this(64,5);
  15.     }
  17.     public Dispatcher(int maxRequests, int maxRequestsPreHost) {
  18.         this.maxRequests = maxRequests;
  19.         this.maxRequestsPreHost = maxRequestsPreHost;
  20.     }

接下来创建一个线程池,所有的任务都在线程池中执行.

  1. //线程池 所有的任务都交给线程池来管理
  2.     private ExecutorService executorService;
  4.     /**
  5.      * 创建一个默认的线程池
  6.      */
  7.     public synchronized ExecutorService executorService(){
  8.         if (executorService == null){
  9.             //线程工厂就是创建线程的
  10.             ThreadFactory threadFactory = new ThreadFactory() {
  12.                 @Override
  13.                 public Thread newThread(Runnable runnable) {
  14.                     return new Thread(runnable,"HttpClient");
  15.                 }
  16.             };
  17.             executorService = new ThreadPoolExecutor(0,Integer.MAX_VALUE,60,
  18.                     TimeUnit.SECONDS,new SynchronousQueue<Runnable>(),threadFactory);
  19.         }
  20.         return executorService;
  21.     }

队列中添加的是Call.AsyncCall,我们先把Call类写完,Call类需要Request的包装类和HttpClient的设置,首先创建一个HttpClient作为的主要类,先完成部分:

  1. public class HttpClient {
  3.  private final Dispatcher dispatcher;
  4.  public HttpClient(Builder builder) {
  5.         this.dispatcher = builder.dispatcher;
  6.  }
  8. public static final class Builder {
  9.      Dispatcher dispatcher;
  10.         /**
  11.          * 用户自定义调度器
  12.          *
  13.          * @param dispatcher
  14.          * @return
  15.          */
  16.         public Builder dispatcher(Dispatcher dispatcher) {
  17.             this.dispatcher = dispatcher;
  18.             return this;
  19.         }
  20.       public HttpClient build() {
  21.             if (null == dispatcher) {
  22.                 dispatcher = new Dispatcher();
  23.             }
  24.             return new HttpClient(this);
  25.         }
  26. }

然后写Call类:

  1. //需要请求的包装类
  2.     Request request;
  4.     //HttpClient 中配置的参数
  5.     HttpClient client;
  7.     //标记是否执行过
  8.     boolean executed = false;
  10.     //标记是否取消请求
  11.     boolean cancel = false;
  13.     public Call(Request request, HttpClient client) {
  14.         this.request = request;
  15.         this.client = client;
  16.     }

在上述代码中Dispatcher,任务是在线程池中执行的,而Call类就是要执行的任务,所以我们需要加入一个执行网络请求的线程内部类AsyncCall如下:
我们先建立一个空类,run如何执行我们在后面讲解,

  1. /**
  2.      * 执行网络请求的线程
  3.      */
  4.     class AsyncCall implements Runnable {
  6.         private CallBack callBack;
  8.         public AsyncCall(CallBack callBack) {
  9.             this.callBack = callBack;
  10.         }
  12.         @Override
  13.         public void run() {}
  15.         public String host() {
  16.             return request.getUrl().getHost();
  17.         }
  18.     }

同时我们还需要一个回调类CallBack

  1. public interface CallBack {
  2.     void onFailure(Call call,Throwable throwable);
  3.     void onResponse(Call call, Response response);
  4. }

然后创建equeue方法,将AsyncCall传递给Dispatcher执行.通过HttpClient获取Dispatcher代码如下:

  1. public void enqueue(CallBack callBack) {
  2.         synchronized (this) {
  3.             if (executed) {
  4.                 throw new IllegalStateException("已经执行过了");
  5.             }
  6.             //标记已经执行过了
  7.             executed = true;
  9.         }
  10.         //把任务交给调度器调度
  11.         client.getDispatcher().enqueue(new AsyncCall(callBack));
  12.     }

然后回调Dispatcher类中,创建enqueue来处理任务将任务加入到线程池中执行.
首先,我们先要判断正在执行的Call是否超过了最大请求数与最大相同host请求数,这里我们就用到了HttpUrl中的host,如下代码,返回了将要执行Call的host请求数,这里的host从Call.AsyncCall中通过Call传递的Request返回的HttpUrl中包装的host.

  1. /**
  2.      * 当前正在执行的host
  3.      * @param call 正在执行的host
  4.      * @return
  5.      */
  6.     private int runningCallsForHost(Call.AsyncCall call){
  7.         int result = 0;
  8.         for (Call.AsyncCall aysncCall : runningAsyncCalls) {
  9.              //正在执行队列 和当前将要执行的call的host进行比对，如果相等计数加1
  10.              if (aysncCall.host().equals(call.host())){
  11.                  result++;
  12.              }
  13.         }
  14.         return result;
  15.     }

如果执行队列超过了数量的限制则将Call加入到等待队列中.enqueue方法如下:

  1. /**
  2.      * 异步任务调度
  3.      */
  4.     public void enqueue(Call.AsyncCall call){
  5.         //将要执行的call，判断正在执行的call不能超过最大请求数与相同host的请求数
  6.         if (runningAsyncCalls.size()<maxRequests && runningCallsForHost(call)<maxRequestsPreHost){
  7.             runningAsyncCalls.add(call);
  8.             executorService().execute(call);
  9.         }else {
  10.             //如果超过了限制的数量 则将call加入到等待队列中
  11.             readyAsyncCalls.add(call);
  12.         }
  13.     }

OK,在上述的代码中,我们实现了调度器异步的任务调度,通过HttpClient来初始化Call,传递Request和HttpClient,然后通过Call类中的enqueue方法new AnsycCall线程,然后交给调度器处理,调度器是通过HttpClient获取到的.
我们要在HttpClient中加入newCall方法:

  1. public Call newCall(Request request) {
  2.         return new Call(request, this);
  3. }

既然有添加任务请求,那么肯定有完成任务请求,因为我们将任务加入到了队列中,当任务完成时,我们需要将任务移除.
那么如何判断任务请求完成了呢? 我们需要在Call类中添加getResponse方法这个方法负责将请求结果返回.我们暂时先写个空方法,我会在后面带大家一步一步实现

  1. private Response getResponse() throws Exception {}

getResponse()方法,在AsyncCall类中的run 方法中调用,这样就补全了上述AsyncCall类中空的run方法,代码如下:
run方法实现也很简单通过一个singalledCallbacked标示是否回调过了,调用getResponse方法获取请求结果,然后判断是否取消请求了的处理,最终在finally中调用了调度器dispacher的finished表示任务请求完毕.

  1. @Override
  2.         public void run() {
  3.             //信号 是否回调过
  4.             boolean singalledCallbacked = false;
  5.             try {
  6.                 //真正的实现请求逻辑
  7.                 Response response = getResponse();
  8.                 //如果取消了请求，就回调一个onFailure
  9.                 if (cancel) {
  10.                     //回调通知过了
  11.                     singalledCallbacked = true;
  12.                     callBack.onFailure(Call.this, new IOException("Canceled"));
  13.                 } else {
  14.                     singalledCallbacked = true;
  15.                     //链接成功了
  16.                     callBack.onResponse(Call.this, response);
  17.                 }
  18.             } catch (Exception e) {
  19.                 e.printStackTrace();
  20.                 //如果信号没有通知过 则回调
  21.                 if (!singalledCallbacked) {
  22.                     callBack.onFailure(Call.this, e);
  23.                 }
  24.             } finally {
  25.                 //将这个任务从调度器移除
  26.                 client.getDispatcher().finished(this);
  27.             }
  28.         }

我们来看一下调度器的finished方法是如何实现的呢? 思路是这样的:首先将这个任务从执行队列中移除,并且检查正在执行队列的数量是否达到最大和等待队列中是否还有等待中的任务,如果有等待中的任务,则通过while循环队列,依次从等待队列中移除和加入到正在执行队列中执行,同时要判断是否达到了最大相同host数量和最大请求数量,代码如下:

  1. public void finished(Call.AsyncCall asyncCall) {
  2.         synchronized (this){
  3.             runningAsyncCalls.remove(asyncCall);
  4.             //检查是否可以运行ready
  5.             checkReady();
  6.         }
  7.     }
  8. private void checkReady() {
  9.         //达到了同时请求最大数
  10.         if(runningAsyncCalls.size()>=maxRequests){
  11.             return;
  12.         }
  13.         //没有等待执行的任务
  14.         if (readyAsyncCalls.isEmpty()){
  15.             return;
  16.         }
  17.         Iterator<Call.AsyncCall> iterator = readyAsyncCalls.iterator();
  18.         while (iterator.hasNext()){
  19.             //获得一个等待执行的任务
  20.             Call.AsyncCall asyncCall = iterator.next();
  21.             //如果等待执行的任务，加入正在执行小于最大相同host数
  22.             if (runningCallsForHost(asyncCall) < maxRequestsPreHost){
  23.                 iterator.remove();//从等待执行列表移除
  24.                 runningAsyncCalls.add(asyncCall);
  25.                 executorService().execute(asyncCall);
  26.             }
  27.             //如果正在执行队列达到了最大值，则不在请求 return
  28.             if (runningAsyncCalls.size() >= maxRequests){
  29.                 return;
  30.             }
  31.         }
  32.     }

调度器的同步请求实现,Call类中添加execute方法,同样也要标记是否执行过了,
然后请求网络返回结果,最终调用finished方法

  1. public Response execute() throws Exception {
  2.         synchronized (this) {
  3.             if (executed) throw new IllegalStateException("Already Executed");
  4.             executed = true;
  5.         }
  6.         try {
  7.             client.getDispatcher().execute(this);
  8.             Response response = getResponse();
  9.             if (response == null) throw new IOException("Canceled");
  10.             return response;
  11.         } catch (Exception e) {
  12.             throw e;
  13.         } finally {
  14.             client.getDispatcher().finished(this);
  15.         }
  16.     }

Dispatcher类添加execute方法,Dispacher中处理就相当简单了,我们只需要将这个任务添加到队列中就可以了,不用在线程池中执行也不用判断是否达到最大请求,因为同步执行,只能执行一个请求,代码如下:

  1. private Deque<Call> runningSyncCall = new ArrayDeque<>();
  2.  public void execute(Call call) {
  3.         runningSyncCall.add(call);
  4.  }
  5.  public void finished(Call call) {
  6.         synchronized (this) {
  7.             if (!runningSyncCall.remove(call)) throw new AssertionError("Call wasn't in-flight");
  8.         }
  9.     }

这样整个调度器的实现就全部完成了. 下面我们来责任链也是最核心的部分,代码地址 还有不清楚的直接看代码.

责任链详细分析

我们知道其实okhttp的每个责任链都是一个拦截器,首先我们要实现拦截器接口

  1. public interface Interceptor {
  2.     Response interceptor(InterceptorChain chain) throws IOException;
  3. }

InterceptorChain 类主要作用是执行拦截器,将链条一条一条的执行下去.

  1. public class InterceptorChain {
  2.     List<Interceptor> interceptors;
  3.     int index;
  4.     Call call;
  6.     public InterceptorChain(List<Interceptor> interceptors, int index, Call call) {
  7.          this.interceptors = interceptors;
  8.         this.index = index;
  9.         this.call = call;
  10.     }
  12.     /**
  13.      * 执行拦截器
  14.      */
  15.     public Response process() throws IOException {
  16.         if (index >= interceptors.size()) throw new IOException("Interceptor Chain Error");
  17.         //获得拦截器 从第0个拦截器开始 
  18.         Interceptor interceptor = interceptors.get(index);
  19.         //链条一条一条执行 同时index加1
  20.         InterceptorChain next = new InterceptorChain(interceptors, index + 1, call);
  21.         Response response = interceptor.interceptor(next);
  22.         return response;
  23.     }
  24. }

重试与重定向拦截器 RetryInterceptor

这里我只是实现了很简单的重试拦截器,通过for循环来循环重试次数,如果有respone响应返回则说明请求成功,跳出循环.okhttp的实现就较为复杂,通过while(true)循环来重试如果返回了response不为空则跳出循环,是一样的原理.

  1. public class RetryInterceptor implements Interceptor {
  2.     private static final String TAG = "RetryInterceptor";
  4.     @Override
  5.     public Response interceptor(InterceptorChain chain) throws IOException {
  6.         Log.e(TAG, "interceptor: RetryInterceptor");
  7.         Call call = chain.call;
  8.         HttpClient client = call.getClient();
  9.         IOException exception = null;
  10.         for (int i = 0; i < client.getRetrys() + 1; i++) {
  11.             //如果取消了则抛出异常
  12.             if (call.isCanceled()) {
  13.                 throw new IOException("Canceled");
  14.             }
  15.             try {
  16.                 //执行链条中下一个拦截器 如果有返回response 则表示请求成功直接return结束for循环
  17.                 Response response = chain.process();
  18.                 return response;
  19.             } catch (IOException e) {
  20.                 exception = e;
  21.             }
  22.         }
  23.         throw exception;
  24.     }
  25. }

请求头拦截器 BridgeInterceptor

负责把用户构造的请求转换为发送到服务器的请求 、把服务器返回的响应转换为用户友好的响应 处理 配置请求头等信息. 从应用程序代码到网络代码的桥梁。首先，它根据用户请求构建网络请求。然后它继续呼叫网络。最后，它根据网络响应构建用户响应。BridgeInterceptor 主要是添加一些默认的请求头,和对响应数据的处理.
代码如下: 通过InterceptorChain获取到正在执行的Call对象,然后通过Call获取到request, BridgeInterceptor 其实就是给Request的headers添加一些默认的请求头.

在Call类中的getResponse()方法中添加如下代码:
将我们写好的两个拦截器添加,注意顺序不能出错,自定义的拦截器一定是添加到第一个位置.

  1. private Response getResponse() throws Exception {
  2.         ArrayList<Interceptor> interceptors = new ArrayList<>();
  3.         //自定义拦截器
  4.         interceptors.addAll(client.getInterceptors());
  5.         //添加重试拦截器
  6.         interceptors.add(new RetryInterceptor());
  7.         //添加请求头拦截器
  8.         interceptors.add(new BridgeInterceptor());
  9.         //创建链
  10.         InterceptorChain chain = new InterceptorChain(interceptors, 0, this);
  11.         //执行责任链
  12.         return chain.process();
  13.     }

BridgeInterceptor的实现:

  1. public class BridgeInterceptor implements Interceptor {
  2.     private static final String TAG = "BridgeInterceptor";
  4.     @Override
  5.     public Response interceptor(InterceptorChain chain) throws IOException {
  6.         Log.e(TAG, "interceptor: BridgeInterceptor");
  7.         //必须有host connection:keep-alive 保持长连接
  8.         Request request = chain.call.request();
  9.         Map<String, String> headers = request.getHeaders();
  10.         //如果没有配置Connection 默认给添加上
  11.         if (!headers.containsKey(HEAD_CONNECTION)) {
  12.              //保持长连接
  13.             headers.put(HEAD_CONNECTION, HEAD_VALUE_KEEP_ALIVE);
  14.         }
  15.         //host 必须和url中的host一致的
  16.         headers.put(HEAD_HOST, request.getUrl().getHost());
  17.         //是否有请求体 如果有请求体需要添加请求体的长度和请求体的类型
  18.         if (null != request.getBody()) {
  19.             //获取到RequestBody
  20.             RequestBody body = request.getBody();
  21.             //获取请求体的长度
  22.             long contentLength = body.contentLength();
  23.             //请求体长度
  24.             if (contentLength != 0) {
  25.                 headers.put(HEAD_CONTENT_LENGTH, String.valueOf(contentLength));
  26.             }
  27.             //请求体类型，这里只实现了一种，其他的同样的道理
  28.             String contentType = body.contentType();
  29.             if (null != contentType) {
  30.                 headers.put(HEAD_CONTENT_TYPE, contentType);
  31.             }
  32.         }
  33.         Log.e(TAG, "BridgeInterceptor: 设置的请求头");
  34.         for (Map.Entry<String, String> entry : headers.entrySet()) {
  35.             Log.e(TAG, "BridgeInterceptor key:" + entry.getKey() + " value:" + entry.getValue());
  36.         }
  37.         //执行下一个链
  38.         return chain.process();
  39.     }
  40. }

一些常量静态变量存放在HttpCodec类中:

  1. //定义拼接常用的常量
  2.     static final String CRLF = "\r\n";
  3.     static final int CR = 13;
  4.     static final int LF = 10;
  5.     static final String SPACE = " ";
  6.     static final String VERSION = "HTTP/1.1";
  7.     static final String COLON = ":";
  9.     public static final String HEAD_HOST = "Host";
  10.     public static final String HEAD_CONNECTION = "Connection";
  11.     public static final String HEAD_CONTENT_TYPE = "Content-Type";
  12.     public static final String HEAD_CONTENT_LENGTH = "Content-Length";
  13.     public static final String HEAD_TRANSFER_ENCODING = "Transfer-Encoding";
  15.     public static final String HEAD_VALUE_KEEP_ALIVE = "Keep-Alive";
  16.     public static final String HEAD_VALUE_CHUNKED = "chunked";

ConnectionInterceptor

获得有效链接的拦截器, 主要功能是从连接池中获取可复用的连接，如果没有可复用的连接则创建连接添加到连接池中，以提供下次的复用.ConnectionInterceptor的实现较为复杂.

首先实现连接池ConnectionPool类,同样的连接池可以被用户自己实现连接池的规则,通过HttpClient来获取连接池,在HttpClient类中加入如下代码:
同时也将自定义拦截器的list也加入

  1. private final ConnectionPool connectionPool;
  3.  private final List<Interceptor> interceptors;
  5.  public ConnectionPool getConnectionPool() {
  6.         return connectionPool;
  7.     }
  8. public HttpClient(Builder builder) {
  9.         ....
  10.         this.connectionPool = builder.connectionPool;
  11.         this.interceptors = builder.interceptors;
  12.         ....
  13.     }
  14.  public List<Interceptor> getInterceptors() {
  15.         return interceptors;
  16.     }

在HttpClient的内部类Builder类中加入如下代码:

  1. ConnectionPool connectionPool;
  2. List<Interceptor> interceptors = new ArrayList<>();
  3.  /**
  4.          * 添加自定义拦截器
  5.          *
  6.          * @param interceptor
  7.          * @return
  8.          */
  9.         public Builder addInterceptor(Interceptor interceptor) {
  10.             interceptors.add(interceptor);
  11.             return this;
  12.         }
  14.         /**
  15.          * 添加自定义的连接池
  16.          *
  17.          * @param connectionPool
  18.          * @return
  19.          */
  20.         public Builder connectionPool(ConnectionPool connectionPool) {
  21.             this.connectionPool = connectionPool;
  22.             return this;
  23.         }
  24. public HttpClient build() {
  25.             if (null == dispatcher) {
  26.                 dispatcher = new Dispatcher();
  27.             }
  28.             if (null == connectionPool) {
  29.                 connectionPool = new ConnectionPool();
  30.             }
  31.             return new HttpClient(this);
  32.         }

Ok,加入完毕后我们来看如何实现连接池ConnectionPool类,首先我们要知道连接池的作用是什么? 实现思路: 首先初始化每个连接的保持时间,如果这个连接超出定义的保持时间则将此连接移除连接池,如果没有超出连接时间则复用这个连接,通过host和port来判断连接池中是否存在这个连接. 那么什么时候将这个连接加入到连接池呢? 在获取到服务器的响应的时候判断响应头``是否允许保持长连接,如果允许将此连接加入到连接池中,然后检查连接池中是否有超过最长时间的限制连接.
首先我们定义三个变量:

  1. /**
  2.      * 每个链接的检查时间,默认60s
  3.      * <p>
  4.      * 例如:每隔5s检查是否可用，无效则将其从链接池移除
  5.      */
  6.     private long keepAlive;
  7.     //是否清理了连接池
  8.     private boolean cleanupRunning = false;
  9.     //HttpConnection 包装的连接类 存储连接队列
  10.     private Deque<HttpConnection> connectionDeque = new ArrayDeque<>();
  11.    public ConnectionPool() {
  12.         this(1, TimeUnit.MINUTES);
  13.     }
  15.     public ConnectionPool(long keepAlive, TimeUnit utils) {
  16.         this.keepAlive = utils.toMillis(keepAlive);
  17.     }

HttpConnction 是真正的请求连接的包装类, 连接池中存储的就是该类,我们先实现连接: 这里我们先不实现Socket请求服务器的方法,我会在后面进行实现,我们先实现连接池中需要用的方法和变量

  1. Socket socket;
  2.     Request request;
  3.     private HttpClient client;
  4.     /**
  5.      * 当前链接的socket是否与对应的host port一致
  6.      *
  7.      * @param host
  8.      * @param port
  9.      * @return
  10.      */
  11.     public boolean isSameAddress(String host, int port) {
  12.         if (null == socket) {
  13.             return false;
  14.         }
  15.         return TextUtils.equals(socket.getInetAddress().getHostName(), host) && (port == socket.getPort());
  16.     }
  18.     /**
  19.      * 释放关闭 socket
  20.      */
  21.     public void close() {
  22.         if (null != socket) {
  23.             try {
  24.                 socket.close();
  25.             } catch (IOException e) {
  26.                 e.printStackTrace();
  27.             }
  28.         }
  29.     }
  31.     public void setRequest(Request request) {
  32.         this.request = request;
  33.     }
  35.     public void setClient(HttpClient client) {
  36.         this.client = client;
  37.     }

好下面我们实现ConnectionPool连接池的获取连接的方法get:
Get方法很简单遍历连接队列,如果有相同的host和port的连接则,将其从连接池中移除复用该连接,注意get方法需要synchronized.

  1. public synchronized HttpConnection get(String host, int port) {
  2.         Iterator<HttpConnection> iterator = connectionDeque.iterator();
  3.         while (iterator.hasNext()) {
  4.             HttpConnection next = iterator.next();
  5.             //如果查找到链接池中存在相同host port 的链接就可以直接使用
  6.             if (next.isSameAddress(host, port)) {
  7.                 iterator.remove();
  8.                 return next;
  9.             }
  10.         }
  11.         return null;
  12.     }

下面看如何将连接加入到连接池中,代码如下.判断是否执行清理连接线程,如果没有则在线程池中执行,这样可以防止多次加入连接池,执行多次线程池.在连接加入到连接池后需要在线程池中遍历所有的闲置连接,超出时间则将连接移除连接池,直到所有连接移除连接池,执行完毕.

  1. /**
  2.      * 加入链接到链接池
  3.      *
  4.      * @param connection
  5.      */
  6.     public void put(HttpConnection connection) {
  7.         //如果没有执行清理线程 则执行
  8.         if (!cleanupRunning) {
  9.             cleanupRunning = true;
  10.             executor.execute(cleanupRunnable);
  11.         }
  12.         connectionDeque.add(connection);
  13.     }
  15.  /**
  16.      * 执行清理线程的线程池
  17.      */
  18.     private static final Executor executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
  19.             new SynchronousQueue<Runnable>(), new ThreadFactory() {
  20.         @Override
  21.         public Thread newThread(@NonNull Runnable r) {
  22.             Thread thread = new Thread("Connection Pool");
  23.             //设置为守护线程 有什么用呢？
  24.             thread.setDaemon(true);
  25.             return thread;
  26.         }
  27.     });

我们来看一下cleanupRunnable线程的实现.代码如下.
看下面的逻辑如果当前的时间传入给cleanup()方法,如果返回-1则说明连接池中已没有任何连接,循环结束.如果返回的不是-1则等待返回的时间,将线程暂时挂起,然后在继续执行.

cleanup方法的实现是遍历所有连接,判断其是否超过了最大的闲置时间,如果超过了则进行移除,关闭socket连接.然后继续遍历下一个,记录下所有连接中没有超过最长闲置时间的最长的时间,然后返回keepAlive - longestIdleDuration.将线程挂起这么长的时间后,重新遍历所有连接然后清除.直到所有连接清除完毕.

  1. /**
  2.      * 清理链接池的线程
  3.      */
  4.     private Runnable cleanupRunnable = new Runnable() {
  5.         @Override
  6.         public void run() {
  7.             while (true) {
  8.                 //得到下次的检查时间
  9.                 long waitDuration = cleanup(System.currentTimeMillis());
  10.                 //如果返回-1 则说明连接池中没有连接 直接结束
  11.                 if (waitDuration < 0) {
  12.                     return;
  13.                 }
  14.                 if (waitDuration > 0) {
  15.                     synchronized (ConnectionPool.this) {
  16.                         try {
  17.                             //线程暂时被挂起
  18.                             ConnectionPool.this.wait(waitDuration);
  19.                         } catch (InterruptedException e) {
  20.                             e.printStackTrace();
  21.                         }
  22.                     }
  23.                 }
  25.             }
  26.         }
  27.     };
  28.  private long cleanup(long now) {
  29.         //记录比较每个链接的闲置时间
  30.         long longestIdleDuration = -1;
  31.         synchronized (this) {
  32.             Iterator<HttpConnection> iterator = connectionDeque.iterator();
  33.             //为什么要迭代它呢？ 如果某个链接在最长的闲置时间没有使用则进行移除
  34.             while (iterator.hasNext()) {
  35.                 HttpConnection connection = iterator.next();
  36.                 //获取这个链接的闲置时间
  37.                 long idleDuration = now - connection.lastUseTime;
  38.                 //如果闲置时间超过了最大的闲置时间则进行移除
  39.                 if (idleDuration > keepAlive) {
  40.                     iterator.remove();
  41.                     //释放关闭连接
  42.                     connection.close();
  43.                     Log.e("connection pool", "cleanup: 超过闲置时间，移除链接池");
  44.                     //继续检查下一个
  45.                     continue;
  46.                 }
  47.                 //记录最长的闲置时间
  48.                 if (longestIdleDuration < idleDuration) {
  49.                     longestIdleDuration = idleDuration;
  50.                 }
  51.             }
  52.             //假如keepAlive 10s longestIdleDuration是5s 那么就等5s后在检查连接池中的连接
  53.             if (longestIdleDuration > 0) {
  54.                 return keepAlive - longestIdleDuration;
  55.             }
  56.             //标记连接池中没有连接
  57.             cleanupRunning = false;
  58.             return longestIdleDuration;
  59.         }
  60.     }

Ok,以上就是整个连接池的实现,是不是get到了很多技能点.你可以去看okhttp的代码也是这样实现的,不过okhttp在ConnectionPool中还加入了最大连接数量的判断,感兴趣的可以去了解一番.

讲了这么多ConnectionInterceptor类还没有实现, ConnectionInterceptor的作用主要是从连接池中获取有效的连接(HttpConnection)然后将这个有效的连接(HttpConnection),传递给下一个拦截器来实现真正的连接请求以及获取服务器的响应.
代码如下.很简单我们通过chain获取到request和HttpClient,然后通过HttpClient来获取到连接池,通过request过去到host和port传递给连接池的get方法来获得有效连接,如果没有可复用的连接则new HttpConnection,创建一个连接.然后将该连接传递到下一个责任链中,来实现真正的通信.当责任链返回最终的服务器的响应然后判断返回的响应是否允许保持长连接,如果允许,将该连接加入到连接池中复用.如果不允许关闭该连接.
OK,至此整个ConnectionInterceptor到逻辑就完成了.

  1. public class ConnectionInterceptor implements Interceptor {
  2.     private static final String TAG = "ConnectionInterceptor";
  4.     @Override
  5.     public Response interceptor(InterceptorChain chain) throws IOException {
  6.         Log.e(TAG, "interceptor: ConnectionInterceptor");
  7.         Request request = chain.call.request();
  8.         HttpClient client = chain.call.getClient();
  9.         //获取http url
  10.         HttpUrl url = request.getUrl();
  11.         //从连接池中获取连接 需要具有相同的host 和 port
  12.         HttpConnection httpConnection = client.getConnectionPool().get(url.getHost(), url.getPort());
  13.         //没有可复用的连接
  14.         if (httpConnection == null) {
  15.             Log.e(TAG, "ConnectionInterceptor: 新建连接");
  16.             httpConnection = new HttpConnection();
  17.         } else {
  18.             Log.e(TAG, "ConnectionInterceptor: 从连接池中获取连接");
  19.         }
  20.         //将request传递给连接
  21.         httpConnection.setRequest(request);
  22.         //将client传递给连接
  23.         httpConnection.setClient(client);
  24.         try {
  25.             //执行下一个拦截器，将连接传递给下一个拦截器
  26.             Response process = chain.process(httpConnection);
  27.             //如果服务器返回的响应，如果服务其允许长连接
  28.             if (process.isKeepAlive) {
  29.                 //将连接添加到连接池中
  30.                 Log.e(TAG, "ConnectionInterceptor: 得到服务器响应：isKeepAlive=true,保持长连接，将此连接加入到连接池中");
  31.                 client.getConnectionPool().put(httpConnection);
  32.             } else {
  33.                 //如果不允许保持连接 则使用连接完毕后直接关闭连接
  34.                 Log.e(TAG, "ConnectionInterceptor: 得到服务器响应：isKeepAlive=false,不保持长连接，关闭连接");
  35.                 httpConnection.close();
  36.             }
  37.             return process;
  38.         } catch (IOException e) {
  39.             httpConnection.close();
  40.             throw e;
  41.         }
  42.     }
  43. }

在InterceptorChain类中加入以下覆写方法,主要是将HttpConnection进行赋值
在通信拦截器中就可以直接获取到连接,来实现通信.

  1. HttpConnection httpConnection; 
  2. public InterceptorChain(List<Interceptor> interceptors, int index, Call call, HttpConnection httpConnection) {
  3.         this.interceptors = interceptors;
  4.         this.index = index;
  5.         this.call = call;
  6.         this.httpConnection = httpConnection;
  7.     }
  9.     /**
  10.      * 使下一个拦截器拿到HttpConnection
  11.      *
  12.      * @param connection
  13.      * @return
  14.      */
  15.     public Response process(HttpConnection connection) throws IOException {
  16.         this.httpConnection = connection;
  17.         return process();
  18.     }

通信拦截器CallServerInterceptor

首先我们了完善HttpConnection类,来实现socket通信.代码如下.很简单的基础问题,注意https需要创建socketFactory,可以允许用户自己设置,在HttpClient中设置即可.

  1. /**
  2.      * 与服务器通信
  3.      *
  4.      * @return InputStream 服务器返回的数据
  5.      */
  6.     public InputStream call(HttpCodec httpCodec) throws IOException {
  7.         //创建socket
  8.         createSocket();
  9.         //发送请求
  10.         //按格式拼接 GET 地址参数 HTTP
  11.         httpCodec.writeRequest(out, request);
  12.         //返回服务器响应(InputStream)
  13.         return in;
  14.     }
  16.     private InputStream in;
  17.     private OutputStream out;
  19.     /**
  20.      * 创建socket
  21.      */
  22.     private void createSocket() throws IOException {
  23.         if (null == socket || socket.isClosed()) {
  24.             HttpUrl httpUrl = request.getUrl();
  25.             //如果是https
  26.             if (httpUrl.getProtocol().equalsIgnoreCase("https")) {
  27.                 //也可以用户自己设置
  28.                 socket = client.getSocketFactory().createSocket();
  29.             } else {
  30.                 socket = new Socket();
  31.             }
  32.             socket.connect(new InetSocketAddress(httpUrl.getHost(), httpUrl.getPort()));
  33.             in = socket.getInputStream();
  34.             out = socket.getOutputStream();
  35.         }
  36.     }

我们在HttpCodec类中来拼接http协议信息,不懂可以看这个地址了解.主要是拼接:请求行(如GET请求:GET /query?type=yuantong&postid=222222222 / HTTP/1.1 rn),请求头这个就不说了,还有请求体post请求需要用到.

  1. public void writeRequest(OutputStream out, Request request) throws IOException {
  2.         StringBuffer sb = new StringBuffer();
  3.         //请求行 GET / 。。。。。/ HTTP/1.1\r\n
  4.         sb.append(request.getMethod());
  5.         sb.append(SPACE);
  6.         sb.append(request.getUrl().getFile());
  7.         sb.append(SPACE);
  8.         sb.append(VERSION);
  9.         sb.append(CRLF);
  11.         //请求头
  12.         Map<String, String> headers = request.getHeaders();
  13.         for (Map.Entry<String, String> entry : headers.entrySet()) {
  14.             sb.append(entry.getKey());
  15.             sb.append(COLON);
  16.             sb.append(SPACE);
  17.             sb.append(entry.getValue());
  18.             sb.append(CRLF);
  19.         }
  20.         sb.append(CRLF);
  22.         //请求体 POST 请求会用到
  23.         RequestBody body = request.getBody();
  24.         if (null != body) {
  25.             sb.append(body.body());
  26.         }
  27.         out.write(sb.toString().getBytes());
  28.         out.flush();
  29.     }

HttpCodec来读取服务器返回来的响应.
例如读取如下响应头信息:

  1. Server:    nginx
  2. Date:    
  3. Mon, 19 Aug 2019 10:43:42 GMT-1m 16s
  4. Content-Type:    text/html;charset=UTF-8
  5. Transfer-Encoding:    chunked
  6. Connection:    keep-alive
  7. Vary:    Accept-Encoding
  8. P3P:    CP="IDC DSP COR ADM DEVi TAIi PSA PSD IVAi IVDi CONi HIS OUR IND CNT"
  9. Cache-Control:    
  10. no-cache
  11. Content-Encoding:    gzip

读取一行

  1. /**
  2.      * 读取一行
  3.      *
  4.      * @param in 服务器返回的数据
  5.      * @return
  6.      * @throws IOException
  7.      */
  8.     public String readLine(InputStream in) throws IOException {
  9.         //清理
  10.         byteBuffer.clear();
  11.         //标记
  12.         byteBuffer.mark();
  13.         boolean isMabeEofLine = false;
  14.         byte b;
  15.         while ((b = (byte) in.read()) != -1) {
  16.             byteBuffer.put(b);
  17.             //如果读到一个\r
  18.             if (b == CR) {
  19.                 isMabeEofLine = true;
  20.             } else if (isMabeEofLine) {
  21.                 //读到\n一行结束
  22.                 if (b == LF) {
  23.                     //一行数据
  24.                     byte[] lineBytes = new byte[byteBuffer.position()];
  25.                     //将标记设置为0
  26.                     byteBuffer.reset();
  27.                     //从allocate获得数据
  28.                     byteBuffer.get(lineBytes);
  29.                     byteBuffer.clear();
  30.                     byteBuffer.mark();
  31.                     //将一行数据返回
  32.                     return new String(lineBytes);
  33.                 }
  34.                 //如果下一个不是\n 置为false
  35.                 isMabeEofLine = false;
  36.             }
  37.         }
  38.         //如果读完了都没有读到 则服务器出现问题
  39.         throw new IOException("Response read line");
  40.     }

读取响应头信息

  1. /**
  2.      * 读取响应头
  3.      *
  4.      * @param in
  5.      * @return
  6.      */
  7.     public Map<String, String> readHeader(InputStream in) throws IOException {
  8.         HashMap<String, String> headers = new HashMap<>();
  9.         while (true) {
  10.             //读取一行
  11.             String line = readLine(in);
  12.             //如果读到空行\r\n 表示响应头读取完毕了
  13.             if (isEmptyLine(line)) {
  14.                 break;
  15.             }
  16.             //读取响应头的key value
  17.             int index = line.indexOf(":");
  18.             if (index > 0) {
  19.                 String key = line.substring(0, index);
  20.                 //key与value还有空格
  21.                 String value = line.substring(index + 2, line.length() - 2);
  22.                 headers.put(key, value);
  23.             }
  24.         }
  25.         return headers;
  26.     }

CallServerInterceptor的完整实现如下代码:最终将响应信息返回.

  1. public class CallServerInterceptor implements Interceptor {
  2.     private static final String TAG = "CallServerInterceptor";
  4.     @Override
  5.     public Response interceptor(InterceptorChain chain) throws IOException {
  6.         Log.e(TAG, "interceptor: CallServerInterceptor");
  7.         HttpConnection connection = chain.httpConnection;
  8.         //进行I/O操作
  9.         HttpCodec httpCodec = new HttpCodec();
  10.         InputStream in = connection.call(httpCodec);
  11.         //读取响应
  12.         //读取响应行: HTTP/1.1 200 OK\r\n
  13.         String statusLine = httpCodec.readLine(in);
  14.         Log.e(TAG, "CallServerInterceptor: 得到响应行：" + statusLine);
  15.         //读取响应头
  16.         Map<String, String> headers = httpCodec.readHeader(in);
  17.         for (Map.Entry<String, String> entry : headers.entrySet()) {
  18.             Log.e(TAG, "CallServerInterceptor: 得到响应头 key:" + entry.getKey() + " value:" + entry.getValue());
  19.         }
  20.         //读取响应体
  21.         //判断请求头是否有 content-length 如果有就直接读取这大的长度就可以
  22.         int content_length = -1;
  23.         if (headers.containsKey(HEAD_CONTENT_LENGTH)) {
  24.             content_length = Integer.valueOf(headers.get(HEAD_CONTENT_LENGTH));
  25.         }
  26.         //根据分块编码解析
  27.         boolean isChunked = false;
  28.         if (headers.containsKey(HEAD_TRANSFER_ENCODING)) {
  29.             isChunked = headers.get(HEAD_TRANSFER_ENCODING).equalsIgnoreCase(HEAD_VALUE_CHUNKED);
  30.         }
  31.         String body = null;
  32.         if (content_length > 0) {
  33.             byte[] bytes = httpCodec.readBytes(in, content_length);
  34.             body = new String(bytes);
  35.         } else if (isChunked) {
  36.             body = httpCodec.readChunked(in);
  37.         }
  38.         //status[1] 就是响应码
  39.         String[] status = statusLine.split(" ");
  40.         //判断服务器是否允许长连接
  41.         boolean isKeepAlive = false;
  42.         if (headers.containsKey(HEAD_CONNECTION)) {
  43.             isKeepAlive = headers.get(HEAD_CONNECTION).equalsIgnoreCase(HEAD_VALUE_KEEP_ALIVE);
  44.         }
  45.         //更新一下这个连接的时间
  46.         connection.updateLastUserTime();
  47.         //返回响应包装类
  48.         return new Response(Integer.valueOf(status[1]), content_length, headers, body, isKeepAlive);
  49.     }
  50. }

需要注意的是,要在HttpConnection中添加updateLastUserTime方法来更新当前连接的时间,用到连接池ConnectionPool中.

最后完善Call类的getResponse方法:

  1. private Response getResponse() throws Exception {
  2.         ArrayList<Interceptor> interceptors = new ArrayList<>();
  3.         //自定义拦截器
  4.         interceptors.addAll(client.getInterceptors());
  5.         //添加重试拦截器
  6.         interceptors.add(new RetryInterceptor());
  7.         //添加请求头拦截器
  8.         interceptors.add(new BridgeInterceptor());
  9.         //添加连接拦截器
  10.         interceptors.add(new ConnectionInterceptor());
  11.         //添加通信拦截器
  12.         interceptors.add(new CallServerInterceptor());
  14.         InterceptorChain chain = new InterceptorChain(interceptors, 0, this, null);
  15.         return chain.process();
  16.     }

测试

OK,我们花费了好大的力气终于写完了一个类似okhttp的网络请求框架.我们需要测试看看是否成立.
创建HttpClient

  1. httpClient = new HttpClient.Builder().retrys(3).build();

测试get方法

  1. Request request = new Request.Builder()
  2.                 .get()
  3.                 .url("http://www.kuaidi100.com/query?type=yuantong&postid=222222222")
  4.                 .build();
  5.         Call call = httpClient.newCall(request);
  6.         call.enqueue(new CallBack() {
  7.             @Override
  8.             public void onFailure(Call call, Throwable throwable) {
  10.             }
  12.             @Override
  13.             public void onResponse(Call call, Response response) {
  14.                 Log.e(TAG, "onResponse: " + response.getBody());
  15.             }
  16.         });

LOG如下:

  1. interceptor: RetryInterceptor
  2. BridgeInterceptor: interceptor: BridgeInterceptor
  3. BridgeInterceptor: BridgeInterceptor: 设置的请求头
  4. BridgeInterceptor: BridgeInterceptor key:Connection value:Keep-Alive
  5. BridgeInterceptor: BridgeInterceptor key:Host value:www.kuaidi100.com
  6. ConnectionInterceptor: interceptor: ConnectionInterceptor
  7. ConnectionInterceptor: ConnectionInterceptor: 新建连接
  8. CallServerInterceptor: interceptor: CallServerInterceptor
  9. CallServerInterceptor: CallServerInterceptor: 得到响应行：HTTP/1.1 200 OK
  10. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:Server value:nginx
  11. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:Cache-Control value:no-cache
  12. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:Connection value:keep-alive
  13. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:Vary value:Accept-Encoding
  14. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:Content-Length value:204
  15. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:P3P value:CP="IDC DSP COR ADM DEVi TAIi PSA PSD IVAi IVDi CONi HIS OUR IND CNT"
  16. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:Date value:Mon, 19 Aug 2019 11:25:24 GMT
  17. CallServerInterceptor: CallServerInterceptor: 得到响应头 key:Content-Type value:text/html;charset=UTF-8
  18. ConnectionInterceptor: ConnectionInterceptor: 得到服务器响应：isKeepAlive=true,保持长连接，将此连接加入到连接池中
  19.  onResponse: {"message":"ok","nu":"222222222","ischeck":"1","com":"yuantong","status":"200","condition":"F00","state":"3","data":[{"time":"2019-08-09 19:25:24","context":"查无结果","ftime":"2019-08-09 19:25:24"}]}

我们在请求GET方法,看下连接是否从连接池中获取.从下面的log可以看到我们从连接池中拿到了可复用的连接,测试成功.

  1. RetryInterceptor: interceptor: RetryInterceptor
  2. BridgeInterceptor: interceptor: BridgeInterceptor
  3. BridgeInterceptor: BridgeInterceptor: 设置的请求头
  4. BridgeInterceptor: BridgeInterceptor key:Connection value:Keep-Alive
  5. BridgeInterceptor: BridgeInterceptor key:Host value:www.kuaidi100.com
  6. ConnectionInterceptor: interceptor: ConnectionInterceptor
  7. ConnectionInterceptor: ConnectionInterceptor: 从连接池中获取连接

OK,我们从头到尾梳理了OKhttp的实现思路,当然我只是以最简单的方式实现了,OKhttp的还有好多细节,比如缓存拦截器,这个拦截器比较复杂但是思路简单:将响应的数据存储到了LruCache中等逻辑大家可以自己分析.