android - Rxjava，如何推迟发射直到有多个订阅者订阅？

Question

我正在制作 Android 应用程序，有一个初学者的问题:有没有办法将 Observables 发射推迟到 2 个订阅者订阅？因为我不想让第二个错过数据。谢谢

Answer 1

是的，通过使用 ConnectableObservable 类型进行确定性多播来支持此用例。

有两种使用方法。

第一个也是最简单的方法是像这样使用 publish 重载:

    Observable<Long> sourceToShare = Observable.interval(10, TimeUnit.MILLISECONDS);

    // multiple subscriptions deterministically subscribed to using publish()
    sourceToShare.publish(l -> {
        Observable<String> a = l.map(t -> "first subscriber " + t);
        Observable<String> b = l.buffer(4).map(t -> "second subscriber " + t);
        return Observable.merge(a, b);
    }).take(20).toBlocking().forEach(System.out::println);

或者如果您需要在连接所有订阅者后手动控制源何时开始和停止，您可以像这样直接使用 ConnectableObservable:

    Observable<Long> sourceToShare = Observable.interval(10, TimeUnit.MILLISECONDS);


    // multiple subscriptions deterministically subscribed to using ConnectableObservable
    ConnectableObservable<Long> published = sourceToShare.publish();

    Observable<String> a = published.map(t -> "first subscriber " + t);
    Observable<String> b = published.buffer(4).map(t -> "second subscriber " + t);
    Observable.merge(a, b).take(20).forEach(s -> System.out.println("published => " + s));

    // now that everything is wired up, connect the source
    published.connect();

    // since we are connecting asynchronously and need to 'connect' after subscribing
    // we have to manually give it time to execute
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

这里是文档的链接:

• http://reactivex.io/RxJava/javadoc/rx/Observable.html#publish(rx.functions.Func1)
• http://reactivex.io/RxJava/javadoc/rx/observables/ConnectableObservable.html
• http://reactivex.io/documentation/operators.html#connectable