可观察到。创建和观察者处置

Question

我试图理解当我使用

    Observable.just(1).subscribe(new Observer<Integer>() {

        Disposable disposable;

        @Override
        public void onSubscribe(Disposable disposable) {
            System.out.println("Subscribed");
            this.disposable = disposable;
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println(integer);
            System.out.println(disposable.isDisposed());
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("Error");
            System.out.println(disposable.isDisposed());
        }

        @Override
        public void onComplete() {
            System.out.println("Complete");
            System.out.println(disposable.isDisposed());
        }
    })

在On完成或OnError之后，disposable.isDispse（）在我使用时返回true

  Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> observableEmitter) throws Exception {
            if (!observableEmitter.isDisposed())
                observableEmitter.onComplete();
        }
    }).subscribe(new Observer<Integer>() {

        Disposable disposable;

        @Override
        public void onSubscribe(Disposable disposable) {
            System.out.println("Subscribed");
            this.disposable = disposable;
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println(integer);
            System.out.println(disposable.isDisposed());
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("Error");
            System.out.println(disposable.isDisposed());
        }

        @Override
        public void onComplete() {
            System.out.println("Complete");
            System.out.println(disposable.isDisposed());
        }
    }); 

我知道是一次性的。isDisposed（）返回false。有人能解释一下到底发生了什么吗？。我理解一个写得很好的观察。create不能在onComplete（）或onError（）之后发出项。

Answer 1

最初，isDispse旨在指示调用了dispose。由于默认情况下onError和on完整不调用dispose，因此您通常不会在终止后获得isDispsetrue。Reactive Streams规范规定，当调用onError和on完整时，应考虑取消订阅，这意味着无需在实现中存在实际代码（因此开销）。

不幸的是，人们开始在流之外使用它作为完成指示符（即，查看特定的订阅（）是否已达到其终端状态）。我不建议依赖isDispse（在Disposable容器之外），因为它将对完成的异步反应与可能的同步（和阻塞）等待完成混为一谈。

在您的特定情况下，在返回onComplete后，一次性将指示isDisposed，因此外部视图将指示isDisposedtrue。

Answer 2

一次性只保留对订阅的引用，为了处理它，您需要调用一次性。dispose（），可观察的不会在complete方法上处理一次性的实例。

 @Override
        public void onComplete() {
            System.out.println("Complete");
            disposable.dispose();
            System.out.println(disposable.isDisposed());
        }