Retrofit源码分析

本文保留早期阅读 Retrofit 时的代码摘录；它们包含不同版本的实现，适合对照思路阅读。新增的设计模式分析统一依据 Retrofit 2.3.0（parent-2.3.0，提交 903e149a4f211a0fb7e01ed4fdf8c48a8b6c7891），重点说明对象之间的职责与协作，不将这些类图套用到其他版本。

基本流程：Retrofit.create() 创建服务接口的动态代理；调用接口方法时解析或复用 ServiceMethod，创建 OkHttpCall，再由 CallAdapter.adapt() 得到声明的返回类型。返回 Call 时由调用者执行 execute() 或 enqueue()；RxJava 适配器通常在订阅时才执行请求。响应实体转换发生在 OkHttpCall.parseResponse() 经 ServiceMethod.toResponse() 调用 Converter.convert() 的阶段。

下面的历史摘录保持原有章节顺序；模式总览与 UML位于文末。

ServiceMethod

ServiceMethod主要用来解析我们声明的方法。 1.Builder构造函数中获取方法上的注解和注解里的值。 2.调用createCallAdapter方法获取CallAdapter。 3.调用createResponseConverter方法获取Converter。 4.调用parseMethodAnnotation来解析方法上的注解 5.调用parseParameterAnnotation来解析参数上的注解，并将这些注解包含的信息封装成一个ParameterHandler对象。 6.在OkHttpCall中调用toRequest的方法来获取请求的参数。

public Builder(Retrofit retrofit, Method method) {
  this.retrofit = retrofit;
  this.method = method;
  this.methodAnnotations = method.getAnnotations();//获取方法上的注解
  this.parameterTypes = method.getGenericParameterTypes();//获取参数类型
  this.parameterAnnotationsArray = method.getParameterAnnotations();//获取参数上的注解  这里是一个二维数组，每个参数可能有多个注解
}

createCallAdapter方法获取方法的返回值和注解，根据注解和返回值获取对应的CallAdapter。

private CallAdapter<T, R> createCallAdapter() {
  Type returnType = method.getGenericReturnType();//获取返回值
  //校验返回值类型
  if (Utils.hasUnresolvableType(returnType)) {
    throw methodError(
        "Method return type must not include a type variable or wildcard: %s", returnType);
  }
  if (returnType == void.class) {
    throw methodError("Service methods cannot return void.");
  }
  Annotation[] annotations = method.getAnnotations();
  try {
    //noinspection unchecked
    //调用Retrofit的 callAdapter方法来获取CallAdapter
    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);
  }
}

Retrofit的callAdapter方法

public CallAdapter<?, ?> callAdapter(Type returnType, Annotation[] annotations) {
return nextCallAdapter(null, returnType, annotations);
}
public CallAdapter<?, ?> nextCallAdapter(CallAdapter.Factory skipPast, Type returnType,
  Annotation[] annotations) {
checkNotNull(returnType, "returnType == null");
checkNotNull(annotations, "annotations == null");
int start = adapterFactories.indexOf(skipPast) + 1;
for (int i = start, count = adapterFactories.size(); i < count; i++) {
    //调用CallAdapter.Factory的get方法来获取对应的CallAdapter
    //这里看出是根据返回值类型来获取不同的CallAdapter
    //
  CallAdapter<?,?> adapter = adapterFactories.get(i).get(returnType, annotations, this);
  if (adapter != null) {
    return adapter;
  }
}
StringBuilder builder = new StringBuilder("Could not locate call adapter for ")
    .append(returnType)
    .append(".\n");
if (skipPast != null) {
  builder.append("  Skipped:");
  for (int i = 0; i < start; i++) {
    builder.append("\n   * ").append(adapterFactories.get(i).getClass().getName());
  }
  builder.append('\n');
}
builder.append("  Tried:");
for (int i = start, count = adapterFactories.size(); i < count; i++) {
  builder.append("\n   * ").append(adapterFactories.get(i).getClass().getName());
}
throw new IllegalArgumentException(builder.toString());
}

public ServiceMethod build() {
  callAdapter = createCallAdapter();//创建CallAdapter
  responseType = callAdapter.responseType();//获取返回值类型
  if (responseType == Response.class || responseType == okhttp3.Response.class) {
    throw methodError("'"
        + Utils.getRawType(responseType).getName()
        + "' is not a valid response body type. Did you mean ResponseBody?");
  }
  responseConverter = createResponseConverter();//创建ResponseConverter

  for (Annotation annotation : methodAnnotations) {
    parseMethodAnnotation(annotation);//遍历解析方法注解
  }

  if (httpMethod == null) {
    throw methodError("HTTP method annotation is required (e.g., @GET, @POST, etc.).");
  }

  if (!hasBody) {
    if (isMultipart) {
      throw methodError(
          "Multipart can only be specified on HTTP methods with request body (e.g., @POST).");
    }
    if (isFormEncoded) {
      throw methodError("FormUrlEncoded can only be specified on HTTP methods with "
          + "request body (e.g., @POST).");
    }
  }

  int parameterCount = parameterAnnotationsArray.length;//获取参数注解长度
  parameterHandlers = new ParameterHandler<?>[parameterCount];//创建参数处理的数组
  //遍历每个参数
  for (int p = 0; p < parameterCount; p++) {
    Type parameterType = parameterTypes[p];
    //校验每个参数的类型
    if (Utils.hasUnresolvableType(parameterType)) {
      throw parameterError(p, "Parameter type must not include a type variable or wildcard: %s",
          parameterType);
    }
    //单个参数上的注解
    Annotation[] parameterAnnotations = parameterAnnotationsArray[p];
    if (parameterAnnotations == null) {
      throw parameterError(p, "No Retrofit annotation found.");
    }

    parameterHandlers[p] = parseParameter(p, parameterType, parameterAnnotations);
  }

  if (relativeUrl == null && !gotUrl) {
    throw methodError("Missing either @%s URL or @Url parameter.", httpMethod);
  }
  if (!isFormEncoded && !isMultipart && !hasBody && gotBody) {
    throw methodError("Non-body HTTP method cannot contain @Body.");
  }
  if (isFormEncoded && !gotField) {
    throw methodError("Form-encoded method must contain at least one @Field.");
  }
  if (isMultipart && !gotPart) {
    throw methodError("Multipart method must contain at least one @Part.");
  }

  return new ServiceMethod<>(this);
}

