Observable.publish() 不会在源 Observable 完成后订阅的观察者上调用 onCompleted()

Question

我正在尝试让Observable 与所有订阅者共享它的排放量，这样它就会被 subscribe()d 恰好一次。 我尝试使用Observable.publish()，但如果订阅源Observable 完成后订阅，则发布的Observable 的订阅者似乎不会收到任何终止消息（onCompleted() 和可能的onError()）。这是一段代码来证明这一点：

static <T> Observer<T> printObserver(String name) {
    return new Observer<T>() {
        @Override public void onCompleted() {
            System.out.println(name + ": onCompleted()");
        }
        @Override public void onError(Throwable e) {
            System.out.println(name + ": onError( " + e + " )");
        }
        @Override public void onNext(T value) {
            System.out.println(name + ": onNext( " + value + " )");
        }
    };
}
public void testRxPublishConnect() throws Exception {
    Observable<Integer> sourceObservable = Observable.range(1, 5);
    ConnectableObservable<Integer> sharedObservable = sourceObservable.publish();
    sharedObservable.subscribe(printObserver("Observer #1"));
    sharedObservable.connect();
    sharedObservable.subscribe(printObserver("Observer #2"));
}

这是打印出来的：

观察者 #1：onNext(1) 观察者 #1: onNext( 2 ) 观察者 #1: onNext( 3 ) 观察者 #1: onNext( 4 ) 观察者 #1: onNext( 5 ) 观察者 #1：onCompleted()

请注意，Observer #2 不会收到 onCompleted()。 我不认为这是期望的行为。我错过了什么吗？ 我在 RxJava 版本 1.0.8 和 1.0.14 中尝试过，结果相同。

Answer 1

试试.share()，即.publish().refCount()。

Answer 2

这是设计使然。如果您在这种情况下调用connect()，您的订阅者将从一开始就收到所有事件。如果终止的publish 将立即终止其子订阅者，您可能无法观察到值，因为一旦连接，publish 会在没有订阅者的情况下缓慢地标记其源。

Answer 3

我 99% 确定这是预期的行为。我不确定 RxJava，但在我所知道的发布和订阅模式的大多数实现中，可观察对象的默认行为是将事件发布给订阅者并忘记它们。这意味着通知不是“追溯的”（即订阅者对过去发出的事件一无所知）。

另外，来自Observable Contract (section 'multiple observers') of the RxJava documentation：

If a second observer subscribes to an Observable that is already emitting items to a first observer, it is up to the Observable whether it will thenceforth emit the same items to each observer ... There is no general guarantee that two observers of the same Observable will see the same sequence of items.

Answer 4

Publish 通过构建所有订阅者的列表来工作，然后一旦调用connect()，它就会开始向其订阅者列表中的所有订阅者生成数据。这意味着在调用 connect 之前必须知道所有订阅者。这是您将如何使用publish() 或更优选publish(Func1<Observable<T>, Observable<R>>) 重载的方法。

已知订阅者数量：发布

关闭所有订阅的函数。

observableStream.publish(new Func1<Observable<Integer>, Observable<Integer>>() {
    @Override
    public Observable<Integer> call(Observable<Integer> subject) {
        Observable<Integer> o1 = subject.doOnNext(somework1());
        Observable<Integer> o2 = subject.doOnNext(somework2());
        return Observable.merge(o1, o2);
    }
});

手动调用连接和订阅：

ConnectableObservable<Integer> subject = observableStream.publish();
subject.subscribe(somework1());
subject.subscribe(somework2());
subject.connect();

如果您不知道您将拥有多少订阅者，那么您可以将输入窗口化为可管理的块，然后在您的 Transformers 集合上发布您的输入。

订阅者数量未知：窗口

final Set<Transformer<Integer, String>> transformers = new HashSet<>();
observableStream
    .window(100, TimeUnit.MILLISECONDS, 1000)
    .flatMap(new Func1<Observable<Integer>, Observable<String>>(){
        @Override
        public Observable<String> call(Observable<Integer> window) {
            return window.publish(new Func1<Observable<Integer>, Observable<String>>() {
                @Override
                public Observable<String> call(Observable<Integer> publish) {
                    Observable<Observable<String>> workObservables = Observable.from(transformers)
                        .map(new Func1<Transformer<Integer, String>, Observable<String>>(){
                            @Override
                            public Observable<String> call(Transformer<Integer, String> transformer) {
                                return publish.compose(transformer);
                            }});
                    return Observable.merge(workObservables);
                }});
        }})
    .subscribe();

还有第三种选择。您可以使用observable.cache()，但这会将来自该可观察流的所有输入数据保存在内存中，因此您要小心使用它。在这种情况下，您可能最终还是会通过窗口来控制缓存主题的范围。