android - 具有多个订阅者的单个 Observable

Question

我有一个 Observable<<List<Foo>> getFoo()从改造服务创建并在调用 .getFoo()方法，我需要与多个订阅者共享它。调用.share()但是，它会导致重新执行网络调用。重播运算符也不起作用。我知道一个潜在的解决方案可能是.cache() ，但我不知道为什么会导致这种行为。

// Create an instance of our GitHub API interface.
Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(API_URL)
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .build();

// Create a call instance for looking up Retrofit contributors.
Observable<List<Contributor>> testObservable = retrofit
        .create(GitHub.class)
        .contributors("square", "retrofit")
        .share();

Subscription subscription1 = testObservable
       .subscribe(new Subscriber<List<Contributor>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onNext(List<Contributor> contributors) {
                System.out.println(contributors);
            }
         });

Subscription subscription2 = testObservable
        .subscribe(new Subscriber<List<Contributor>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onNext(List<Contributor> contributors) {
                System.out.println(contributors + " -> 2");
            }
         });

subscription1.unsubscribe();
subscription2.unsubscribe();

上面的代码可以重现上述行为。可以调试一下，看到收到的Lists属于不同的MemoryAddress。

我也将 ConnectableObservables 视为一种潜在的解决方案，但这需要我随身携带原始的 observable，并调用 .connect()每次我想添加一个新的订阅者。

这种行为与 .share()在改造 1.9 之前工作正常。它停止在 Retrofit 2 - beta 上工作。我还没有使用几小时前发布的 Retrofit 2 发布版本对其进行测试。

编辑:2017 年 1 月 2 日

为了以后的读者，我写了一篇文章here详细解释案件!

Answer 1

在与 RxJava 开发人员 Dávid Karnok 联系后，我想对这里发生的事情提出一个完整的解释。

share() 定义为 publish().refCount()，即。 e.源 Observable 首先通过 publish() 转换为 ConnectableObservable 而不必调用 connect() “手动”该部分由 refCount() 处理。特别是，refCount 将在 ConnectableObservable 本身收到第一个订阅时调用 connect()；然后，只要至少有一个订阅者，它将保持订阅状态；最后，当订阅者数量降至 0 时，它将向上取消订阅。使用 cold Observables，就像 Retrofit 返回的那样，这将停止任何正在运行的计算。

如果在这些周期之一之后另一个订阅者出现，refCount 将再次调用 connect 并因此触发对源 Observable 的新订阅。在这种情况下，它会触发另一个网络请求。

现在，这在 Retrofit 1 (实际上是 this commit 之前的任何版本)中通常不会变得明显，因为这些旧版本的 Retrofit 默认将所有网络请求移动到另一个线程。这通常意味着您的所有 subscribe() 调用将在第一个请求/Observable 仍在运行时发生，因此新的 Subscriber 将简单被添加到 refCount，因此不会触发额外的请求/Observables。

然而，较新版本的 Retrofit 默认情况下不再将工作转移到另一个线程 - 您必须通过调用(例如，subscribeOn(Schedulers.io()) 显式执行此操作)。如果你不这样做，一切都将停留在当前线程上，这意味着第二个 subscribe() 只会在第一个 Observable 调用 onCompleted< 之后发生，因此在所有 Subscribers 都已取消订阅并且所有内容都已关闭之后。现在，正如我们在第一段中看到的，当第二个 subscribe() 被调用时，share() 没有选择，只能引起另一个 Subscription 到源 Observable 并触发另一个网络请求。

所以，要回到从 Retrofit 1 开始的习惯行为，只需添加 subscribeOn(Schedulers.io())。

这应该只执行网络请求 - 大多数时候。但原则上，您仍然可以获得多个请求(并且您总是可以使用 Retrofit 1)，但前提是您的网络请求非常快和/或 subscribe() 调用发生相当大的延迟，这样，当第二个 subscribe() 发生时，第一个请求再次完成。

因此，Dávid 建议使用 cache()(但它有你提到的缺点)或 replay().autoConnect()。根据这些 release notes , autoConnect 只像 refCount 的前半部分一样工作，或者更准确地说，它是

similar in behavior to refCount(), except that it doesn't disconnect when subscribers are lost.

这意味着请求只会在第一个 subscribe() 发生时触发，但随后所有的 Subscriber 都会收到所有发出的项目，无论是否有，在两者之间的任何时间，0 个订阅者。