private void parseMethodAnnotation(Annotation annotation) {
  if (annotation instanceof DELETE) {
    parseHttpMethodAndPath("DELETE", ((DELETE) annotation).value(), false);
  } else if (annotation instanceof GET) {
    parseHttpMethodAndPath("GET", ((GET) annotation).value(), false);
  } else if (annotation instanceof HEAD) {
    parseHttpMethodAndPath("HEAD", ((HEAD) annotation).value(), false);
    if (!Void.class.equals(responseType)) {
      throw methodError("HEAD method must use Void as response type.");
    }
  } else if (annotation instanceof PATCH) {
    parseHttpMethodAndPath("PATCH", ((PATCH) annotation).value(), true);
  } else if (annotation instanceof POST) {
    parseHttpMethodAndPath("POST", ((POST) annotation).value(), true);
  } else if (annotation instanceof PUT) {
    parseHttpMethodAndPath("PUT", ((PUT) annotation).value(), true);
  } else if (annotation instanceof OPTIONS) {
    parseHttpMethodAndPath("OPTIONS", ((OPTIONS) annotation).value(), false);
  } else if (annotation instanceof HTTP) {
    HTTP http = (HTTP) annotation;
    parseHttpMethodAndPath(http.method(), http.path(), http.hasBody());
  } else if (annotation instanceof retrofit2.http.Headers) {
    String[] headersToParse = ((retrofit2.http.Headers) annotation).value();
    if (headersToParse.length == 0) {
      throw methodError("@Headers annotation is empty.");
    }
    headers = parseHeaders(headersToParse);
  } else if (annotation instanceof Multipart) {
    if (isFormEncoded) {
      throw methodError("Only one encoding annotation is allowed.");
    }
    isMultipart = true;
  } else if (annotation instanceof FormUrlEncoded) {
    if (isMultipart) {
      throw methodError("Only one encoding annotation is allowed.");
    }
    isFormEncoded = true;
  }
}

ParameterHandler最主要的方法就是apply方法，该方法的作用是为build添加参数。

abstract void apply(RequestBuilder builder, T value) throws IOException;

  final ParameterHandler<Iterable<T>> iterable() {
    return new ParameterHandler<Iterable<T>>() {
      @Override void apply(RequestBuilder builder, Iterable<T> values) throws IOException {
        if (values == null) return; // Skip null values.

        for (T value : values) {
          ParameterHandler.this.apply(builder, value);
        }
      }
    };
  }

  final ParameterHandler<Object> array() {
    return new ParameterHandler<Object>() {
      @Override void apply(RequestBuilder builder, Object values) throws IOException {
        if (values == null) return; // Skip null values.

        for (int i = 0, size = Array.getLength(values); i < size; i++) {
          //noinspection unchecked
          ParameterHandler.this.apply(builder, (T) Array.get(values, i));
        }
      }
    };
  }

ServiceMethod的toRequest先创建一个RequestBuilder，然后遍历参数，依次向build中添加参数。

 /** Builds an HTTP request from method arguments. */
  Request toRequest(Object... args) throws IOException {
    RequestBuilder requestBuilder = new RequestBuilder(httpMethod, baseUrl, relativeUrl, headers,
        contentType, hasBody, isFormEncoded, isMultipart);

    @SuppressWarnings("unchecked") // It is an error to invoke a method with the wrong arg types.
    ParameterHandler<Object>[] handlers = (ParameterHandler<Object>[]) parameterHandlers;

    int argumentCount = args != null ? args.length : 0;
    if (argumentCount != handlers.length) {
      throw new IllegalArgumentException("Argument count (" + argumentCount
          + ") doesn't match expected count (" + handlers.length + ")");
    }

    for (int p = 0; p < argumentCount; p++) {
      handlers[p].apply(requestBuilder, args[p]);
    }

    return requestBuilder.build();
  }

  /** Builds a method return value from an HTTP response body. */
  R toResponse(ResponseBody body) throws IOException {
    return responseConverter.convert(body);
  }

RequestBuilder用来构建请求的Request。

