java - 取消 Observable 的执行 (RxJava 1)

Question

我已经使用订阅创建了 Observable:

private Subscription mSearchSubscription = null;

mSearchSubscription = Observable.fromCallable(() -> {
    many network requests
}).subscribe(doing sth);

当我的 Activity 被破坏时，我正在执行以下行:

 if (mSearchSubscription != null){
        mSearchSubscription.unsubscribe();
    }

根据在互联网上找到的信息，这一行应该取消该 block 的所有代码执行。但事实并非如此:(该 block 内的所有网络请求都用订阅包装了 Observables，因此当一个请求完成时，下一个请求正在执行，依此类推。正如我注意到的，在执行取消订阅并且 Activity 被销毁后，Observable 内的代码仍在执行。

有什么更好的方法可以阻止它吗？我在这个项目中使用的是RxJava1

更新:我有 MVC 架构， Controller 注入(inject)到 Activity 中。通过重写 Activity 中的 onDestroy，我调用 Controller 中的方法来取消订阅。这是整个代码。 fromCallable(()-> ) block 内的每个网络请求都像这样 mWebService.getSth(bodyparameters).subscribe(doSth) (在接口(interface)中返回 Observable )。我不希望在内部可观察的执行期间停止，而是停止执行这段代码，就像 return 一样。之所以进行包装，是因为需要将许多请求一一执行，并在一切准备就绪后将结果返回给 Activity。

Answer 1

您没有详细说明如何以及为何这样做，因此我最好的猜测建议是使用 create:

Observable.create(emitter -> {
    CompositeSubscription resources = new CompositeSubscription();
    emitter.setSubscription(resources);

    if (resources.isUnsubscribed()) {
        return;
    }

    // add these if you want to cancel the inner sources
    // resources.add(

    mWebService.getFirst(/* params */).subscribe(/* postprocess */);

    // );

    if (resources.isUnsubscribed()) {
        return;
    }
    mWebService.getSecond(/* params */).subscribe(/* postprocess */);

    if (resources.isUnsubscribed()) {
        return;
    }

    mWebService.getThird(/* params */).subscribe(/* postprocess */);

    emitter.onNext(/* the result object*/
    emitter.onComplete();
}, Emitter.BackpressureMode.BUFFER)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(/* work with the data */);