Fork me on GitHub

Retrofit与OkHttp学习

发表于 | 分类于
字数统计: 3,221 字 | 阅读时长 ≈ 17 分钟

Android网络请求框架okHttp源码分析

OkHttp

偷贴一张Piasy的图

基本用法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
OkHttpClient client=new OkHttpClient.Builder().build();
Request request = new Request.Builder().url("xxx.com").build();

//同步
Response syncRep = client.newCall(request).execute();
//异步
client.newCall(request).enqueue(new Callback() {
    @Override
    public void onFailure(Call call, IOException e) {

    }

    @Override
    public void onResponse(Call call, Response response) throws IOException {

    }
});

同步请求

先从同步请求入手，newCall返回的是RealCall实例，RealCall的execute

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
 @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) {
        // 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);
    }
  }
}

主要还是调用了getResponseWithInterceptorChain，最后在通知dispatcher结束。其他的都是取消、错误异常处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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) {
    interceptors.addAll(client.networkInterceptors());
  }
  interceptors.add(new CallServerInterceptor(forWebSocket));

  Interceptor.Chain chain = new RealInterceptorChain(
      interceptors, null, null, null, 0, originalRequest);
  return chain.proceed(originalRequest);
}

为请求添加拦截器（下面会讲），先添加自定义的拦截器，然后添加一堆默认的桥接、缓存等拦截器，最后添加的是CallServerInterceptor，划重点，要考的

最后调用RealInterceptorChain实例的proceed

1
2
3
4
5
6
7
8
9
10
11
12
13
public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
      RealConnection connection) throws IOException {

    calls++;

    // Call the next interceptor in the chain.
    RealInterceptorChain next = new RealInterceptorChain(
        interceptors, streamAllocation, httpCodec, connection, index + 1, request);
    Interceptor interceptor = interceptors.get(index);
    Response response = interceptor.intercept(next);

    return response;
  }

第一次传进来的index为0，后面每次都是index+1，就是依次按上面interceptors的添加顺序，把拦截器取出来，调用intercept的方法。在拦截器内部又调用proceed，一直到最后来调用到上面说的最后添加的CallServerInterceptor这个拦截器来进行真正的网络操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@Override 
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();
    httpCodec.writeRequestHeaders(request);

    Response.Builder responseBuilder = null;
  //对Request的请求头"Expect: 100-continue"做个处理
  if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
      if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
        httpCodec.flushRequest();
        responseBuilder = httpCodec.readResponseHeaders(true);
      }

      if (responseBuilder == null) {
        // Write the request body if the "Expect: 100-continue" expectation was met.
        Sink requestBodyOut = httpCodec.createRequestBody(request, request.body().contentLength());
        BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
        request.body().writeTo(bufferedRequestBody);
        bufferedRequestBody.close();
      } else if (!connection.isMultiplexed()) {
        // If the "Expect: 100-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) {
      responseBuilder = httpCodec.readResponseHeaders(false);
    }

    Response response = responseBuilder
        .request(request)
        .handshake(streamAllocation.connection().handshake())
        .sentRequestAtMillis(sentRequestMillis)
        .receivedResponseAtMillis(System.currentTimeMillis())
        .build();

    int code = response.code();
    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()
          .body(httpCodec.openResponseBody(response))
          .build();
    }


    return response;
  }

实际上使用的是HttpCodec类来进行socket操作，基于Okio封装了BufferSink和BufferSource。

拦截器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public interface Interceptor {
  Response intercept(Chain chain) throws IOException;

  interface Chain {
    Request request();

    Response proceed(Request request) throws IOException;

    /**
     * Returns the connection the request will be executed on. This is only available in the chains
     * of network interceptors; for application interceptors this is always null.
     */
    @Nullable Connection connection();
  }
}

Intercept只有一个内部类Chain，一个intercept方法，参数为Chain的实例。

通过这个chain.request和chain.proceed来分别对Request和Response做额外的处理。

下面是一个自定义的拦截器，输出请求/响应的日志

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class LoggingInterceptor implements Interceptor {
  @Override public Response intercept(Interceptor.Chain chain) throws IOException {
    Request request = chain.request();
    
    long t1 = System.nanoTime();
    logger.info(String.format("Sending request %s on %s%n%s",
        request.url(), chain.connection(), request.headers()));

    Response response = chain.proceed(request);

    long t2 = System.nanoTime();
    logger.info(String.format("Received response for %s in %.1fms%n%s",
        response.request().url(), (t2 - t1) / 1e6d, response.headers()));

    return response;
  }
}

异步请求

下面再来看看异步请求，重新贴一下代码

1
2
3
4
5
6
7
8
9
10
11
client.newCall(request).enqueue(new Callback() {
    @Override
    public void onFailure(Call call, IOException e) {

    }

    @Override
    public void onResponse(Call call, Response response) throws IOException {

    }
});