final class RequestBuilder {
  private static final char[] HEX_DIGITS =
      { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F' };
  private static final String PATH_SEGMENT_ALWAYS_ENCODE_SET = " \"<>^`{}|\\?#";

  private final String method;

  private final HttpUrl baseUrl;
  private String relativeUrl;
  private HttpUrl.Builder urlBuilder;

  private final Request.Builder requestBuilder;
  private MediaType contentType;

  private final boolean hasBody;
  private MultipartBody.Builder multipartBuilder;
  private FormBody.Builder formBuilder;
  private RequestBody body;

  RequestBuilder(String method, HttpUrl baseUrl, String relativeUrl, Headers headers,
      MediaType contentType, boolean hasBody, boolean isFormEncoded, boolean isMultipart) {
    this.method = method;
    this.baseUrl = baseUrl;
    this.relativeUrl = relativeUrl;
    this.requestBuilder = new Request.Builder();
    this.contentType = contentType;
    this.hasBody = hasBody;

    if (headers != null) {
      requestBuilder.headers(headers);//添加header
    }

    if (isFormEncoded) {//form请求
      // Will be set to 'body' in 'build'.
      formBuilder = new FormBody.Builder();
    } else if (isMultipart) {
      // Will be set to 'body' in 'build'.
      multipartBuilder = new MultipartBody.Builder();
      multipartBuilder.setType(MultipartBody.FORM);
    }
  }

  void setRelativeUrl(Object relativeUrl) {
    if (relativeUrl == null) throw new NullPointerException("@Url parameter is null.");
    this.relativeUrl = relativeUrl.toString();
  }

  void addHeader(String name, String value) {
    if ("Content-Type".equalsIgnoreCase(name)) {
      MediaType type = MediaType.parse(value);
      if (type == null) {
        throw new IllegalArgumentException("Malformed content type: " + value);
      }
      contentType = type;
    } else {
      requestBuilder.addHeader(name, value);
    }
  }

  void addPathParam(String name, String value, boolean encoded) {
    if (relativeUrl == null) {
      // The relative URL is cleared when the first query parameter is set.
      throw new AssertionError();
    }
    relativeUrl = relativeUrl.replace("{" + name + "}", canonicalizeForPath(value, encoded));
  }

  private static String canonicalizeForPath(String input, boolean alreadyEncoded) {
    int codePoint;
    for (int i = 0, limit = input.length(); i < limit; i += Character.charCount(codePoint)) {
      codePoint = input.codePointAt(i);
      if (codePoint < 0x20 || codePoint >= 0x7f
          || PATH_SEGMENT_ALWAYS_ENCODE_SET.indexOf(codePoint) != -1
          || (!alreadyEncoded && (codePoint == '/' || codePoint == '%'))) {
        // Slow path: the character at i requires encoding!
        Buffer out = new Buffer();
        out.writeUtf8(input, 0, i);
        canonicalizeForPath(out, input, i, limit, alreadyEncoded);
        return out.readUtf8();
      }
    }

    // Fast path: no characters required encoding.
    return input;
  }

  private static void canonicalizeForPath(Buffer out, String input, int pos, int limit,
      boolean alreadyEncoded) {
    Buffer utf8Buffer = null; // Lazily allocated.
    int codePoint;
    for (int i = pos; i < limit; i += Character.charCount(codePoint)) {
      codePoint = input.codePointAt(i);
      if (alreadyEncoded
          && (codePoint == '\t' || codePoint == '\n' || codePoint == '\f' || codePoint == '\r')) {
        // Skip this character.
      } else if (codePoint < 0x20 || codePoint >= 0x7f
          || PATH_SEGMENT_ALWAYS_ENCODE_SET.indexOf(codePoint) != -1
          || (!alreadyEncoded && (codePoint == '/' || codePoint == '%'))) {
        // Percent encode this character.
        if (utf8Buffer == null) {
          utf8Buffer = new Buffer();
        }
        utf8Buffer.writeUtf8CodePoint(codePoint);
        while (!utf8Buffer.exhausted()) {
          int b = utf8Buffer.readByte() & 0xff;
          out.writeByte('%');
          out.writeByte(HEX_DIGITS[(b >> 4) & 0xf]);
          out.writeByte(HEX_DIGITS[b & 0xf]);
        }
      } else {
        // This character doesn't need encoding. Just copy it over.
        out.writeUtf8CodePoint(codePoint);
      }
    }
  }
//添加get请求参数
  void addQueryParam(String name, String value, boolean encoded) {
    if (relativeUrl != null) {
      // Do a one-time combination of the built relative URL and the base URL.
      urlBuilder = baseUrl.newBuilder(relativeUrl);
      if (urlBuilder == null) {
        throw new IllegalArgumentException(
            "Malformed URL. Base: " + baseUrl + ", Relative: " + relativeUrl);
      }
      relativeUrl = null;
    }
    if (encoded) {
      urlBuilder.addEncodedQueryParameter(name, value);
    } else {
      urlBuilder.addQueryParameter(name, value);
    }
  }
//添加form表单请求参数
  void addFormField(String name, String value, boolean encoded) {
    if (encoded) {
      formBuilder.addEncoded(name, value);
    } else {
      formBuilder.add(name, value);
    }
  }

  void addPart(Headers headers, RequestBody body) {
    multipartBuilder.addPart(headers, body);
  }

  void addPart(MultipartBody.Part part) {
    multipartBuilder.addPart(part);
  }

  void setBody(RequestBody body) {
    this.body = body;
  }

  Request build() {
    HttpUrl url;
    HttpUrl.Builder urlBuilder = this.urlBuilder;
    if (urlBuilder != null) {
      url = urlBuilder.build();
    } else {
      // No query parameters triggered builder creation, just combine the relative URL and base URL.
      url = baseUrl.resolve(relativeUrl);
      if (url == null) {
        throw new IllegalArgumentException(
            "Malformed URL. Base: " + baseUrl + ", Relative: " + relativeUrl);
      }
    }

    RequestBody body = this.body;
    if (body == null) {
      // Try to pull from one of the builders.
      if (formBuilder != null) {
        body = formBuilder.build();
      } else if (multipartBuilder != null) {
        body = multipartBuilder.build();
      } else if (hasBody) {
        // Body is absent, make an empty body.
        body = RequestBody.create(null, new byte[0]);
      }
    }

    MediaType contentType = this.contentType;
    if (contentType != null) {
      if (body != null) {
        body = new ContentTypeOverridingRequestBody(body, contentType);
      } else {
        requestBuilder.addHeader("Content-Type", contentType.toString());
      }
    }

    return requestBuilder
        .url(url)
        .method(method, body)
        .build();
  }

  private static class ContentTypeOverridingRequestBody extends RequestBody {
    private final RequestBody delegate;
    private final MediaType contentType;

    ContentTypeOverridingRequestBody(RequestBody delegate, MediaType contentType) {
      this.delegate = delegate;
      this.contentType = contentType;
    }

    @Override public MediaType contentType() {
      return contentType;
    }

    @Override public long contentLength() throws IOException {
      return delegate.contentLength();
    }

    @Override public void writeTo(BufferedSink sink) throws IOException {
      delegate.writeTo(sink);
    }
  }
}

OkHttpCall

OkHttpCall算是OkHttp的包装类，用它跟OkHttp对接，所有OkHttp需要的参数都可以看这个类。当然也还是可以扩展一个新的Call的，比如HttpUrlConnectionCall。

@Override public Response<T> execute() throws IOException {
okhttp3.Call call;
synchronized (this) {
  if (executed) throw new IllegalStateException("Already executed.");
  executed = true;
  if (creationFailure != null) {
    if (creationFailure instanceof IOException) {
      throw (IOException) creationFailure;
    } else {
      throw (RuntimeException) creationFailure;
    }
  }
  call = rawCall;
  if (call == null) {
    try {
      call = rawCall = createRawCall();
    } catch (IOException | RuntimeException e) {
      creationFailure = e;
      throw e;
    }
  }
}
if (canceled) {
  call.cancel();
}
return parseResponse(call.execute());
}

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

//解析返回值
Response<T> parseResponse(okhttp3.Response rawResponse) throws IOException {
ResponseBody rawBody = rawResponse.body();
// Remove the body's source (the only stateful object) so we can pass the response along.
rawResponse = rawResponse.newBuilder()
    .body(new NoContentResponseBody(rawBody.contentType(), rawBody.contentLength()))
    .build();

int code = rawResponse.code();
if (code < 200 || code >= 300) {
  try {
    // Buffer the entire body to avoid future I/O.
    ResponseBody bufferedBody = Utils.buffer(rawBody);
    return Response.error(bufferedBody, rawResponse);
  } finally {
    rawBody.close();
  }
}
if (code == 204 || code == 205) {
  return Response.success(null, rawResponse);
}
ExceptionCatchingRequestBody catchingBody = new ExceptionCatchingRequestBody(rawBody);
try {
  T body = serviceMethod.toResponse(catchingBody);//将body转化为实体
  return Response.success(body, rawResponse);
} catch (RuntimeException e) {
  // If the underlying source threw an exception, propagate that rather than indicating it was
  // a runtime exception.
  catchingBody.throwIfCaught();
  throw e;
}
}

 Response<T> parseResponse(okhttp3.Response rawResponse) throws IOException {
    ResponseBody rawBody = rawResponse.body();

    // Remove the body's source (the only stateful object) so we can pass the response along.
    rawResponse = rawResponse.newBuilder()
        .body(new NoContentResponseBody(rawBody.contentType(), rawBody.contentLength()))
        .build();

    int code = rawResponse.code();
    if (code < 200 || code >= 300) {
      try {
        // Buffer the entire body to avoid future I/O.
        ResponseBody bufferedBody = Utils.buffer(rawBody);
        return Response.error(bufferedBody, rawResponse);
      } finally {
        rawBody.close();
      }
    }

    if (code == 204 || code == 205) {
      rawBody.close();
      return Response.success(null, rawResponse);
    }

    ExceptionCatchingRequestBody catchingBody = new ExceptionCatchingRequestBody(rawBody);
    try {
      T body = serviceMethod.toResponse(catchingBody);//将ResponseBody转换为实体
      return Response.success(body, rawResponse);
    } catch (RuntimeException e) {
      // If the underlying source threw an exception, propagate that rather than indicating it was
      // a runtime exception.
      catchingBody.throwIfCaught();
      throw e;
    }
  }

CallAdapter

RxJavaCallAdapterFactory

Single类和Observable对象类似，只不过Single只能发送单个值，并且Single只能发出一个错误或成功的值，不能发送onComplete通知。

Single的just方法。

public static <T> Single<T> just(final T value) {
    return ScalarSynchronousSingle.create(value);
}

Single.create(new Single.OnSubscribe<Object>() {
    @Override
    public void call(SingleSubscriber<? super Object> singleSubscriber) {
      //SingleSubscriber没有onComplete方法
    }
});

Completable也和Observable类似，但是Completable不发送值，值发送错误和完成。

Completable.create(new Completable.CompletableOnSubscribe() {
    @Override
    public void call(Completable.CompletableSubscriber completableSubscriber) {
        //CompletableSubscriber 没有onNext方法
    }
});

当我们的请求不需要处理返回值的时候我们可以让该请求返回Completable。

public final class RxJavaCallAdapterFactory extends CallAdapter.Factory {
  /**
   * Returns an instance which creates synchronous observables that do not operate on any scheduler
   * by default.
   */
  public static RxJavaCallAdapterFactory create() {
    return new RxJavaCallAdapterFactory(null);
  }

