RxJava。将列表的热可观察对象转换为单项流，对其进行处理并转换回列表答案

【问题标题】：RxJava. Convert hot observable of list to single item stream, process it and convert to list backRxJava。将列表的热可观察对象转换为单项流，对其进行处理并转换回列表
【发布时间】：2017-05-28 23:55:14
【问题描述】：

假设我有一个 hot 可观察到的一些项目列表；

Observable<List<Item>> observable = ...;

我需要将其转换为单个项目流并对每个项目执行一些操作，例如过滤，然后我应该将其转换回列表并在订阅者的onNext 方法中处理它：

observable.flatMap(Observable::from)
    .filter(Item::isFiltered)
    .toList()
    .subscribe(this::onNext, this::onError)

public void onNext(List<Item> items) {...}

乍一看似乎没问题，但事实并非如此，因为我们的 observable 是 hot，所以 toList() 永远不会被执行（因为它等待 source observable 完成）和整个流卡住了。

我该如何解决这个问题？另请注意，filter 附近可能是对单个项目的任何数量的附加操作。

【问题讨论】：

标签： java rx-java observable

【解决方案1】：

您可以对单个项目和最终的 toList 运算符执行您在 flatMap 中创建的 Observable 的所有操作。这样，您会收到 onComplete 调用，toList 将收集和转换项目。

observable.flatMap(list -> {
     return Observable.from(list)
                        .filter(Item::isFiltered)
                        .toList()
  })
  .subscribe(this::onNext, this::onError)

【讨论】：