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

Question

我有一个我的同事在我们使用 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 只是一个简单的类，将retrofit api 调用包装在里面，唯一特别的是它的重试功能。重试函数将检查 throwable 是否是一种 IOException 或 SocketTimeoutException，如果是，它将调用接口(interface)以便我们可以向用户呈现重试对话框以询问他们是否要重试操作。我们的用法类似于以下代码 fragment

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。
• 所以，我调用接口(interface)(Activity)来呈现重试对话框
• 我在飞行模式下按下重试按钮以重复执行
• 正如预期的那样，在用户按下重试按钮失败后发生的每次执行，我总是在重试函数中得到 UnknownHostException。
• 如果我一直按重试按钮，重试对话框将一直出现，直到我关闭飞行模式。

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

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

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

• 第一次执行失败，我在重试函数中遇到了 UnknownHostException，和以前一样。
• 所以，我调用接口(interface)(Activity)来呈现重试对话框
• 我在飞行模式下按下重试按钮以重复执行
• 但是这次，在重试函数中，我没有像原来那样接收 UnknowHostException，而是接收到 NetworkOnMainThreadException
• 因此条件不匹配，接口(interface)未被调用，结果只向用户显示 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!

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

最终编辑

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

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

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

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

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

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

Answer 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());