RealCall的enqueue

1
2
3
4
5
6
7
8
@Override public void enqueue(Callback responseCallback) {
  synchronized (this) {
    if (executed) throw new IllegalStateException("Already Executed");
    executed = true;
  }
  captureCallStackTrace();
  client.dispatcher().enqueue(new AsyncCall(responseCallback));
}

主要是使用AsyncCall包装一下回调请求，然后通过client.dispatcher()的enquue来执行

1
2
3
4
5
6
7
8
synchronized void enqueue(AsyncCall call) {
  if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
    runningAsyncCalls.add(call);
    executorService().execute(call);
  } else {
    readyAsyncCalls.add(call);
  }
}

如果没有超过最大请求数，就使用executorService()返回的线程池实例执行call，否则添加到等待的readyAsyncCalls中。

再来介绍一下AsyncCall，是RealCall的一个内部类，继承自NamedRunnable，NamedRunnable实现Runnable。NamedRunnable在重写run方法的时执行了一个抽象方法execute，由AsyncCall来实现。所以当线程执行异步请求的时候实际上走的就是AsyncCall的execute方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
 @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) {
...
   } finally {
     client.dispatcher().finished(this);
   }
 }

跟RealCall的execute很像啊，getResponseWithInterceptorChain来执行。基本可以说明异步的请求跟同步的区别就是加了个线程池来实现网络请求，并通过callback回调。

Retrofit

再来分析一下Retrofit是如何对OkHttp进行封装的

基本用法

这块暂时不看，看下一个

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public static RetrofitClient getInstance() {
    if (mInstance == null) {
        synchronized (RetrofitClient.class) {
            if (mInstance == null) {
                mInstance = new RetrofitClient();
            }
        }
    }
    return mInstance;
}
private RetrofitClient() {
    mHttpClient = new OkHttpClient.Builder()
            .addInterceptor(new LoggingInterceptor())
            .connectTimeout(10, TimeUnit.SECONDS)
            .build();
    mRetrofit = new Retrofit.Builder()
            .baseUrl(BASE_URL)
            .addConverterFactory(CustomConverterFactory.create())
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .client(mHttpClient)
            .build();

    mApi = mRetrofit.create(ApiService.class);
}

↓↓看这个↓↓

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public interface BlogService {
    @GET("blog/{id}")
    Call<ResponseBody> getBlog(@Path("id") int id);
}


Retrofit retrofit = new Retrofit.Builder()
    .baseUrl("http://localhost:4567/")
    .build();
BlogService service = retrofit.create(BlogService.class);
Call<ResponseBody> call = service.getBlog(2);
call.enqueue(new Callback<ResponseBody>() {
  @Override
  public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {

  }

  @Override
  public void onFailure(Call<ResponseBody> call, Throwable t) {

  }
});

create

1
2
3
4
5
6
7
8
9
10
11
public <T> T create(final Class<T> service) {
  return (T) Proxy.newProxyInstance(
        service.getClassLoader()
        , new Class<?>[] { service }
        , new InvocationHandler() {
            @Override public Object invoke(Object proxy, Method method, @Nullable Object[] args)
                throws Throwable {
  				...
            }
      });
}

通过动态代理技术直接返回代理类实例

动态代理

通过jdk在内存中动态生成一个代理对象，涉及Proxy类的静态方法newProxyInstance和InvocationHandler接口。

下面大体讲一下使用，直接手打可能有点错误，意思到了就行了。。。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
//接口
interface UserDao{
  String getName();
  int getAge();
}
//委托类
class UserDaoImp{
  String getName(){
    return "lewis";
  }
  int getAge(){
    return 22;
  }
}
//动态代理类
public class MyInvocationHandler implements InvocationHandler {  
    private Object target;  
    MyInvocationHandler(Object target) {  
        super();  
        this.target = target;  //在这里获取到委托类的实例
    }  
     
    @Override  
    public Object invoke(Object o, Method method, Object[] args) throws Throwable {  
            System.out.println("当前invoke的方法为：" + method.getName() );  
            return method.invoke(target, args); 
    }  
}
//使用
public static void main(String[] args) {  
 	//实例化委托类传入代理类
  	UserDao userImp = new UserDaoImp();
  	UserDao userProxy = (UserDao)Proxy.newProxyInstance(
      			userImp.getClass().getClassLoader()
      			,userImp.getClass().getInterfaces()
      			,new MyInvocationHandler(userImp)
      			);       
   System.out.println(userProxy.getName()); 
}  
//输出
当前invoke的方法为：getName
lewis

再回到create中InvocationHandler的invoke方法的代码

