0

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

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

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

4

1 回答 1

2

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

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));
于 2021-06-11T14:08:46.047 回答