如何在RxJava2中使用带有lambda表达式的DisposableObserver

Question

我的用例是想在我的onNext中的某个条件之后处理.所以尝试使用DisposableObserver.这是有效的代码

Observable.just(1, 2, 3, 4)
    .subscribe(new DisposableObserver<Integer>() {
                     @Override
                     public void onNext(Integer integer) {
                       System.out.println("onNext() received: " + integer);
                       if (integer == 2) {
                         dispose();
                       }
                     }
                     @Override
                     public void onError(Throwable e) { System.out.println("onError()"); }
                     @Override
                     public void onComplete() { System.out.println("onComplete()"); }
                   }
    );

现在,如果你尝试用lambda替换它,它会将lambda视为

subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,Action onComplete)

现在这样做.通过从onSubscribe保存一次性用品,然后调用disposable.dispose(); 来自onNext.

  private Disposable disposable;
  private void disposableObserverTest() {
    Observable.just(1, 2, 3, 4)
        .subscribe(integer -> {
              System.out.println("onNext() received: " + integer);
              if (integer == 2) {
                disposable.dispose();
              }

            }, throwable -> System.out.println("error"),
            () -> System.out.println("complete"),
            disposable1 -> {
              this.disposable = disposable1;
            });
  }

但是,如果你想直接调用dispose()如何用lambdas做呢？

Answer 1

这是因为在第一种情况下你打电话

subscribe(DisposableObserver observer)

而在第二种情况下你打电话

subscribe(Action1<? extends Integer> onNext, Action1<? extends Throwable> onError, Action0 onComplete)

这意味着在第二种情况下，您不持有对 的引用DisposableObserver，因此您无法调用dispose()它。