1，主要類是CallExecuteObservable 

@Override protected void subscribeActual(Observer<? super Response<T>> observer) {
  // Since Call is a one-shot type, clone it for each new observer.
  Call<T> call = originalCall.clone();
  CallDisposable disposable = new CallDisposable(call);
  observer.onSubscribe(disposable);

  boolean terminated = false;
  try {
    //執行網路請求
 

    Response<T> response = call.execute();
    //新增是否取消訂閱的判斷
    if (!disposable.isDisposed()) {
      //正常會執行這裡，呼叫我們的觀察者的onNext
      observer.onNext(response);
    }
    if (!disposable.isDisposed()) {
      terminated = true;
      //呼叫我們的觀察者的onComplete
      observer.onComplete();
    }
  } catch (Throwable t) {
    Exceptions.throwIfFatal(t);
    if (terminated) {
      RxJavaPlugins.onError(t);
    } else if (!disposable.isDisposed()) {
      try {
        //呼叫我們的觀察者的onError
        observer.onError(t);
 

      } catch (Throwable inner) {
        Exceptions.throwIfFatal(inner);
        RxJavaPlugins.onError(new CompositeException(t, inner));
      }
    }
  }
}

2,用retrofit +rxjava+okhttp封裝的網路呼叫

以下是簡單的呼叫,具體回撥怎麼實現，看了上面的原始碼應該很清晰了

OkhttpManager.getInstance().test1(map).subscribe(new Observer() {             @Override             public void onSubscribe(Disposable disposable) {

            }

            @Override             public void onNext(Object o) {

            }

            @Override             public void onError(Throwable throwable) {

            }

            @Override             public void onComplete() {

            }         });