java - 如何使用 Java RX Observable 链接异步操作？

Question

我想重复发出 HTTP 请求并根据结果采取行动。我从 public Observable<NewsItem> fetchItems(NewsFeed feed) 开始.一个请求获得了一些新闻条目，但我决定将其扁平化。

想法是使用 Observable.interval()多次发出请求，然后将结果 Observables 合并为一个。

       Observable
            .interval(timePerItem, TimeUnit.MILLISECONDS)
            .map(i -> feed)
            .map(feed -> fetchItems(feed))
            .subscribe(result -> System.out.println(result));

但结果是Observable<Observable<NewsItem>>不是Observable<NewsItem> .如何管理它们？

我找到了 marge() 运算符 (RX-Java doc: Marge) .但它似乎不适合用例。

在之前的版本中我使用了CompletableFuture<List<NewsItem>> fetchNewsItems()但我无法将它放入 Observable 链中。

Answer 1

不确定我是否理解这个问题，但你不就是在寻找 flatMap 吗？

Observable
    .interval(timePerItem, TimeUnit.MILLISECONDS)
    .flatMap(i -> fetchItems(feed))
    .subscribe(result -> System.out.println(result));