如何在RxJava中正确转换多播可观察对象

Question

假设我有一个要转换为反应流的事件发出数据源。数据源由资源（例如，定期发送更新状态的套接字）绑定，所以我希望共享对该资源的单个订阅。使用带有回放的单个observable（新订阅者可以立即获得当前值）和refCount操作符似乎非常适合这样做。例如，这是他的MyDataProvider的外观：

private final Observable<MyData> myDataObservable = Observable.<MyData>create(emitter -> {
    // Open my resource here and emit data into observable
})
    .doOnDispose(() -> {
        // Close my resource here
    })
    .replay(1)
    .refCount();

public Observable<MyData> getMyDataObservable() {
    return myDataObservable;
}

然而，现在假设我有另一个数据源，它需要第一个数据源的结果来计算其自身的值：

private final Observable<AnotherData> anotherDataObservable = getMyDataProvider().getMyDataObservable()
    .flatMap(myData -> {
        // Call another data source and return the result here
    })

public Observable<AnotherData> getAnotherDataObservable() {
    return anotherDataObservable;
}

在这里，我的设置开始崩溃。第一个可观察到的多播仅在refCount操作员之前有效。之后，一切都再次单播。这意味着，如果对另一个数据提供者进行了两次单独的订阅，则会调用两次flatMap操作符。我认为有两种解决方法，但我不喜欢这两种方法：

对于我来说，最简单的解决方法似乎是在进行多播操作之前，将myDataObservable的单播变体保存在某个地方，然后在另一个DataObservable中执行该多播操作。然而，如果这两个Observable位于不同的模块中，这种解决方法会使代码非常不美观，要求MyDataProvider公开两个看起来返回相同数据的不同观测值。

第二个解决方法似乎是在antherDataWatable中再次应用那些replay和refCount运算符。但这会导致效率低下，因为myDataWatable中的第一个多播运算符已经应用，但现在什么也不做，除了浪费内存和CPU周期。

这两种解决方法还涉及将另一个数据提供者耦合到MyDataProvider。如果将来MyDataProvider发生变化，不再需要多播，我还必须更新另一个数据提供程序，从中删除多播html" target="_blank">操作符。

解决这个问题的更优雅的方法是什么？我能更好地构建它以完全避免这个问题吗？

Answer 1

您可以分割单播流和多播流，但这是多余的。我认为第二种方法更好，顺便说一句，replay和refcount操作符实际上是在做事情，不是浪费。

调用启用多播时，您正在将myDataObservable的可观测转换为可观测
然后使用refcount（）内部订阅，这也为后续订阅提供了一个单点；在这一点之后，一切都再次单播。

在另一个可观察的数据中，您真正想要实现的是相同的，因此，请执行与在myDataObservable中完全相同的操作。

Answer 2

您可以使用“publish（）.refCount（）”串联来允许共享单个订阅者。由于它们经常使用，因此它们有一个别名share（）。

您也可以使用ConnectableWatable。但是在使用ConnectableWatables重播时要小心。

如果您在将可观察对象转换为可连接可观察对象之前将重播运算符应用于可观察对象，则生成的可连接可观察对象将始终向任何未来的观察者发出相同的完整序列，即使是那些在可连接可观察对象开始向其他订阅观察者发出项目后订阅的观察者。如文档所述：

Answer 3

关于您的第一种方法，在当前设置中，您的另一个DataObservable使用myDataObservable，据我所知，它们是逻辑耦合的，因为它们使用相同的源。因此，您需要为它们提供一些基本的共享逻辑。我将其提取到一个公共模块中，该模块将公开observable的单播版本，然后使myDataObservable和另一个DataObservable在不同的模块中使用，每个模块都添加了多播逻辑。

另一种选择是，让一个类来监视您的资源，方法是像在myDataObservable中那样订阅它，在onNext中进行处理，并发布带有主题的映射结果，即如果您希望始终能够访问上次发布的值，则使用BehavioralSubject，并使用另一个主题发布原始结果。客户端将订阅该主题，并将获得仅在监控类中计算一次的映射值或原始值。

P、 在订阅之前，记住在你的主题中添加背压策略。

如果这些选项不适合您，请思考避免多次调用flatMap是否真的很重要？您的代码非常简单，这是一个重要的指标。如果flatMap不重，您可以让它运行多次。