如何在 onComplete 之后显式取消订阅 observable答案

【问题标题】：How to explicitly unSubscribe from observable after the onComplete如何在 onComplete 之后显式取消订阅 observable
【发布时间】：2017-12-24 07:44:48
【问题描述】：

在下面的代码中，在完成onComplete 之后，如何以及在何处放置unSubscribe 以明确取消订阅Observable？。

getObservable()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeOn(Schedulers.io())
    .subscribe(new Subscriber<Boolean>() {
        @Override
        public void onCompleted() {
            doSomething();
        }

        @Override
        public void onError(Throwable e) {
            thereIsError();
        }

        @Override
        public void onNext(Boolean status) {
            updateView();
        }
    });

【问题讨论】：

标签： rx-java rx-java2 rx-android

【解决方案1】：

你可以保存subscribe方法返回的Disposable，并在onCompleted回调中使用

Disposable d = 
getObservable()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeOn(Schedulers.io())
    .subscribe(new Subscriber<Boolean>() {
        @Override
        public void onCompleted() {
            doSomething();
            d.dispose();
        }

        @Override
        public void onError(Throwable e) {
            thereIsError();
        }

        @Override
        public void onNext(Boolean status) {
            updateView();
        }
    });

【讨论】：

如果出现异常并调用onError 怎么办？是否会因为流没有被释放而导致内存泄漏？

【解决方案2】：

 disposables.add(sampleObservable()
            // Run on a background thread
            .subscribeOn(Schedulers.io())
            // Be notified on the main thread
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new DisposableObserver<String>() {
                @Override
                public void onComplete() {
                    // Do some work for completion
                }

                @Override
                public void onError(Throwable e) {
                     // Do some work for error
                }

                @Override
                public void onNext(String value) {
                    // Do some work for next
                }
            }));

清除并取消订阅

@Override
protected void onDestroy() {
    super.onDestroy();
    disposables.clear(); // do not send event after activity has been destroyed
    disposables.dispose();
}

【讨论】：

【解决方案3】：

简短的回答是，您没有理由明确取消订阅，因为这是由观察者链完成的。您可以通过添加此运算符来自己测试：

.doOnUnsubscribe( () -> System.out.println("Unsubscribed") )

【讨论】：

任何终止事件（错误或完成）都会取消订阅（除非您使用类似 .retry() 的东西）

【解决方案4】：

如果您想立即取消订阅，只需使用 take(1)。在你调用 take(1) 之后，它应该调用 onComplete 。

Disposable d = 
getObservable()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeOn(Schedulers.io())
    .take(1)//take the first, then stop emissions
    .subscribe(new Subscriber<Boolean>() {
        @Override
        public void onCompleted() {
            doSomething();
            d.dispose();
        }

        @Override
        public void onError(Throwable e) {
            thereIsError();
        }

        @Override
        public void onNext(Boolean status) {
            updateView();
        }
    });

【讨论】：