  /**
   * Returns an instance which creates synchronous observables that
   * {@linkplain Observable#subscribeOn(Scheduler) subscribe on} {@code scheduler} by default.
   */
  public static RxJavaCallAdapterFactory createWithScheduler(Scheduler scheduler) {
    if (scheduler == null) throw new NullPointerException("scheduler == null");
    return new RxJavaCallAdapterFactory(scheduler);
  }

  private final Scheduler scheduler;

  private RxJavaCallAdapterFactory(Scheduler scheduler) {
    this.scheduler = scheduler;
  }

  @Override
  public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
    Class<?> rawType = getRawType(returnType);
    boolean isSingle = rawType == Single.class;//判断返回值是否是Single.class
    boolean isCompletable = "rx.Completable".equals(rawType.getCanonicalName());
    if (rawType != Observable.class && !isSingle && !isCompletable) {
      return null;
    }

    if (isCompletable) {
      return new RxJavaCallAdapter(Void.class, scheduler, false, true, false, true);
    }

    boolean isResult = false;
    boolean isBody = false;
    Type responseType;
    if (!(returnType instanceof ParameterizedType)) {
      String name = isSingle ? "Single" : "Observable";
      throw new IllegalStateException(name + " return type must be parameterized"
          + " as " + name + "<Foo> or " + name + "<? extends Foo>");
    }

    Type observableType = getParameterUpperBound(0, (ParameterizedType) returnType);
    Class<?> rawObservableType = getRawType(observableType);
    if (rawObservableType == Response.class) {
      if (!(observableType instanceof ParameterizedType)) {
        throw new IllegalStateException("Response must be parameterized"
            + " as Response<Foo> or Response<? extends Foo>");
      }
      responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
    } else if (rawObservableType == Result.class) {
      if (!(observableType instanceof ParameterizedType)) {
        throw new IllegalStateException("Result must be parameterized"
            + " as Result<Foo> or Result<? extends Foo>");
      }
      responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
      isResult = true;
    } else {
      responseType = observableType;
      isBody = true;
    }

    return new RxJavaCallAdapter(responseType, scheduler, isResult, isBody, isSingle, false);
  }
}

RxJavaCallAdapter的adapt方法

 @Override public Object adapt(Call<R> call) {
    OnSubscribe<Response<R>> callFunc = new CallOnSubscribe<>(call);//创建CallOnSubscribe

    //获取OnSubscribe对象
    OnSubscribe<?> func;
    if (isResult) {
      func = new ResultOnSubscribe<>(callFunc);
    } else if (isBody) {
      func = new BodyOnSubscribe<>(callFunc);

    } else {
      func = callFunc;
    }
    Observable<?> observable = Observable.create(func);

    if (scheduler != null) {
      observable = observable.subscribeOn(scheduler);
    }

    if (isSingle) {
      return observable.toSingle();//Observable转换为Single
    }
    if (isCompletable) {
      return CompletableHelper.toCompletable(observable);//Observable转换为Completable
    }
    return observable;
  }

CallOnSubscribe

final class CallOnSubscribe<T> implements OnSubscribe<Response<T>> {
  private final Call<T> originalCall;

  CallOnSubscribe(Call<T> originalCall) {
    this.originalCall = originalCall;
  }

  @Override public void call(Subscriber<? super Response<T>> subscriber) {
    // Since Call is a one-shot type, clone it for each new subscriber.
    Call<T> call = originalCall.clone();//clone 了原来的 call，因为 okhttp3.Call 是只能用一次的，所以每次都是新 clone 一个进行网络请求；
    CallArbiter<T> arbiter = new CallArbiter<>(call, subscriber);
    subscriber.add(arbiter);
    subscriber.setProducer(arbiter);//调用producer的request方法
    Response<T> response;
    try {
      response = call.execute();//调用OkHttpCall的execute()方法返回Response
    } catch (Throwable t) {
      Exceptions.throwIfFatal(t);
      arbiter.emitError(t);//发送错误
      return;
    }
    arbiter.emitResponse(response);//发送response
  }

  static final class CallArbiter<T> extends AtomicInteger implements Subscription, Producer {
    private static final int STATE_WAITING = 0;
    private static final int STATE_REQUESTED = 1;
    private static final int STATE_HAS_RESPONSE = 2;
    private static final int STATE_TERMINATED = 3;

    private final Call<T> call;
    private final Subscriber<? super Response<T>> subscriber;

    private volatile Response<T> response;

    CallArbiter(Call<T> call, Subscriber<? super Response<T>> subscriber) {
      super(STATE_WAITING);

      this.call = call;
      this.subscriber = subscriber;
    }

    @Override public void unsubscribe() {
      call.cancel();
    }

    @Override public boolean isUnsubscribed() {
      return call.isCanceled();
    }

    @Override public void request(long amount) {

      if (amount == 0) {
        return;
      }
      while (true) {
        int state = get();
        switch (state) {
          case STATE_WAITING:
            if (compareAndSet(STATE_WAITING, STATE_REQUESTED)) {
              return;
            }
            break; // State transition failed. Try again.

          case STATE_HAS_RESPONSE:
            if (compareAndSet(STATE_HAS_RESPONSE, STATE_TERMINATED)) {
              deliverResponse(response);
              return;
            }
            break; // State transition failed. Try again.

          case STATE_REQUESTED:
          case STATE_TERMINATED:
            return; // Nothing to do.

          default:
            throw new IllegalStateException("Unknown state: " + state);
        }
      }
    }

    void emitResponse(Response<T> response) {
      while (true) {
        int state = get();//获取当前值
        switch (state) {
          case STATE_WAITING:
            this.response = response;
            if (compareAndSet(STATE_WAITING, STATE_HAS_RESPONSE)) {//如果当前值==STATE_WAITING,设置值为STATE_HAS_RESPONSE
              return;
            }
            break; //状态改变失败重试

          case STATE_REQUESTED:
            if (compareAndSet(STATE_REQUESTED, STATE_TERMINATED)) {
              deliverResponse(response);
              return;
            }
            break; // State transition failed. Try again.

          case STATE_HAS_RESPONSE:
          case STATE_TERMINATED:
            throw new AssertionError();
          default:
            throw new IllegalStateException("Unknown state: " + state);
        }
      }
    }

    private void deliverResponse(Response<T> response) {
      try {
        if (!isUnsubscribed()) {
          subscriber.onNext(response);
        }
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        try {
          subscriber.onError(t);
        } catch (Throwable inner) {
          Exceptions.throwIfFatal(inner);
          CompositeException composite = new CompositeException(t, inner);
          RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
        }
        return;
      }
      try {
        subscriber.onCompleted();//发送完成
      } catch (Throwable t) {
        Exceptions.throwIfFatal(t);
        RxJavaPlugins.getInstance().getErrorHandler().handleError(t);
      }
    }

    void emitError(Throwable t) {
      set(STATE_TERMINATED);

      if (!isUnsubscribed()) {
        try {
          subscriber.onError(t);
        } catch (Throwable inner) {
          Exceptions.throwIfFatal(inner);
          CompositeException composite = new CompositeException(t, inner);
          RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
        }
      }
    }
  }
}

