如何限制flatMap的并发性？

Question

我正在尝试使用RxJS编写一个脚本来处理数百个日志文件,每个文件大约1GB.脚本的骨架看起来像

Rx.Observable.from(arrayOfLogFilePath)
.flatMap(function(logFilePath){
   return Rx.Node.fromReadStream(logFilePath)
   .filter(filterLogLine)
})
.groupBy(someGroupingFunc)
.map(someFurtherProcessing)
.subscribe(...)

代码有效,但请注意所有日志文件的过滤步骤将同时启动.但是,从文件系统IO性能的角度来看,最好一个接一个地处理一个文件(或者至少将并发限制为几个文件而不是同时打开所有数百个文件).在这方面,我如何以"功能反应方式"实施？

我曾想过调度程序,但无法弄清楚它在这里有什么用处.

Answer 1

您可以使用.merge(maxConcurrent)来限制并发性.因为.merge(maxConcurrent)将metaobservable(可观察的可观察对象)展平为可观察对象,所以需要替换.flatMapwith以.map使输出是metaobservable("unflat"),然后调用.merge(maxConcurrent).

Rx.Observable.from(arrayOfLogFilePath)
.map(function(logFilePath){
   return Rx.Node.fromReadStream(logFilePath)
   .filter(filterLogLine)
})
.merge(2) // 2 concurrent 
.groupBy(someGroupingFunc)
.map(someFurtherProcessing)
.subscribe(...)

此代码尚未经过测试(因为我无法访问您拥有的开发环境),但这是如何继续进行的.RxJS没有很多具有并发参数的运算符,但您几乎可以随时执行所需的操作.merge(maxConcurrent).