rxJava 在取消订阅时停止观察答案

【问题标题】：rxJava stop observable on unsubscriberxJava 在取消订阅时停止观察
【发布时间】：2016-07-21 21:31:18
【问题描述】：

我已经冷了一个 observable，它在订阅计算调度程序时会产生很多项目并发送给另一个对它们做更重工作的 o。

问题是我想在取消订阅时停止所有可观察链。

有任何运营商吗？我试图检查我是否有订阅者，但此时我的大部分项目都已发出。

【问题讨论】：

调用subscriber.unsubscribe() 将取消订阅整个链。你有单链吗？你能显示一些代码吗？
@DaveMoten，是的，我的第一个可观察到的停止发射项目，但我的（慢）flatmap 函数继续处理在调用 en unsubscribe 之前发射的项目。
@GuilhermeTorresCastro 您的线程将在取消订阅时被中断 - 在您工作时检查中断状态并在设置时提前退出。
@TassosBassoukos 我测试过，中断状态没有改变。

标签： rx-java

【解决方案1】：

创建一个可以完成繁重工作并可以检查订阅的 observable：

obs //emits stuff fast
  .flatMap(x -> 
    Observable.create(
        sub -> {
          //frequently check subscription status
          while (!sub.isUnsubscribed()) {
            // do some work
            if (finished) break;
          }
          if (!sub.isUnsubscribed()) {
            sub.onNext(result);
            sub.onCompleted();
          }
        }))
      //ensure backpressure compliant
      .onBackpressureBuffer() 
    ).subscribe(subscriber);

【讨论】：

我因为没有考虑到这一点而感到愚蠢。谢谢！