BodyOnSubscribe

final class BodyOnSubscribe<T> implements OnSubscribe<T> {
  private final OnSubscribe<Response<T>> upstream;

  BodyOnSubscribe(OnSubscribe<Response<T>> upstream) {
    this.upstream = upstream;
  }

  @Override public void call(Subscriber<? super T> subscriber) {
    upstream.call(new BodySubscriber<>(subscriber));
  }

  private static class BodySubscriber<R> extends Subscriber<Response<R>> {
    private final Subscriber<? super R> subscriber;
    /** Indicates whether a terminal event has been sent to {@link #subscriber}. */
    private boolean subscriberTerminated;

    BodySubscriber(Subscriber<? super R> subscriber) {
      super(subscriber);
      this.subscriber = subscriber;
    }

    @Override public void onNext(Response<R> response) {
      if (response.isSuccessful()) {
        subscriber.onNext(response.body());
      } else {
        subscriberTerminated = true;
        Throwable t = new HttpException(response);
        try {
          subscriber.onError(t);
        } catch (Throwable inner) {
          Exceptions.throwIfFatal(inner);
          CompositeException composite = new CompositeException(t, inner);
          RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
        }
      }
    }

    @Override public void onError(Throwable throwable) {
      if (!subscriberTerminated) {
        subscriber.onError(throwable);
      } else {
        // This should never happen! onNext handles and forwards errors automatically.
        Throwable broken = new AssertionError(
            "This should never happen! Report as a Retrofit bug with the full stacktrace.");
        //noinspection UnnecessaryInitCause Two-arg AssertionError constructor is 1.7+ only.
        broken.initCause(throwable);
        RxJavaPlugins.getInstance().getErrorHandler().handleError(broken);
      }
    }

    @Override public void onCompleted() {
      if (!subscriberTerminated) {
        subscriber.onCompleted();
      }
    }
  }
}

ResultOnSubscribe

final class ResultOnSubscribe<T> implements OnSubscribe<Result<T>> {
  private final OnSubscribe<Response<T>> upstream;

  ResultOnSubscribe(OnSubscribe<Response<T>> upstream) {
    this.upstream = upstream;
  }

  @Override public void call(Subscriber<? super Result<T>> subscriber) {
    upstream.call(new ResultSubscriber<T>(subscriber));
  }

  private static class ResultSubscriber<R> extends Subscriber<Response<R>> {
    private final Subscriber<? super Result<R>> subscriber;

    ResultSubscriber(Subscriber<? super Result<R>> subscriber) {
      super(subscriber);
      this.subscriber = subscriber;
    }

    @Override public void onNext(Response<R> response) {
      subscriber.onNext(Result.response(response));
    }

    @Override public void onError(Throwable throwable) {
      try {
        subscriber.onNext(Result.<R>error(throwable));
      } catch (Throwable t) {
        try {
          subscriber.onError(t);
        } catch (Throwable inner) {
          Exceptions.throwIfFatal(inner);
          CompositeException composite = new CompositeException(t, inner);
          RxJavaPlugins.getInstance().getErrorHandler().handleError(composite);
        }
        return;
      }
      subscriber.onCompleted();
    }

    @Override public void onCompleted() {
      subscriber.onCompleted();
    }
  }
}

• 拆轮子系列：拆 Retrofit

设计模式分析

本节对照 Retrofit 2.3.0 固定版本源码分析。以下 UML 只保留解释模式所需的类型和方法，省略泛型的完整展开、异常与无关成员；嵌套类以完整类名标注。UserApi、运行时代理和匿名实现使用教学名称，均会说明其来源。

模式	源码中的落点	解决的问题
建造者	Retrofit.Builder	分步收集配置，统一校验并组装 Retrofit。
动态代理	Retrofit.create() 与 InvocationHandler	将声明式接口调用转为请求描述与调用对象。
工厂方法	CallAdapter.Factory.get()	由具体工厂决定适配器实现，扩展返回类型。
适配器	RxJavaCallAdapter	把 Call<R> 转为调用者声明的 RxJava 返回类型。
策略	Converter 与具体转换器	将响应解析算法从请求执行流程中分离。
责任链思想	nextCallAdapter() 等工厂查找循环	按顺序尝试扩展点，找到首个能处理当前类型的工厂。
装饰者	ExecutorCallbackCall	保持 Call 接口，增加异步回调的线程调度。
原型	Call.clone()、OkHttpCall.clone()	根据调用描述创建一个可重新执行的调用实例。

建造者：分步配置 Retrofit

Retrofit 需要基础 URL、网络调用工厂、转换器工厂、适配器工厂和回调执行器。如果全部塞进一个公开构造器，调用者很难看清每个参数的作用。Retrofit.Builder 把收集配置与最终创建分开，让 build() 统一完成校验和默认值补全。

类或对象	模式角色	职责
Retrofit.Builder	具体建造者	保存配置，提供 baseUrl()、addConverterFactory()、addCallAdapterFactory() 等分步操作。
Retrofit	产品	持有创建服务接口和执行请求所需的配置与工厂集合。
Converter.Factory、CallAdapter.Factory	产品所需组件	分别创建数据转换器与调用适配器；由 Builder 收集。
应用初始化代码	构建步骤的组织者	决定配置内容和调用顺序；源码没有另设一个 Director 类。

classDiagram
    direction TB
    class RetrofitBuilder["Retrofit.Builder"] {
        -HttpUrl baseUrl
        +baseUrl(String baseUrl) Builder
        +addConverterFactory(Converter.Factory factory) Builder
        +addCallAdapterFactory(CallAdapter.Factory factory) Builder
        +build() Retrofit
    }
    class Retrofit {
        +create(Class service) Object
    }
    class ConverterFactory["Converter.Factory"] {
        <<abstract>>
    }
    class CallAdapterFactory["CallAdapter.Factory"] {
        <<abstract>>
    }
    RetrofitBuilder ..> Retrofit : build 创建
    RetrofitBuilder --> "1..*" ConverterFactory : 收集并补入内置转换器
    RetrofitBuilder --> "0..*" CallAdapterFactory : 收集用户配置
    Retrofit --> "1..*" ConverterFactory : 保存只读列表
    Retrofit --> "1..*" CallAdapterFactory : 含平台默认工厂

build() 先检查是否设置 baseUrl，在没有自定义网络工厂时创建 OkHttpClient，根据平台补齐回调执行器，再复制工厂列表并追加默认 CallAdapter.Factory。构造出的 Retrofit 持有不可修改的列表视图；复制的是列表，不是把其中每个工厂对象深复制。

链式返回 this 只是易用性设计。这里体现建造者的关键是**配置收集、校验、默认值处理与产品创建集中在 Builder 中**。本版本的 ServiceMethod.Builder 也负责组装复杂对象，不过它通过反射解析方法注解，并不是同一种公开链式配置 API。

源码：Retrofit.Builder 与 build()、Retrofit 字段和构造器。

动态代理：接口方法如何变成一次调用

应用声明 UserApi 接口后，retrofit.create(UserApi.class) 返回运行时代理。调用接口方法时，JDK 把调用交给 Retrofit 创建的匿名 InvocationHandler；它读取方法信息，准备 OkHttpCall，再把适配结果返回给应用。

