订阅者是在所有事件之后还是在每个事件之后得到通知？

Question

我是 RxJava 的新手，我的印象是每个事件都会通知每个订阅者。因此，如果我们有 N 个订阅者和一个 X 事件流，则将调用 N 个订阅者中每个订阅者的 onNext。但是当我运行以下代码时：

public static void main(String[] args) {
        Observable<String> source = Observable.create(emitter -> {
            emitter.onNext("Hello");
            emitter.onNext("Foo");
            emitter.onNext("Bar");
            emitter.onNext("RxJava");
        });

        source.subscribe(e -> System.out.println("Observer 1: " + e));
        source.subscribe(e -> System.out.println("Observer 2: " + e));
    }

我明白了：

Observer 1: Hello
Observer 1: Foo
Observer 1: Bar
Observer 1: RxJava
Observer 2: Hello
Observer 2: Foo
Observer 2: Bar
Observer 2: RxJava

所以基本上在完成所有 onNext 之后，才会触发下一个观察者。

我期待看到：

Observer 1: Hello 
Observer 2: Hello
Observer 1: Foo
Observer 2: Foo
Observer 1: Bar
Observer 2: Bar
Observer 1: RxJava
Observer 2: RxJava

在我看来，对于很长的流来说效率很低，我做错了什么吗？

Answer 1

RxJava 序列默认是同步的，因此上面的订阅调用将运行你的发射代码就在那里。要实现交错，您需要一种方法来告知源何时两个消费者都准备好接收。这可以通过多种方式完成：

ConnectableObservable<String> source = Observable.<String>create(emitter -> {
            emitter.onNext("Hello");
            emitter.onNext("Foo");
            emitter.onNext("Bar");
            emitter.onNext("RxJava");
        }).publish();

        source.subscribe(e -> System.out.println("Observer 1: " + e));
        source.subscribe(e -> System.out.println("Observer 2: " + e));

        source.connect();

或

ConnectableObservable<String> source = Observable.<String>create(emitter -> {
            emitter.onNext("Hello");
            emitter.onNext("Foo");
            emitter.onNext("Bar");
            emitter.onNext("RxJava");
        }).publish().refCount(2);

        source.subscribe(e -> System.out.println("Observer 1: " + e));
        source.subscribe(e -> System.out.println("Observer 2: " + e));

订阅者是在所有事件之后还是在每个事件之后得到通知？

Are subscribers notified after all events or per event?

java

observable

rx-java

reactive-programming

rx-java3