1
2
3
4
5
6
 @Override 
 public Object invoke(Object proxy, Method method, @Nullable Object[] args) throws Throwable {
    ServiceMethod<Object, Object> serviceMethod = (ServiceMethod<Object, Object>) loadServiceMethod(method);
    OkHttpCall<Object> okHttpCall = new OkHttpCall<>(serviceMethod, args);
    return serviceMethod.callAdapter.adapt(okHttpCall);      
}

loadServiceMethod

1
2
3
4
5
6
7
8
9
10
11
12
13
14
ServiceMethod<?, ?> loadServiceMethod(Method method) {
  ServiceMethod<?, ?> result = serviceMethodCache.get(method);
  if (result != null) return result;

  synchronized (serviceMethodCache) {
    result = serviceMethodCache.get(method);
    if (result == null) {
      //为每个被动态代理的Method创建一个ServiceMethod实例 缓存起来
      result = new ServiceMethod.Builder<>(this, method).build();
      serviceMethodCache.put(method, result);
    }
  }
  return result;
}

先从缓存中查找，如果没有就通过new ServiceMethod.Builder<>(this, method).build()新建一个

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public ServiceMethod build() {
  		//从retrofit.adapterFactories中找到对应的callAdapter给这个serviceMethod
  		//adapterFactories.add(platform.defaultCallAdapterFactory(callbackExecutor));
      callAdapter = createCallAdapter();
  	 	//根据callAdapter中的responseType 
  		//从retrofit.responseBodyConverter中找到合适的convert
      //converterFactories.add(new BuiltInConverters());
      responseConverter = createResponseConverter();

      for (Annotation annotation : methodAnnotations) {
        //解析注解
        parseMethodAnnotation(annotation);
      }

      return new ServiceMethod<>(this);
    }

主要逻辑如上

  • createCallAdapter是根据委托类方法的返回值，来确定要使用的CallAdapter，比如我们的返回值是Observable，那么就会使用RxJava2CallAdapterFactory，如果我们在创建Retrofit实例的时候没有通过addCallAdapterFactory(RxJava2CallAdapterFactory.create())来添加与返回值相应的处理函数，那么这里就会抛出异常。
  • createResponseConverter也是同样的道理，比如我们使用addConverterFactory添加GsonConverterFactory
  • parseMethodAnnotation解析每个方法的注解GET、POST、PUT等等

new OkHttpCall()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
final class OkHttpCall<T> implements Call<T> {
  private final ServiceMethod<T, ?> serviceMethod;
  private final @Nullable Object[] args;
  @GuardedBy("this")
  private @Nullable okhttp3.Call rawCall;
  
  OkHttpCall(ServiceMethod<T, ?> serviceMethod, @Nullable Object[] args) {
    this.serviceMethod = serviceMethod;
    this.args = args;
  }

  
    private okhttp3.Call createRawCall() throws IOException {
    Request request = serviceMethod.toRequest(args);
    okhttp3.Call call = serviceMethod.callFactory.newCall(request);
    if (call == null) {
      throw new NullPointerException("Call.Factory returned null.");
    }
    return call;
  }
}

对内部okhttp3.Call类型的rawCall进行封装，默认使用的是OkHttpClient的实例。

1
2
3
4
5
serviceMethod.callFactory = = builder.retrofit.callFactory(); //serviceMethod.java
 okhttp3.Call.Factory callFactory = this.callFactory; // Retrofit.Builder
      if (callFactory == null) {
        callFactory = new OkHttpClient();
      }

parseResponse

OkHttpCall内部对返回值做了处理，通过responseConverter

1
2
3
4
T body = serviceMethod.toResponse(catchingBody);
  R toResponse(ResponseBody body) throws IOException {
    return responseConverter.convert(body);
  }

callAdapter.adapt

再关注动态代理方法invoke的返回值

1
return serviceMethod.callAdapter.adapt(okHttpCall);

调用serviceMethod的成员变量callAdapter.adapt来适配委托类的方法返回值。

在loadServiceMethod中创建ServiceMethod时，callAdapter = createCallAdapter()，跟进

