Rxjs递归调用-并发性

Question

我有以下示例代码以递归方式从某些REST资源中提取数据。每个端点都提供以下JSON内容

{ data: '123',  urls: ['https//example/bar', 'https//example/bim'] }

我可以构建一个Rx流，像这样递归提取内容：

function getData(url) {
  return fromFetch(url).pipe(    
    mergeMap(resp => resp.json()),
    mergeMap(content => merge(of(content.data), ...content.urls.map(getData)))
  )
}

我的问题是在这里控制HTTP并发的最佳方法是什么。

• 可以通过自定义计划程序来实现吗？
• 或者我是否应该简单地将fromFetch包装到一个类中，该类控制在任何时间点可以进行多少个HTTP调用的并发？

Answer 1

expand可能最多只能获得1个并发，通过subject + mergeMap可以更轻松地控制并发，因此您只有一个工作通道。

const fetchSubject=new Subject();
fetchSubject.pipe(
    mergeMap(url=>fromFetch(url),null,CONCURRENCY)    
    mergeMap(resp => resp.json()),
    tap(url=>content.urls.map(url=>fetchSubject.next(url))
    scan((acc,curr) =>{
      return [curr.data,...acc]
    })
  )