我是否必须取消订阅已完成的 observable？

Question

如果 observable 完成，我是否仍然需要取消订阅/处置（在 RxJava2 中）observable 以删除 Observer（防止内存泄漏），或者一旦发生onComplete 或 onError 事件，这是否由 RxJava 内部处理？

Single、Completable、Flowable 等其他类型呢？

Answer 1

是的，你是对的。

流终止后（ onComplete / onError 已被调用），订阅者自动取消订阅。您应该能够使用 Subscription 对象上的isUnsubscribed() 方法来测试这些行为。

Answer 2

虽然您不需要手动取消订阅已终止的流，但如果您不小心，仍然可以使用 RxJava2 造成内存泄漏。

考虑以下代码：

repository.getData()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(data -> myTextView.setText(data.toString()));

订阅中的 lambda 参数是匿名内部类的“语法糖”：

subscribe(new Consumer<Data>() {
    @Override
    public void accept(final Data data) {
        myTextView.setText(data.toString());
    }
});

在 JVM 上，匿名内部类维护对外部类的引用。

假设对于上述幼稚代码，外部类是一个 Activity（对于 Fragment、Service、BroadcastReceiver 或任何生命周期由 Android 操作系统控制的类也是如此）。

Activity 订阅了 Observer，但随后在内存不足的情况下被 Android 操作系统销毁（您可以通过打开 Developer Options/Don't Keep Activity 来模拟这种效果）。如果 Activity 销毁时Schedulers.io() 上的工作仍在运行，那么仍然会通过匿名内部类维护对 Activity 的引用。这意味着内存泄漏会阻止 Activity 被垃圾收集器完成。如果 Activity 有多个 View，或者说是一个 Bitmap 对象，那么内存泄漏可能会非常严重。

这里有很多解决方案，但其中之一是维护一个CompositeDisposable 对象并在Android Activity 的onDestroy() 生命周期方法中清除它：

public class MyActivity extends Activity {

   DataRepository dataRepository;
   CompositeDisposable disposables;

   @Override
   public void onCreate(Bundle savedInstanceState) {
       super.onCreate(savedInstanceState);
       disposables = new CompositeDisposable();
   }

   public void onButtonClick(View v) {
       repository.getData()             
          .subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
          .doOnSubscribe(disposable -> disposables.add(disposable))
          .subscribe(data -> myTextView.setText(data.toString()));
   }

   @Override
   public void onDestroy() {
       disposables.clear();
       super.onDestroy();
   }
}

您可以参考官方Google Android Architecture Blueprints 中有关如何在 Android 应用程序中使用 RxJava 的一个很好的示例。