类或对象	模式角色	职责
UserApi	业务接口	应用定义的 HTTP API；这里用作教学名称。
RuntimeProxy	代理对象	JDK 动态生成，实现在 create() 中传入的服务接口；不是源码里的固定类名。
InvocationHandler	调用处理接口	JDK 提供的代理调用入口。
RetrofitInvocationHandler	具体调用处理器	Retrofit.create() 中匿名实现的图示别名，持有外层 Retrofit 的访问能力。
Retrofit、ServiceMethod	解析与缓存协作者	前者按 Method 缓存后者；后者保存请求描述、转换器和适配器。
OkHttpCall、CallAdapter	调用与返回值适配协作者	前者封装一次请求，后者决定接口方法最终返回的对象。

classDiagram
    direction TB
    class UserApi {
        <<interface>>
        +getUser() Call
    }
    class RuntimeProxy["运行时代理（示意）"]
    class InvocationHandler {
        <<interface>>
        +invoke(Object proxy, Method method, Object[] args) Object
    }
    class RetrofitInvocationHandler["create 中的匿名 InvocationHandler"] {
        +invoke(Object proxy, Method method, Object[] args) Object
    }
    class Retrofit {
        +create(Class service) Object
        ~loadServiceMethod(Method method) ServiceMethod
    }
    class ServiceMethod
    class OkHttpCall
    class CallAdapter {
        <<interface>>
        +adapt(Call call) Object
    }
    UserApi <|.. RuntimeProxy
    InvocationHandler <|.. RetrofitInvocationHandler
    RuntimeProxy --> InvocationHandler : 调用分派
    Retrofit ..> RuntimeProxy : 通过 JDK Proxy 创建
    RetrofitInvocationHandler --> Retrofit : 访问外层对象
    Retrofit --> ServiceMethod : 缓存方法解析结果
    RetrofitInvocationHandler ..> OkHttpCall : 每次调用创建
    OkHttpCall --> ServiceMethod : 使用请求描述
    ServiceMethod --> CallAdapter : 持有
    RetrofitInvocationHandler ..> CallAdapter : 调用 adapt

对于普通 HTTP 接口方法，处理器依次调用 loadServiceMethod(method)、new OkHttpCall<>(serviceMethod, args)、serviceMethod.callAdapter.adapt(okHttpCall)。默认按需解析方法；开启 validateEagerly 后会在创建代理时提前校验。Object 方法和 Java 8 默认接口方法有单独的处理分支。

这里应用的是动态代理机制：并没有一个手写的 RealUserApi 对象等待代理转发。处理器把接口调用解释为 HTTP 操作，因此类图不能为了凑齐教科书角色而虚构真实服务实现类。创建代理或返回一个 Call，本身也不等于网络请求已经执行。

源码：Retrofit.create()、loadServiceMethod()。

工厂方法：由具体工厂创建适配器

CallAdapter.Factory 声明抽象方法 get(returnType, annotations, retrofit)。具体工厂通过继承这个抽象类，决定自己是否支持方法返回类型，以及应该创建什么适配器。Retrofit 依赖工厂抽象，无须在核心中写出每种 RxJava 返回类型的构造过程。

类或对象	模式角色	职责
CallAdapter.Factory	抽象创建者	定义可被子类覆写的适配器创建操作 get()。
RxJavaCallAdapterFactory	具体创建者	识别 Observable、Single、Completable 等返回类型，创建 RxJavaCallAdapter。
DefaultCallAdapterFactory	具体创建者	处理普通 Call 返回类型，创建直接返回原始 Call 的匿名适配器。
CallAdapter	产品抽象	暴露响应实体类型与适配操作。
RxJavaCallAdapter、IdentityCallAdapter	具体产品	前者是真实类；后者是默认工厂内匿名 CallAdapter 的教学别名。
Retrofit	工厂的使用者	查询工厂并使用返回的产品。

classDiagram
    direction TB
    class Retrofit
    class CallAdapterFactory["CallAdapter.Factory"] {
        <<abstract>>
        +get(Type returnType, Annotation[] annotations, Retrofit retrofit) CallAdapter*
    }
    class RxJavaCallAdapterFactory {
        +get(Type returnType, Annotation[] annotations, Retrofit retrofit) CallAdapter
    }
    class DefaultCallAdapterFactory {
        +get(Type returnType, Annotation[] annotations, Retrofit retrofit) CallAdapter
    }
    class CallAdapter {
        <<interface>>
        +responseType() Type
        +adapt(Call call) Object
    }
    class RxJavaCallAdapter
    class IdentityCallAdapter["匿名 CallAdapter（直接返回 Call）"]
    CallAdapterFactory <|-- RxJavaCallAdapterFactory
    CallAdapterFactory <|-- DefaultCallAdapterFactory
    CallAdapter <|.. RxJavaCallAdapter
    CallAdapter <|.. IdentityCallAdapter
    Retrofit --> CallAdapterFactory : 查询
    CallAdapterFactory ..> CallAdapter : 返回产品
    RxJavaCallAdapterFactory ..> RxJavaCallAdapter : 创建
    DefaultCallAdapterFactory ..> IdentityCallAdapter : 创建

这里按工厂方法理解的是 get() 将具体产品的创建交给子类。RxJavaCallAdapterFactory.create() 则是创建工厂自身的静态便捷方法，不能仅因为方法名叫 create 就把它认作 GoF 工厂方法。Android 默认使用带执行器的 ExecutorCallAdapterFactory，后文会分析它创建的包装对象。

Converter.Factory 同时提供请求体、响应体、字符串转换器的创建入口，但方法数量不是抽象工厂的判定依据。如果从“同一序列化技术下的一组相关转换器”角度讨论，可以说明它具有产品族的组织方式；实际扩展机制允许不同工厂分别处理不同方向或类型，并不要求一次选择成套替换所有转换器。因此本节重点分析可覆写的创建方法与工厂查找流程，不直接把整个机制定性为严格的抽象工厂。

源码：CallAdapter.Factory、默认工厂的匿名产品、RxJavaCallAdapterFactory、Converter.Factory。

适配器：把 Call 转为 RxJava 返回类型

以业务接口返回 Observable<User> 为例，底层仍用 Call<User> 表达一次请求，但调用者希望使用订阅、取消和响应式操作符。RxJavaCallAdapter 通过 RxJava 的订阅源将两套使用方式接起来。

类或对象	模式角色或协作者	职责
Call、OkHttpCall	被适配接口与具体调用	提供 execute()、enqueue()、cancel()、clone()。
CallAdapter、RxJavaCallAdapter	适配契约与具体适配器	将 Call<R> 转为接口声明的返回对象。
rx.Observable	调用者需要的使用形式	对外提供 RxJava 订阅协议；Single、Completable 由后续转换得到。
Observable.OnSubscribe	订阅源接口	在真正订阅时连接网络调用。
CallExecuteOnSubscribe、CallEnqueueOnSubscribe	同步与异步订阅源	保存 originalCall，每次订阅时克隆它，再分别执行 execute() 或 enqueue()。