1
2
3
4
5
6
7
8
private CallAdapter<T, R> createCallAdapter() {
   try {
     //noinspection unchecked
     return (CallAdapter<T, R>) retrofit.callAdapter(returnType, annotations);
   } catch (RuntimeException e) { // Wide exception range because factories are user code.
     throw methodError(e, "Unable to create call adapter for %s", returnType);
   }
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public CallAdapter<?, ?> callAdapter(Type returnType, Annotation[] annotations) {
    return nextCallAdapter(null, returnType, annotations);
}

final List<CallAdapter.Factory> adapterFactories;

public CallAdapter<?, ?> nextCallAdapter(@Nullable CallAdapter.Factory skipPast, Type returnType,
      Annotation[] annotations) {
 
    int start = adapterFactories.indexOf(skipPast) + 1;
    for (int i = start, count = adapterFactories.size(); i < count; i++) {
      CallAdapter<?, ?> adapter = adapterFactories.get(i).get(returnType, annotations, this);
      if (adapter != null) {
        return adapter;
      }
    }
  }

通过adapterFactories.get(i).get(returnType, annotations, this);可以知道是遍历retrofit的adapterFactories，并调用每个item的get方法来获取与返回值对应的Adapter

adapterFactories

这个adapterFactories是在构造Retrofit时由Builder默认添加了一个

1
2
3
4
5
6
7
  public Retrofit build() {
    ...
     // Make a defensive copy of the adapters and add the default Call adapter.
     List<CallAdapter.Factory> adapterFactories = new ArrayList<>(this.adapterFactories);
     adapterFactories.add(platform.defaultCallAdapterFactory(callbackExecutor));
...
 }

以默认的DefaultCallAdapterFactory为例，看看他的get方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
final class DefaultCallAdapterFactory extends CallAdapter.Factory {
  static final CallAdapter.Factory INSTANCE = new DefaultCallAdapterFactory();

  @Override
  public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
    if (getRawType(returnType) != Call.class) {
      return null;
    }

    final Type responseType = Utils.getCallResponseType(returnType);
    return new CallAdapter<Object, Call<?>>() {
      @Override public Type responseType() {
        return responseType;
      }

      @Override public Call<Object> adapt(Call<Object> call) {
        return call;
      }
    };
  }
}

返回的new CallAdapter对传入的OkHttpCall毫无包装，直接返回Call，毕竟是默认的。

我们再以常用的RxJava2CallAdapterFactory为例，看RxJava2CallAdapterFactory的get方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Override
 public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
   Class<?> rawType = getRawType(returnType);

   if (rawType == Completable.class) {
     // Completable is not parameterized (which is what the rest of this method deals with) so it
     // can only be created with a single configuration.
     return new RxJava2CallAdapter(Void.class, scheduler, isAsync, false, true, false, false,
         false, true);
   }

   return new RxJava2CallAdapter(responseType, scheduler, isAsync, isResult, isBody, isFlowable,
       isSingle, isMaybe, false);
 }

返回的是RxJava2CallAdapter的实例，在看看RxJava2CallAdapter的adapt方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Override public Object adapt(Call<R> call) {
   Observable<Response<R>> responseObservable = isAsync
       ? new CallEnqueueObservable<>(call)
       : new CallExecuteObservable<>(call);

   Observable<?> observable;
   if (isResult) {
     observable = new ResultObservable<>(responseObservable);
   } else if (isBody) {
     observable = new BodyObservable<>(responseObservable);
   } else {
     observable = responseObservable;
   }
...
   return observable;
 }

根据各种条件做了一些判断，对Call进行封装，具体见源码。

call.enqueue

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@Override public void enqueue(final Callback<T> callback) {

  okhttp3.Call call;
  Throwable failure;

  synchronized (this) {
    if (executed) throw new IllegalStateException("Already executed.");
    executed = true;

    call = rawCall;
    failure = creationFailure;
    if (call == null && failure == null) {
      try {
        call = rawCall = createRawCall();
      } catch (Throwable t) {
        failure = creationFailure = t;
      }
    }
  }

	//还是包装的okhttp的东西
  call.enqueue(new okhttp3.Callback() {
    @Override public void onResponse(okhttp3.Call call, okhttp3.Response rawResponse)
        throws IOException {
      Response<T> response;
      try {
        //解析response
        response = parseResponse(rawResponse);
      } catch (Throwable e) {
        callFailure(e);
        return;
      }
      callSuccess(response);
    }

    @Override public void onFailure(okhttp3.Call call, IOException e) {
      try {
        callback.onFailure(OkHttpCall.this, e);
      } catch (Throwable t) {
        t.printStackTrace();
      }
    }

    private void callFailure(Throwable e) {
      try {
        callback.onFailure(OkHttpCall.this, e);
      } catch (Throwable t) {
        t.printStackTrace();
      }
    }

    private void callSuccess(Response<T> response) {
      try {
        callback.onResponse(OkHttpCall.this, response);
      } catch (Throwable t) {
        t.printStackTrace();
      }
    }
  });
}


Response<T> parseResponse(okhttp3.Response rawResponse) throws IOException {
    ResponseBody rawBody = rawResponse.body();
    try {
      T body = serviceMethod.toResponse(catchingBody);
      return Response.success(body, rawResponse);
    } catch (RuntimeException e) {
      throw e;
    }
  }
//这里调用上面创建serviceMethod里创建的ReponseConvert
R toResponse(ResponseBody body) throws IOException {
    return responseConverter.convert(body);
  }

再上一张图来帮助理解create的这个三部曲。