.net - 在 Rx 中聚合 ForkJoin 的结果

Question

鉴于这段代码：

var loadAll =
   Observable.ForkJoin(
      service1.FindBooksAsObservable().Select(s => s),
      service2.FindBooksAsObservable().Select(s => s),
      service3.FindBooksAsObservable().Select(s => s)
);

loadAll.Subscribe(
   result =>
   {
      var aggregatedListOfBooks = result.SelectMany(b => b);
   });

如您所见，问题在于每个FindBooksAsObservable()方法都返回一个IObservable<IEnumerable<Book>>，因此Subscribe()中的结果变量是一个 数组。IEnumerable<Book>

有没有其他方法可以聚合ForkJoin()的结果？我希望将Merge()之类的东西与 ForkJoin 一起使用。

Answer 1

假设所有三个服务都返回一个列表Books，您可以使用SelectMany来合并列表：

IObservable<Book> loadAll = 
    Observable.ForkJoin(
        service1.FindBooksAsObservable().Select(s => s),
        service2.FindBooksAsObservable().Select(s => s),
        service3.FindBooksAsObservable().Select(s => s)
    )
    .Select(books => books.SelectMany(list => list).ToList());

loadAll.Subscribe(
    book => { /* will be called once with a single list of all items */ });

ToList()如果您不需要输出为列表，则可以删除该调用。

Answer 2

Observable.ForkJoin不在 Reactive Extensions (Rx) (v2.1.30214.0) 的最新稳定版本上。ForkJoin 截至 2013 年 6 月，仅在 Rx 的实验版本中。

Dave Sexton 提出了一种解决方法： http ://social.msdn.microsoft.com/Forums/en-US/rx/thread/3cfccb74-9ce3-47dc-94fd-cf60270c1ed5