classDiagram
    direction TB
    class CallAdapter {
        <<interface>>
        +responseType() Type
        +adapt(Call call) Object
    }
    class RxJavaCallAdapter {
        +responseType() Type
        +adapt(Call call) Object
    }
    class Call {
        <<interface>>
        +execute() Response
        +enqueue(Callback callback)
        +clone() Call
    }
    class OkHttpCall
    class Observable["rx.Observable"]
    class OnSubscribe["rx.Observable.OnSubscribe"] {
        <<interface>>
        +call(Subscriber subscriber)
    }
    class CallExecuteOnSubscribe {
        -Call originalCall
        +call(Subscriber subscriber)
    }
    class CallEnqueueOnSubscribe {
        -Call originalCall
        +call(Subscriber subscriber)
    }
    CallAdapter <|.. RxJavaCallAdapter
    Call <|.. OkHttpCall
    RxJavaCallAdapter ..> Call : adapt 参数
    RxJavaCallAdapter ..> Observable : 构造并返回
    RxJavaCallAdapter ..> CallExecuteOnSubscribe : 同步模式创建
    RxJavaCallAdapter ..> CallEnqueueOnSubscribe : 异步模式创建
    Observable --> OnSubscribe : 订阅源
    OnSubscribe <|.. CallExecuteOnSubscribe
    OnSubscribe <|.. CallEnqueueOnSubscribe
    CallExecuteOnSubscribe --> Call : originalCall
    CallEnqueueOnSubscribe --> Call : originalCall

协作过程分为两个阶段：

1. 适配阶段：adapt() 根据 isAsync 创建同步或异步订阅源，再按响应形态选择 BodyOnSubscribe、ResultOnSubscribe 或直接使用原始源，最后创建 Observable。若有 Scheduler，应用 subscribeOn()；声明为 Single 或 Completable 时，再转换返回形式。这个阶段不会执行 HTTP 请求。
2. 订阅阶段：实际订阅到源时，先调用 originalCall.clone()，再执行克隆对象的 execute() 或 enqueue()。CallArbiter 协调响应交付、需求和取消；它控制的是响应交付，不能理解为“收到背压请求后才启动 HTTP”。

RxJavaCallAdapterFactory.create() 默认使用同步调用；配置 Scheduler 后，它仍然调用阻塞的 execute()，只是执行订阅源的线程可能改变。createAsync() 才选择 enqueue() 路径。RxJavaCallAdapter 自身没有保存 Call 的字段，图中是参数依赖；实际持有 Call 的是订阅源。

这个适配器是由多个协作者完成协议转换的实现：它实现的是 Retrofit 的 CallAdapter，并不继承 Observable。CallAdapter 改变调用结果的使用方式，Converter 则把响应体变为实体，两者职责不同。例如 Observable<User> 的 responseType() 是 User，而不是 Observable<User>。

源码：RxJavaCallAdapter、RxJavaCallAdapterFactory、同步订阅源、异步订阅源、CallArbiter。

策略：将数据转换算法交给 Converter

请求执行流程需要把响应体转换为 Java 对象，却不应该依赖某个 JSON 库。ServiceMethod 保存 Converter<ResponseBody, R>，toResponse() 只调用其 convert()；具体解析由 Gson 或 Moshi 等转换器完成。这是从使用者角度观察到的策略结构。

类或对象	模式角色	职责
ServiceMethod	上下文	持有响应转换策略，为 OkHttpCall 提供 toResponse()。
Converter	策略接口	用 convert(F value) 描述从输入到输出的转换。
GsonResponseBodyConverter	具体策略	使用 Gson 的 TypeAdapter 读取响应体。
MoshiResponseBodyConverter	具体策略	使用 Moshi 的 JsonAdapter 解析响应体。
OkHttpCall	上游调用者	在成功响应解析阶段调用 ServiceMethod.toResponse()。

classDiagram
    direction TB
    class OkHttpCall {
        ~parseResponse(okhttp3.Response response) Response
    }
    class ServiceMethod {
        -Converter responseConverter
        ~toResponse(ResponseBody body) Object
    }
    class Converter {
        <<interface>>
        +convert(Object value) Object
    }
    class GsonResponseBodyConverter {
        +convert(ResponseBody value) Object
    }
    class MoshiResponseBodyConverter {
        +convert(ResponseBody value) Object
    }
    OkHttpCall --> ServiceMethod : 解析响应实体
    ServiceMethod --> Converter : responseConverter
    Converter <|.. GsonResponseBodyConverter
    Converter <|.. MoshiResponseBodyConverter

构建 ServiceMethod 时，Builder 先从 CallAdapter.responseType() 获得真正的响应实体类型，再通过 Retrofit.responseBodyConverter() 选定转换器。请求成功且有需要解析的响应体时，OkHttpCall 使用这项策略；非 2xx 响应作为错误响应处理，204/205 则返回空实体，不经过这条实体转换路径。

这是在方法解析阶段选好并保存的策略，不是在每次请求中随意切换 Gson 与 Moshi。图中两者表示可选择的实现，不能理解为会依次解析同一份响应体。CallAdapter 从“替换返回值处理方式”角度也有策略性，但它解决的主要问题是接口适配，前文已单独讨论。

源码：ServiceMethod 的转换器字段与 toResponse()、GsonResponseBodyConverter、MoshiResponseBodyConverter、OkHttpCall 的响应解析。

责任链思想：按顺序选择第一个匹配的工厂

nextCallAdapter() 遍历 adapterFactories，调用各个工厂的 get()。返回 null 表示该工厂不能处理，继续尝试下一个；返回非空适配器就立即结束。这符合“多个处理者按顺序尝试，找到能够处理请求的对象”的责任链思想。

类或对象	对应角色	职责
Retrofit	分派者	维护工厂顺序、跳过指定工厂、检查结果并在全部不匹配时报错。
CallAdapter.Factory	处理者抽象	接收返回类型和注解，返回适配器或 null。
RxJavaCallAdapterFactory、ExecutorCallAdapterFactory	具体处理者	分别尝试处理 RxJava 类型与带回调执行器的 Call 类型。
CallAdapter	处理结果	当前返回类型选中的适配方式。

classDiagram
    direction TB
    class Retrofit {
        ~List adapterFactories
        +nextCallAdapter(CallAdapter.Factory skipPast, Type returnType, Annotation[] annotations) CallAdapter
    }
    class CallAdapterFactory["CallAdapter.Factory"] {
        <<abstract>>
        +get(Type returnType, Annotation[] annotations, Retrofit retrofit) CallAdapter*
    }
    class RxJavaCallAdapterFactory
    class ExecutorCallAdapterFactory
    class CallAdapter {
        <<interface>>
    }
    Retrofit --> "1..*" CallAdapterFactory : 按列表顺序询问
    CallAdapterFactory <|-- RxJavaCallAdapterFactory
    CallAdapterFactory <|-- ExecutorCallAdapterFactory
    CallAdapterFactory ..> CallAdapter : 匹配时返回

例如注册 RxJava 工厂后，在 Android 默认配置下查询 Observable<User>，RxJava 工厂可以直接命中；查询 Call<User> 时它返回 null，后面的平台默认工厂继续处理。nextCallAdapter(skipPast, ...) 从指定工厂之后开始查找，可供包装型扩展向后委托，避免再次匹配自己。

这里用**列表加循环**实现分派，工厂对象之间没有 next 字段，不能画出源码中不存在的 handler.next 关联；也不是 OkHttp 拦截器那种各层都可能参与的执行链。若某个工厂抛出异常，查找也会直接中断，不会把异常当成不匹配而自动尝试下一个。

