java - DisposableObserver 与(常规)观察者

Question

从我在源代码中看到的，在实现方面的区别在于 DisposableObserver 实现了 Disposable 接口(interface)。

在 dispose 中，它调用了一个辅助函数来自行处置。

@Override
public final void dispose() {
    DisposableHelper.dispose(s);
}

但实际上，两者之间有什么区别。常规的 Observable 垃圾不会在不再需要时立即收集吗？

在 Observer 上使用 DisposableObserver 的默认用例是什么？

Answer 1

主要是为了避免内存泄漏，因为它可以让你随时取消订阅。拥有订阅对象的对象可能会在其生命周期结束时订阅正在执行一些需要更多时间的繁重工作。在这种情况下，订阅将保留在内存中，从而泄漏拥有它的对象。为避免这种情况，您可以存储订阅的返回值(一次性)并稍后在所有者对象的生命周期结束时调用 dispose。通常人们最终会将每个一次性用品添加到 CompositeDisposable(基本上是一袋一次性用品)，然后使用 CompositeDisposable.clear() 将它们全部清除，这样您就不必' 必须跟踪每个单独的一次性用品。

假设您有一个名为 ItemPublisher 的类，它提供无限的事件流:

class ItemPublisher {
  // ...

  public Flowable<Item> getItemsStream() {
    // ...
  }
}

现在，假设您有一个订阅该流的名为 ItemHandler 的类:

class ItemHandler {
  // ...

  public void observeItems() {

    itemPublisher.getItemsStream()
      .subscribe(
        // handle onNext events,
        // handle onError events
        // infinite stream, so we don't need to handle onComplete
      );
  }
}

那么，您的软件正在运行，一切都很好。在某个时刻，您的 ItemHandler 实例到达其生命周期的终点。该实例应该被销毁。然而，由于我们在这里处理观察者模式，ItemPublisher 保留了对 ItemHandler 实例的隐式引用，当您调用 subscribe 时传递给它 方法。由于流是无限的，该引用永远不会被删除，从而阻止 GC 清理 ItemHandler 实例，从而导致内存泄漏。这不仅仅发生在无限流中:如果 ItemPublisher 的生命周期比 ItemHandler 长，同样的情况也会发生。

这就是为什么我们有 Disposable 接口(interface)。当您调用 subscribe 时，您可以使用返回 Disposable 的方法的重载版本。当您不再需要订阅时，您可以对其调用 dispose()。所以，在我们的例子中:

class ItemHandler {
  private Disposable subscriber;  // this gets initialized somewhere

  // ...

  public void observeItems() {

    itemPublisher.getItemsStream()
      .subscribe(this.getSubscriber());
  }


  // later, when you know that this ItemHandler instance is not needed anymore
  public void wrapItUp() {
    // ...
    subscriber.dispose();
  }
}

正如我之前提到的，您还有 CompositeDisposable，当您有很多订阅时，它会派上用场。通过使用它，您可以将所有订阅集中到一个地方，然后一次性全部删除。例如:

class ItemHandler {
  private CompositeDisposable disposables;

  // ...

  public void observeItems() {

    disposables.add(itemPublisher.getItemsStream()
      .subscribe(
        // ...
      )
    );
  }


  public void observeSomethingElse() {
     disposables.add(somethingElse.getStreamOfSomethingElse()
      .subscribe(
        // ...
      )
    );
  }

  // later, when you know that this ItemHandler instance is not needed anymore
  public void wrapItUp() {
    // ...
    disposables.clear();
  }
}