更新到 Retrofit 2.0 后，PublishSubject 在不同线程中的 onNext 调用答案

【问题标题】：PublishSubject's onNext call in different thread after I update to Retrofit 2.0更新到 Retrofit 2.0 后，PublishSubject 在不同线程中的 onNext 调用
【发布时间】：2016-06-22 15:01:36
【问题描述】：

我有一个我的同事在我们使用 Retrofit 1.9 时创建的以下课程

public class SomeApiCallAction {

  private Subscription subscription;
  private NoInternetConnectionInterface noInternetConnectionInterface;

  public interface NoInternetConnectionInterface {
      PublishSubject<Integer> noInternetConnection(Throwable throwable);
  }

  public void execute(Subscriber subscriber, NoInternetConnectionInterface noInternetConnectionInterface) {
      this.noInternetConnectionInterface = noInternetConnectionInterface;
      this.subscription = retrofit.someService().someApiCall()
          .subscribeOn(Schedulers.newThread())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(subscriber)
          .retryWhen(retryFunction);
  }

  public void cancel() {
      if (this.subscription != null) {
          this.subscription.unsubscribe();
      }
  }             

  private Func1<Observable<? extends Throwable>, Observable<?>> retryFunction = new Func1<Observable<? extends Throwable>, Observable<?>>() {
      @Override
      public Observable<?> call(Observable<? extends Throwable> observable) {
          return observable.flatMap(new Func1<Throwable, Observable<?>>() {
              @Override
              public Observable<?> call(final Throwable throwable) {
                  if (noInternetConnectionInterface!= null && (throwable instanceof IOException || throwable instanceof SocketTimeoutException)) {
                      return noInternetConnectionInterface.noInternetConnection(throwable);
                  }else{
                      return Observable.error(throwable);
                  }
              }
          });
      }
}

SomeApiCallAction 只是一个简单的类，里面封装了改造 api 调用，唯一特别的是它的重试功能。 retry 函数将检查 throwable 是否是 IOException 或 SocketTimeoutException 类型，如果是，它将调用接口，以便我们可以向用户显示重试对话框，询问他们是否要重试操作。我们的用法类似于下面的sn-p

public class SomeActivity implement NoInternetConnectionInterface {

    @OnClick(R.id.button)
    public void do(View v) {
        new SomeApiCallAction().execute(
            new Subscriber(),
            this
        )
    }

    @Override
    public PublishSubject<Integer> noInternetConnection(final Throwable throwable) {
        Log.i("Dev", Thread.currentThread() + " Error!");
        final PublishSubject<Integer> subject = PublishSubject.create();

        runOnUiThread(new Runnable() {
            @Override
            public void run() {
                NoInternetDialogFragment dialog = NoInternetDialogFragment.newInstance();
                dialog.setNoInternetDialogFragmentListener(new NoInternetDialogFragmentListener{
                    @Override
                    public void onUserChoice(boolean retry, NoInternetDialogFragment dialog) {
                        Log.i("Dev", Thread.currentThread() + " Button Click!");
                        if (retry) {
                            subject.onNext(1);
                        } else {
                            subject.onError(throwable);
                        }

                        dialog.dismiss();

                    }
                });
                dialog.show(getSupportFragmentManager(), NoInternetDialogFragment.TAG);
            }
        });
        return subject;
    }
}

当我们使用 Retrofit 1.9.0 时，此实现运行良好。我们通过开启飞行模式进行测试，按下按钮执行api调用。

第一次执行失败，我在重试函数中得到了 UnknownHostException。
所以，我调用接口（Activity）来呈现重试对话框
我在飞行模式下按下重试按钮以重复执行
正如预期的那样，用户按下重试按钮后发生的每次执行都失败了，我总是在重试函数中得到 UnknownHostException。
如果我一直按重试按钮，重试对话框将永远出现，直到我关闭飞行模式。

但是在我们将依赖项更新为

'com.squareup.retrofit2:retrofit:2.0.2'
'com.squareup.retrofit2:adapter-rxjava:2.0.2'

我们再试一次，但这次行为改变了，

第一次执行失败，我在重试函数中得到了 UnknownHostException 和以前一样。
所以，我调用接口（Activity）来呈现重试对话框
我在飞行模式下按下重试按钮以重复执行
但是这一次，在重试函数中，我没有像以前那样收到 UnknowHostException，而是收到了 NetworkOnMainThreadException
所以条件不匹配，接口没有被调用，结果是只向用户显示了 1 个重试对话框。

以下是上述代码的日志

Thread[android_0,5,main] Error!
Thread[main,5,main] Button Click!

您知道是什么原因造成的吗？任何建议，评论将不胜感激。

注意：以下是我们一直在使用并可能相关的其他依赖项。但是它们最近没有更新，从这个项目开始就一直在使用这些版本。

'com.jakewharton:butterknife:8.0.1'

'io.reactivex:rxandroid:1.1.0'
'io.reactivex:rxjava:1.1.0'

'com.google.dagger:dagger-compiler:2.0'
'com.google.dagger:dagger:2.0'
'javax.annotation:jsr250-api:1.0'

更多信息

我只是将我的代码重置回我们使用 Retrofit 1.9 时的点，我发现打印日志不同

Thread[Retrofit-Idle,5,main] Error!
Thread[main,5,main] Button Click!

不确定这是否与问题相关，但很明显，在 1.9.0 中，与 2.0.0 相比，我在不同线程中调用接口

最终编辑

在阅读了@JohnWowUs 的答案并点击他提供的链接后，我发现在 Retrofit 2 中，网络调用默认是同步的

要解决我的问题，有两种方法可以做到这一点

1.) 按照@JohnWowUs 的建议为 retryFunction 指定线程

this.subscription = retrofit.someService().someApiCall()
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(subscriber)
  .retryWhen(retryFunction, Schedulers.io());

2.) 创建改造对象时，创建 RxJavaCallAdapterFactory 时指定线程

retrofit = new Retrofit.Builder()
  .baseUrl(AppConfig.BASE_URL)
  .client(client)
  .addConverterFactory(GsonConverterFactory.create(getGson()))
  .addCallAdapterFactory(
     RxJavaCallAdapterFactory.createWithScheduler(
       Schedulers.from(threadExecutor)
     )
   )
  .build();

【问题讨论】：

标签： android retrofit rx-java retrofit2

【解决方案1】：

我认为问题在于，当您重新订阅时，由于在retryWhen 中使用了默认的蹦床调度程序，您正在订阅主线程。 Retrofit 1.9 为您处理了日程安排，因此使用subscribeOn 毫无意义。问题讨论是here。在 Retrofit 2 中，我相信这已经改变了，所以你应该尝试类似

this.subscription = retrofit.someService().someApiCall()
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(subscriber)
      .retryWhen(retryFunction, Schedulers.io());

【讨论】：

哇，感谢您提供的信息，包括讨论链接，这是我正在寻找的正确解决方案！

【解决方案2】：

为了快速解决这个问题，我会这样做

new Thread(new Runnable(){
    public void run() {
        choice.onNext(1);
    }
}).start();

该应用再次按预期运行。但是，我认为这不是解决此问题的正确方法，因此我将保留此问题以供进一步评论。

【讨论】：