转换器有类似的查找机制。2.3.0 的内置转换器被最先加入；用户转换器按注册顺序尝试。GsonConverterFactory 假定自己可以处理所有类型，和更专用的转换器混用时通常要放在后面，否则会提前截获查询。这里创建的 Converter 是响应处理策略，查找工厂则是另一层职责。

源码：nextCallAdapter()、nextResponseBodyConverter()、GsonConverterFactory 的顺序约定。

装饰者：为 Call 增加回调调度

ExecutorCallAdapterFactory 创建的匿名适配器会用 ExecutorCallbackCall 包住原始 Call。这个包装对象仍然实现 Call<T>，调用者继续使用相同接口；增强的是异步回调的交付方式。

类或对象	模式角色	职责
Call	组件接口	统一执行、取消、克隆和状态查询等操作。
OkHttpCall	具体组件	实际封装并执行网络请求；delegate 也允许其他 Call 实现。
ExecutorCallAdapterFactory.ExecutorCallbackCall	具体装饰者	持有 delegate 和 callbackExecutor，把大部分操作委托给前者。
Executor	调度协作者	接收交付回调的 Runnable。Android 默认实现向主线程 Handler 投递任务。
ExecutorCallAdapterFactory	装饰对象的创建入口	通过 get() 返回的匿名适配器，在 adapt() 时组装包装层。

classDiagram
    direction TB
    class Call {
        <<interface>>
        +enqueue(Callback callback)
        +execute() Response
        +cancel()
        +clone() Call
    }
    class OkHttpCall
    class ExecutorCallbackCall["ExecutorCallAdapterFactory.ExecutorCallbackCall"] {
        ~Call delegate
        ~Executor callbackExecutor
        +enqueue(Callback callback)
        +execute() Response
        +cancel()
        +clone() Call
    }
    class Executor {
        <<interface>>
        +execute(Runnable task)
    }
    class ExecutorCallAdapterFactory
    Call <|.. OkHttpCall
    Call <|.. ExecutorCallbackCall
    ExecutorCallbackCall --> Call : delegate
    ExecutorCallbackCall --> Executor : callbackExecutor
    ExecutorCallAdapterFactory ..> ExecutorCallbackCall : 通过匿名适配器创建

enqueue(callback) 把内部回调交给 delegate.enqueue()；底层返回结果后，内部回调再向 callbackExecutor 提交 Runnable，最终通知用户回调。成功响应的 Runnable 执行时会再次检查取消状态：如果已经取消，就交付取消错误。传给用户回调的 Call 是外层 ExecutorCallbackCall，保持调用者看到的对象一致。

execute() 直接调用 delegate.execute()，不会因此自动切换线程。cancel()、状态查询和 request() 也主要委托给被包装对象。只有在 Android 的默认平台配置下，这个执行器才对应主线程；自定义 Executor 可以改变回调交付位置，不能把“主线程”当作装饰类本身的固定属性。

它和前面的 RxJava 适配器区别在于：装饰者保持 Call 接口并增加行为；RxJava 适配器把调用转成另一套对外使用形式。类名中出现 Adapter，不代表其创建的每一层对象都只能按适配器理解。

源码：ExecutorCallAdapterFactory 和 ExecutorCallbackCall、Platform 的默认执行器。

原型：通过 clone 创建可再次执行的调用

Call 是一次性调用对象，同一个实例不能重复执行。RxJava 的冷 Observable 可能被重新订阅，订阅源因此保存一份原始 Call，并在每次真正订阅到该源时通过接口克隆出新的调用对象。

类或对象	模式角色	职责
Call	原型接口	声明 clone()，让调用者通过统一契约复制调用描述。
OkHttpCall	具体原型	创建新的 OkHttpCall，复用请求描述与参数引用，重新开始执行状态。
CallExecuteOnSubscribe、CallEnqueueOnSubscribe	使用原型的客户端	持有 originalCall，无需了解具体构造器即可创建每次订阅所需的 Call。
ServiceMethod	共享的请求描述	保存方法解析结果、适配器、转换器等信息。

classDiagram
    direction TB
    class Call {
        <<interface>>
        +clone() Call
    }
    class OkHttpCall {
        -ServiceMethod serviceMethod
        -Object[] args
        -boolean executed
        -boolean canceled
        +clone() OkHttpCall
    }
    class ServiceMethod
    class CallExecuteOnSubscribe {
        -Call originalCall
        +call(Subscriber subscriber)
    }
    class CallEnqueueOnSubscribe {
        -Call originalCall
        +call(Subscriber subscriber)
    }
    Call <|.. OkHttpCall
    CallExecuteOnSubscribe --> Call : originalCall 与 clone
    CallEnqueueOnSubscribe --> Call : originalCall 与 clone
    OkHttpCall --> ServiceMethod : 克隆前后共享
    OkHttpCall ..> OkHttpCall : clone 创建新实例

OkHttpCall.clone() 返回 new OkHttpCall<>(serviceMethod, args)。因此新实例有独立的 executed、canceled 状态，底层 rawCall 和创建异常也从空值开始；它没有复制已经执行的网络对象，也没有调用已有 okhttp3.Call.clone()。

但它同样**不是完整的深复制**：serviceMethod、args 数组及数组里的参数对象都继续共享。后续构建请求前，如果可变参数被修改，克隆调用也可能看到变化。外层 ExecutorCallbackCall.clone() 会创建新包装层并调用 delegate.clone()，执行器仍然共享。

原型思想体现在“客户端通过现有对象的复制契约得到新对象”，不要求内部必须调用 Object.clone()。每次订阅源产生新调用，也不等于缓存响应；使用 share() 或缓存操作符后，下游订阅是否再次触发源，则由 RxJava 的组合语义决定。

源码：Call 的 clone 契约、OkHttpCall 字段和 clone()、同步订阅源、异步订阅源。

对照历史摘录时注意版本差异

上方旧代码保留了学习过程，不是本节固定版本的完整源码。核对时尤其注意：

历史摘录	Retrofit 2.3.0 中的实现或准确含义
CallOnSubscribe 及其内部 CallArbiter	同步与异步源分别为 CallExecuteOnSubscribe、CallEnqueueOnSubscribe；CallArbiter 已是独立顶层类。
CompletableHelper.toCompletable(observable)	固定版本直接调用 observable.toCompletable()。
RxJavaCallAdapterFactory 没有异步分支	固定版本包含 createAsync() 和传入适配器的 isAsync 标志。
两份 parseResponse() 的 204/205 分支不同	固定版本在返回空实体前调用 rawBody.close()。
将 callFactory.newCall(request) 注释为执行请求	这里只创建底层调用对象；网络执行在 execute() 或 enqueue() 阶段。
将订阅时的 clone 注释为克隆 OkHttp 调用	直接调用的是 retrofit2.Call.clone()；OkHttpCall 通过构造器重建自身。

设计模式描述的是职责分配，不是类名标签。Factory、Adapter、Builder 的名称可以帮助定位代码，但最终仍需查看对象如何创建、持有什么引用，以及调用怎样流转。方法解析结果的缓存也不等于单例，状态字段或条件分支也不能直接算作状态模式。

评论