java - Observable.take() 导致 NPE

Question

我有一个像这样的简单程序:

public class MainApp {
    public static void main(String[] args) {
        getAcronyms()
                .flatMap(Observable::fromIterable)
                .flatMap(MainApp::getTitle)
                .filter(Objects::nonNull)
                .subscribe(System.out::println);

    }

    private static Observable<List<String>> getAcronyms(){
        List<String> strings = new ArrayList<>();
        strings.add("YOLO");
        strings.add("LMAO");
        strings.add("ROFL");
        strings.add("AYY LMAO");
        return new Observable<List<String>>() {
            @Override
            protected void subscribeActual(Observer<? super List<String>> observer) {
                observer.onNext(strings);
                observer.onComplete();
            }
        };
    }

    private static Observable<String> getTitle(String url) {
        return new Observable<String>() {
            @Override
            protected void subscribeActual(Observer<? super String> observer) {
                observer.onNext(url + " title!");
                observer.onComplete();
            }
        };
    }
}

这工作正常，但是当我链接take时:

getAcronyms()
        .flatMap(Observable::fromIterable)
        .flatMap(MainApp::getTitle)
        .filter(Objects::nonNull)
        .take(2)
        .subscribe(System.out::println);

它打印 2 个值，但给我一个 NPE:

YOLO title!

LMAO title!

Exception in thread "main" java.lang.NullPointerException at io.reactivex.internal.operators.observable.ObservableTake$TakeObserver.onComplete(ObservableTake.java:83) at io.reactivex.internal.operators.observable.ObservableTake$TakeObserver.onNext(ObservableTake.java:64) at io.reactivex.internal.operators.observable.ObservableFilter$FilterObserver.onNext(ObservableFilter.java:52) at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.tryEmit(ObservableFlatMap.java:262) at io.reactivex.internal.operators.observable.ObservableFlatMap$InnerObserver.onNext(ObservableFlatMap.java:559) at MainApp$2.subscribeActual(MainApp.java:41) at io.reactivex.Observable.subscribe(Observable.java:10842) at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.subscribeInner(ObservableFlatMap.java:162) at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.onNext(ObservableFlatMap.java:139) at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.drainLoop(ObservableFlatMap.java:436) at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.drain(ObservableFlatMap.java:323) at io.reactivex.internal.operators.observable.ObservableFlatMap$InnerObserver.onSubscribe(ObservableFlatMap.java:546) at io.reactivex.internal.operators.observable.ObservableFromIterable.subscribeActual(ObservableFromIterable.java:55) at io.reactivex.Observable.subscribe(Observable.java:10842) at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.subscribeInner(ObservableFlatMap.java:162) at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.onNext(ObservableFlatMap.java:139) at MainApp$1.subscribeActual(MainApp.java:31) at io.reactivex.Observable.subscribe(Observable.java:10842) at io.reactivex.internal.operators.observable.ObservableFlatMap.subscribeActual(ObservableFlatMap.java:55) at io.reactivex.Observable.subscribe(Observable.java:10842) at io.reactivex.internal.operators.observable.ObservableFlatMap.subscribeActual(ObservableFlatMap.java:55) at io.reactivex.Observable.subscribe(Observable.java:10842) at io.reactivex.internal.operators.observable.ObservableFilter.subscribeActual(ObservableFilter.java:30) at io.reactivex.Observable.subscribe(Observable.java:10842) at io.reactivex.internal.operators.observable.ObservableTake.subscribeActual(ObservableTake.java:30) at io.reactivex.Observable.subscribe(Observable.java:10842) at io.reactivex.Observable.subscribe(Observable.java:10828) at io.reactivex.Observable.subscribe(Observable.java:10731) at MainApp.main(MainApp.java:18)

有人可以帮我弄清楚为什么会发生这种情况以及我做错了什么吗？

Answer 1

发生此异常的原因是take，在发出定义数量的项目后，在内部尝试处置未设置的Disposable对象。

因此，您必须在创建初始 Observable 时通过在 subscribeActual 方法实现中调用 observer.onSubscribe(disposable) 来提供它。但不要重新发明轮子，此外，通过调用其公共(public)构造函数来创建 Observable 是针对自定义运算符的。只需使用静态工厂方法即可。在您的情况下，最好的选择是 Observable.fromCallable:

private static Observable<List<String>> getAcronyms(){
    return Observable.fromCallable(new Callable<List<String>>() {
        @Override
        public List<String> call() throws Exception {
            List<String> strings = new ArrayList<>();
            strings.add("YOLO");
            strings.add("LMAO");
            strings.add("ROFL");
            strings.add("AYY LMAO");
            return strings;
        }
    });
}

另请查看这篇文章:RxJava 2 Disposable -  Under the hood