如何退订 PublishSubject 和 BehaviorSubject？答案

【问题标题】：How can PublishSubject and BehaviorSubject be unsubscribed from?如何退订 PublishSubject 和 BehaviorSubject？
【发布时间】：2014-01-21 12:13:16
【问题描述】：

在subjects 包下，您有PublishSubject 和BehaviorSubject 之类的类，我想可以将它们描述为一些可用的示例Observables。

如何取消订阅这些主题？没有 unsubscribe 方法并且调用 onCompleted 完全结束了 Observable 对吗？

【问题讨论】：

标签： java rx-java subject-observer

【解决方案1】：

Subject 同时是 Observable 和 Observer，它可以像普通的 observable 一样被取消订阅。主题的特别之处在于它是可观察者和观察者之间的桥梁。它可以通过重新发射它观察到的项目，也可以发射新项目。主题是可观察的，就像承诺是期货一样。

以下是主题家族的简要说明：

AsyncSubject：只发出源 Observable 的最后一个值

BehaviorSubject：当观察者订阅源 Observable 时，发出最近发出的项以及源 Observable 的所有后续项。

PublishSubject：在订阅时发出源 Observable 的所有后续项。

ReplaySubject：无论订阅者何时订阅，都会发出源 Observable 的所有项目。

official doc 带有一些漂亮的大理石图，使其更容易理解

【讨论】：

BehaviorSubject 和 ReplaySubject(1) 本质上一样吗？
@DzmitryLazerka：BehaviorSubject 总是有一个初始值。如果您正在寻找没有初始值的 BehaviorSubject，请使用 ReplaySubject(1)。
不错的答案，尽管稍作修正 - ReplaySubject 还可以在 create() 中指定一个缓冲区，这将是订阅时要发出的项目数

【解决方案2】：

主题本质上是 Observables 和 Observers。

Observable 本质上是一个具有接受 Observer 并返回订阅的函数的东西。因此，例如，给定简单的 observable：

    Observable<Integer> observable = Observable.create(new Observable.OnSubscribeFunc<Integer>() {
        @Override
        public Subscription onSubscribe(Observer<? super Integer> observer) {
            observer.onNext(3);
            observer.onNext(2);
            observer.onNext(1);
            observer.onCompleted();

            return Subscriptions.empty();
        }
    });

在这里我们将订阅它，为每个整数打印一行：

    Subscription sub = observable.subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer integer) {
            System.out.println(integer);
        }
    });

您可以致电sub.unsubscribe() 取消订阅。

这是一个 PublishSubject，它做的事情大致相同：

    PublishSubject<Integer> publishSubject = PublishSubject.create();
    Subscription subscription = publishSubject.subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer integer) {
            System.out.println(integer);
        }
    });

    publishSubject.onNext(3);
    publishSubject.onNext(2);
    publishSubject.onNext(1);

您可以通过调用subscription.unsubscribe() 以同样的方式取消订阅。

【讨论】：

您能否描述一下我们如何让订阅者并行处理 onNext？如果有多个订阅者而不仅仅是一个，我们如何让它们并行执行
@vach - 根据上面的代码，您可以通过执行以下操作让订阅者并行处理 onNext：Subscription subscription1 = publishSubject.observeOn(Schedulers.newThread()).subscribe(...); 和 Subscription subscription2 = publishSubject.observeOn(Schedulers.newThread()).subscribe(...);
我一直想知道如何做到这一点，直到现在我才明白，谢谢@praveer09 :)

【解决方案3】：

所有Subjects 扩展Observable，您可以使用多个subscribe(...) 方法中的任何一个来订阅。调用任何subscribe(...) 方法都会返回Subscription。

Subscription subscription = anySubject.subscribe(...);

当您想停止侦听来自主题的事件时，请使用此 subscription 实例的 unsubscribe() 方法。

subscription.unsubscribe();

【讨论】：

PublishSubject 怎么样？
@IgorGanapolsky - anySubject 变量 cab 是任何类型的主题，因此同样适用于 PublishSubject